AI Agents Interview Questions

AI Agents Interview Questions & Answers (1–40)

Section 1: AI Agents Fundamentals

Q1: What is an AI Agent and how does it differ from a regular LLM call?

Answer: An AI Agent is an autonomous system that uses an LLM as its reasoning engine to perceive context, plan actions, use tools, and work towards a goal — potentially over many steps with a feedback loop.

Aspect	Single LLM Call	AI Agent
Steps	One prompt → one response	Multi-step reasoning loop
Tools	None	Web search, code execution, APIs, DB
Memory	None (stateless)	Short-term (context) + Long-term (vector DB)
Autonomy	Zero	Decides what to do next
Goal type	Answer a question	Complete a complex task
Error recovery	None	Can retry, try different approach
Cost	Single API call	Multiple API calls (can get expensive!)

ReAct loop (Reason + Act) — the core agent loop:

User Goal: "Research and summarize the latest K8s security CVEs"
     ↓
[THINK] What do I need? → I'll search for recent CVEs
     ↓
[ACT]   search_web("kubernetes CVE 2024 critical")
     ↓
[OBSERVE] Got 10 results: CVE-2024-xxxx, CVE-2024-yyyy...
     ↓
[THINK] I have results, now I'll summarize them
     ↓
[ACT]   search_web("CVE-2024-xxxx kubernetes details")
     ↓
[OBSERVE] Got detailed description...
     ↓
[THINK] I have enough info. Generate final answer.
     ↓
[FINAL ANSWER] "The top 3 K8s CVEs in 2024 are..."

Simple agent loop in Python:

def agent_loop(goal: str, tools: dict, max_steps: int = 10):
    memory = []   # Tracks thoughts + observations

    for step in range(max_steps):
        # LLM decides what to do given goal + history
        decision = llm.decide(goal=goal, history=memory, available_tools=tools)

        if decision.type == "final_answer":
            return decision.content   # Done!

        # Execute the chosen tool
        tool_fn = tools[decision.tool_name]
        result = tool_fn(**decision.tool_args)

        # Remember what happened
        memory.append({
            "step": step,
            "thought": decision.thought,
            "tool": decision.tool_name,
            "args": decision.tool_args,
            "observation": result
        })

    return "Max steps reached — task incomplete"

Interview tip: “An AI Agent is NOT just an LLM with tools. The key difference is the autonomous loop — the agent decides WHEN to use tools, WHICH tool, and WHAT to do with the result, iterating until the goal is achieved. This creates emergent problem-solving behavior.”

Q2: What are the core components of an AI Agent?

Answer:

                    AI Agent Architecture
┌─────────────────────────────────────────────────────┐
│                      AGENT                          │
│  ┌─────────────────────────────────────────────┐   │
│  │              LLM Brain (GPT-4, Claude)       │   │
│  │  Reasoning • Planning • Decision Making      │   │
│  └───────────────────┬─────────────────────────┘   │
│           ┌──────────┼──────────┐                   │
│           ▼          ▼          ▼                   │
│  ┌──────────────┐  ┌──────┐  ┌────────────────┐   │
│  │   Memory     │  │Tools │  │  Orchestrator  │   │
│  │ - Short-term │  │ Web  │  │  (Agent loop)  │   │
│  │   (context)  │  │ Code │  │  Tool routing  │   │
│  │ - Long-term  │  │ APIs │  │  Output parse  │   │
│  │   (vector DB)│  │ DB   │  └────────────────┘   │
│  └──────────────┘  └──────┘                         │
└─────────────────────────────────────────────────────┘

Component	Purpose	Example
LLM Brain	Reasoning + planning	GPT-4o, Claude 3.5, Gemini 1.5
Short-term Memory	Current conversation context	Last 10 messages in prompt
Long-term Memory	Persistent knowledge across sessions	Chroma / Pinecone vector DB
Tools	What the agent can DO	Web search, code exec, email, SQL
Orchestrator	Manages the think-act-observe loop	LangChain, LangGraph, AutoGen
Output Parser	Extracts structured actions from LLM text	JSON parsing, Pydantic models

# Full conceptual agent loop with all components
class Agent:
    def __init__(self, llm, tools, vector_memory):
        self.llm = llm
        self.tools = {t.name: t for t in tools}
        self.vector_memory = vector_memory   # long-term
        self.short_term = []                  # in-session context

    def run(self, goal: str, max_steps: int = 10) -> str:
        # Retrieve relevant long-term memories
        past_context = self.vector_memory.search(goal, k=3)

        for step in range(max_steps):
            # Build full context: goal + past + current session
            context = self._build_context(goal, past_context)

            # LLM reasons about what to do
            decision = self.llm.think(context, available_tools=self.tools)

            if decision.is_final:
                # Save to long-term memory before returning
                self.vector_memory.save(goal=goal, result=decision.answer)
                return decision.answer

            # Execute tool
            tool_result = self.tools[decision.tool_name].run(decision.args)

            # Update short-term memory
            self.short_term.append({
                "thought": decision.thought,
                "tool": decision.tool_name,
                "result": tool_result
            })

        return "Task incomplete after max steps"

Interview tip: Memory is the most overlooked component. Short-term memory fills context window → gets expensive. Long-term memory via vector DB → semantic search across sessions. Production agents need BOTH.

Q3: What is RAG (Retrieval-Augmented Generation)?

Answer: RAG lets an LLM answer questions using your private/up-to-date documents that weren’t in its training data, by retrieving relevant chunks and injecting them into the prompt.

Why RAG?

Without RAG:                    With RAG:
LLM only knows training data    LLM knows your internal docs
Knowledge cutoff = stale        Always up-to-date
Can't access private data       Can search your codebase, PDFs
Hallucinates when unsure        Grounded in real retrieved text

RAG Pipeline:

[INDEXING - done once]
Documents → Chunk → Embed → Store in Vector DB

[QUERYING - done per request]
Question → Embed → Search Vector DB → Top-K chunks
                                          ↓
                        [LLM Prompt] = "Answer using this context:
                                        <chunk1> <chunk2> <chunk3>
                                        Question: {user_question}"
                                          ↓
                                      Answer

from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import Chroma
from langchain.chains import RetrievalQA
from langchain_community.document_loaders import DirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter

# STEP 1: Load documents
loader = DirectoryLoader('./docs', glob="**/*.md")
documents = loader.load()

# STEP 2: Chunk (smaller chunks = more precise retrieval)
splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,     # Characters per chunk
    chunk_overlap=200    # Overlap prevents losing context at boundaries
)
chunks = splitter.split_documents(documents)

# STEP 3: Embed + Store
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"  # Persistent storage
)

# STEP 4: Create retriever
retriever = vectorstore.as_retriever(
    search_type="mmr",     # Maximal Marginal Relevance (diverse results, less redundancy)
    search_kwargs={"k": 5}  # Return top 5 chunks
)

# STEP 5: Build RAG chain
qa_chain = RetrievalQA.from_chain_type(
    llm=ChatOpenAI(model="gpt-4o", temperature=0),  # temperature=0 for factual answers
    chain_type="stuff",          # Stuff all chunks into one prompt
    retriever=retriever,
    return_source_documents=True  # Know WHERE the answer came from
)

# STEP 6: Query
result = qa_chain.invoke({"query": "How do I configure Kubernetes RBAC?"})
print(result["result"])           # The answer
print(result["source_documents"]) # Which chunks were used

RAG chunk strategies:

Strategy	Chunk size	Best for
Small chunks	200-500 chars	Precise Q&A, facts
Medium chunks	500-1500 chars	Most use cases
Large chunks	1500-3000 chars	Long-form content, context-heavy answers
Sliding window	Any + overlap	Never lose boundary context

Common mistake: Using chunk_overlap=0. Without overlap, a sentence split across two chunks may never be retrieved. Always use 10-20% overlap relative to chunk size.

Q4: What are tool/function calling in LLMs?

Answer: Tool calling allows an LLM to say “I need to call function X with these arguments” instead of making up an answer. The application runs the function and gives the result back to the LLM.

How it works:

User: "What's the weather in Tokyo?"
  ↓
LLM sees tools available → decides to call get_weather
  ↓
LLM outputs: {"tool": "get_weather", "args": {"city": "Tokyo"}}
  ↓
Your code runs get_weather("Tokyo") → returns real data
  ↓
LLM receives result → generates final natural language answer

from openai import OpenAI
import json

client = OpenAI()

# STEP 1: Define tools (JSON Schema format)
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Get current weather for a city. Use when user asks about weather.",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "City name e.g. Tokyo, London"
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "Temperature unit"
                    }
                },
                "required": ["city"]  # city is mandatory, unit is optional
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "search_web",
            "description": "Search the web for current info. Use for recent events.",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "Search query"}
                },
                "required": ["query"]
            }
        }
    }
]

messages = [{"role": "user", "content": "What's the weather in Tokyo right now?"}]

# STEP 2: First LLM call — LLM decides which tool to use
response = client.chat.completions.create(
    model="gpt-4o",
    tools=tools,
    tool_choice="auto",   # "auto" = LLM decides, "required" = must use a tool
    messages=messages
)

tool_call = response.choices[0].message.tool_calls[0]
print(tool_call.function.name)       # get_weather
print(tool_call.function.arguments)  # {"city": "Tokyo", "unit": "celsius"}

# STEP 3: Run the actual function
args = json.loads(tool_call.function.arguments)
weather_data = get_weather(city=args["city"], unit=args.get("unit", "celsius"))

# STEP 4: Give result back to LLM
messages.append(response.choices[0].message)   # LLM's tool call decision
messages.append({
    "role": "tool",
    "tool_call_id": tool_call.id,     # Must match the tool call ID
    "content": json.dumps(weather_data)
})

# STEP 5: Second LLM call — LLM generates human-friendly answer
final_response = client.chat.completions.create(
    model="gpt-4o",
    messages=messages
)
print(final_response.choices[0].message.content)
# "The current weather in Tokyo is 22°C and sunny."

Parallel tool calls (OpenAI supports multiple tools in one call):

# LLM can call multiple tools simultaneously
# e.g., "What's the weather in Tokyo AND London?"
# → get_weather(Tokyo) + get_weather(London) in parallel
for tool_call in response.choices[0].message.tool_calls:  # Note: plural!
    args = json.loads(tool_call.function.arguments)
    result = execute_tool(tool_call.function.name, args)
    messages.append({"role": "tool", "tool_call_id": tool_call.id, "content": str(result)})

Interview tip: The description field in tool definitions is CRITICAL. The LLM reads it to decide whether to call this tool. Write descriptions like documentation: what it does, when to use it, what format arguments should be in.

Section 2: LangChain & Agent Frameworks

Q5: What is LangChain and how do you build an agent with it?

Answer: LangChain is a framework for composing LLM applications with reusable components: chains, agents, memory, and tools. Think of it as “building blocks” for LLM apps.

LangChain core concepts:

Chain       = sequence of operations (prompt → LLM → output parser)
Agent       = chain with a loop that decides what tool to call next
Tool        = callable function the agent can use
Memory      = state persisted across chain calls
Callback    = hooks for logging/tracing each step
Runnable    = any composable LangChain object (LCEL)

from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_community.tools import DuckDuckGoSearchRun
from langchain.tools import tool
import requests

# STEP 1: Define custom tools using @tool decorator
@tool
def get_k8s_pod_status(namespace: str, pod_name: str) -> str:
    """Get the status of a Kubernetes pod. Use when asked about pod health."""
    # In production, use kubernetes python client
    result = subprocess.run(
        ["kubectl", "get", "pod", pod_name, "-n", namespace, "-o", "json"],
        capture_output=True, text=True
    )
    return result.stdout

@tool
def execute_python(code: str) -> str:
    """Execute Python code and return output. Use for calculations and data processing."""
    try:
        import io, sys
        old_stdout = sys.stdout
        sys.stdout = buffer = io.StringIO()
        exec(code, {})
        sys.stdout = old_stdout
        return buffer.getvalue() or "Code executed successfully (no output)"
    except Exception as e:
        return f"Error: {type(e).__name__}: {e}"

# STEP 2: Build tools list
tools = [
    DuckDuckGoSearchRun(),    # Web search
    get_k8s_pod_status,       # Custom: K8s integration
    execute_python            # Custom: code execution
]

# STEP 3: Create LLM
llm = ChatOpenAI(
    model="gpt-4o",
    temperature=0,    # Deterministic for agents (avoid hallucination)
    streaming=True    # Enable streaming responses
)

# STEP 4: Create prompt with scratchpad placeholder
prompt = ChatPromptTemplate.from_messages([
    ("system", """You are a DevOps assistant. Use tools to answer questions accurately.
    Always verify information with tools before responding.
    If a tool fails, try a different approach."""),
    MessagesPlaceholder(variable_name="chat_history", optional=True),  # Memory
    ("human", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad")  # Tool call history
])

# STEP 5: Create agent + executor
agent = create_tool_calling_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,            # Print reasoning steps
    max_iterations=10,       # Prevent infinite loops
    handle_parsing_errors=True,  # Don't crash on bad LLM output
    return_intermediate_steps=True  # Get full tool call history
)

# STEP 6: Run with memory
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory

message_history = ChatMessageHistory()
agent_with_memory = RunnableWithMessageHistory(
    agent_executor,
    lambda session_id: message_history,
    input_messages_key="input",
    history_messages_key="chat_history"
)

result = agent_with_memory.invoke(
    {"input": "Check the nginx pod in the default namespace and tell me if it's healthy"},
    config={"configurable": {"session_id": "user123"}}
)
print(result["output"])

Interview tip: The agent_scratchpad placeholder is REQUIRED — it’s where LangChain injects the tool call history so the agent can see its own reasoning trace. Without it, the agent can’t remember what tools it already called in the current run.

Q6: What is LangGraph and when should you use it over LangChain agents?

Answer: LangGraph builds stateful, multi-actor workflows as explicit graphs (nodes + edges). It gives you full control over the agent’s flow, unlike the linear loop of standard agents.

LangChain Agent vs LangGraph:

LangChain Agent:              LangGraph:
Fixed loop:                   Custom graph:
think → act → observe →      research → [conditional] → code
  think → act → ...                                   → answer
                              Full control of flow
Good for: simple tasks        Good for: complex, branching workflows

When to use LangGraph:

Complex multi-step workflows with branching
Human-in-the-loop (pause + wait for approval)
Multiple specialized agents collaborating
Workflows that need to restart/retry specific nodes
You need to checkpoint and resume long workflows

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, List
import operator

# STEP 1: Define shared state (passed between all nodes)
class AgentState(TypedDict):
    messages: Annotated[List, operator.add]  # operator.add = append, not overwrite
    search_results: List[str]
    code_output: str
    needs_code: bool
    final_answer: str

# STEP 2: Define node functions (each receives state, returns partial state)
def research_node(state: AgentState) -> dict:
    """Search for information based on the question"""
    query = state["messages"][-1].content
    results = web_search(query)

    # Determine if coding is needed
    needs_code = any(word in query.lower() for word in ["calculate", "compute", "analyze"])

    return {
        "search_results": results,
        "needs_code": needs_code
    }

def code_node(state: AgentState) -> dict:
    """Generate and execute code based on research"""
    code = llm.generate_code(state["search_results"], state["messages"][-1].content)
    output = sandbox_execute(code)
    return {"code_output": output}

def answer_node(state: AgentState) -> dict:
    """Synthesize final answer from all gathered info"""
    context = {
        "research": state["search_results"],
        "code_output": state.get("code_output", ""),
        "question": state["messages"][-1].content
    }
    answer = llm.synthesize(context)
    return {"final_answer": answer}

# STEP 3: Conditional routing function
def route_after_research(state: AgentState) -> str:
    """Decide: do we need to run code, or go straight to answer?"""
    return "code" if state["needs_code"] else "answer"

# STEP 4: Build the graph
workflow = StateGraph(AgentState)

# Add nodes
workflow.add_node("research", research_node)
workflow.add_node("code", code_node)
workflow.add_node("answer", answer_node)

# Define flow
workflow.set_entry_point("research")
workflow.add_conditional_edges(
    "research",
    route_after_research,
    {
        "code": "code",    # goes to code_node
        "answer": "answer" # skips code, goes to answer_node
    }
)
workflow.add_edge("code", "answer")  # After code, always go to answer
workflow.add_edge("answer", END)     # After answer, done

# STEP 5: Compile and run
app = workflow.compile(
    checkpointer=MemorySaver()  # Enable checkpointing for long-running workflows
)

from langchain_core.messages import HumanMessage
result = app.invoke(
    {"messages": [HumanMessage(content="Calculate compound interest for $10000 at 7% for 10 years")]},
    config={"configurable": {"thread_id": "session-001"}}  # For checkpointing
)
print(result["final_answer"])

Interview tip: LangGraph is the production standard for complex agents. It solves the biggest pain point of standard agents — you can’t control WHERE in the loop you are. With LangGraph you define explicit nodes, so you can add human approval gates, retry specific steps, and observe intermediate state.

Q7: What is AutoGen and how does it enable multi-agent systems?

Answer: AutoGen (Microsoft) enables multiple AI agents to collaborate by chatting with each other.

import autogen

# Configuration
config_list = [{"model": "gpt-4o", "api_key": "YOUR_API_KEY"}]
llm_config = {"config_list": config_list, "cache_seed": 42}

# Define agents
user_proxy = autogen.UserProxyAgent(
    name="User_Proxy",
    human_input_mode="NEVER",
    max_consecutive_auto_reply=10,
    is_termination_msg=lambda x: x.get("content", "").rstrip().endswith("TERMINATE"),
    code_execution_config={
        "work_dir": "coding",
        "use_docker": False
    }
)

planner = autogen.AssistantAgent(
    name="Planner",
    system_message="""You are a planning expert. Break down tasks into steps.
    Coordinate with Coder and Reviewer agents.
    When done, output TERMINATE.""",
    llm_config=llm_config
)

coder = autogen.AssistantAgent(
    name="Coder",
    system_message="You write clean, tested Python code based on the Planner's steps.",
    llm_config=llm_config
)

reviewer = autogen.AssistantAgent(
    name="Code_Reviewer",
    system_message="You review code for bugs, security issues, and best practices.",
    llm_config=llm_config
)

# Group chat
groupchat = autogen.GroupChat(
    agents=[user_proxy, planner, coder, reviewer],
    messages=[],
    max_round=20
)
manager = autogen.GroupChatManager(groupchat=groupchat, llm_config=llm_config)

# Start conversation
user_proxy.initiate_chat(
    manager,
    message="Build a Python script that analyzes CSV files and generates summary statistics."
)

Section 3: Agent Memory & State

Q8: How do you implement memory in AI Agents?

Answer:

# 1. Short-term memory (conversation buffer)
from langchain.memory import ConversationBufferWindowMemory, ConversationSummaryMemory

# Keep last N turns
buffer_memory = ConversationBufferWindowMemory(k=10)

# Summarize old messages (saves tokens)
summary_memory = ConversationSummaryMemory(
    llm=ChatOpenAI(),
    max_token_limit=2000
)

# 2. Long-term memory (vector store)
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain.memory import VectorStoreRetrieverMemory

vectorstore = Chroma(embedding_function=OpenAIEmbeddings())
retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

vector_memory = VectorStoreRetrieverMemory(retriever=retriever)

# Save interaction
vector_memory.save_context(
    {"input": "My name is Alice"},
    {"output": "Nice to meet you, Alice!"}
)

# Later retrieval
relevant = vector_memory.load_memory_variables({"prompt": "What's my name?"})

# 3. Entity memory (tracks specific entities)
from langchain.memory import ConversationEntityMemory
entity_memory = ConversationEntityMemory(llm=ChatOpenAI())

Q9: What are vector embeddings and how do they work?

Answer: Embeddings convert text to dense numerical vectors, where semantically similar text has similar vectors.

from openai import OpenAI
import numpy as np

client = OpenAI()

def get_embedding(text: str) -> list:
    response = client.embeddings.create(
        model="text-embedding-3-small",
        input=text
    )
    return response.data[0].embedding

def cosine_similarity(a: list, b: list) -> float:
    a, b = np.array(a), np.array(b)
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

# Example: semantic search
documents = [
    "Docker containers are lightweight virtualization",
    "Kubernetes orchestrates containerized applications",
    "Python is a programming language",
    "Machine learning models need training data"
]

embeddings = [get_embedding(doc) for doc in documents]
query = "container orchestration"
query_embedding = get_embedding(query)

similarities = [cosine_similarity(query_embedding, emb) for emb in embeddings]
ranked = sorted(zip(similarities, documents), reverse=True)

for score, doc in ranked:
    print(f"{score:.3f}: {doc}")
# 0.891: Kubernetes orchestrates containerized applications
# 0.742: Docker containers are lightweight virtualization

Q10: How do you implement a vector database for an agent?

Answer:

# Using Chroma (local)
from chromadb import Client
from chromadb.config import Settings
import chromadb

client = chromadb.PersistentClient(path="./chroma_storage")

collection = client.create_collection(
    name="knowledge_base",
    metadata={"hnsw:space": "cosine"}  # Use cosine similarity
)

# Add documents
collection.add(
    documents=["Kubernetes RBAC controls access", "Docker uses namespaces", "Helm manages K8s charts"],
    metadatas=[{"topic": "k8s"}, {"topic": "docker"}, {"topic": "k8s"}],
    ids=["doc1", "doc2", "doc3"]
)

# Query
results = collection.query(
    query_texts=["how to control kubernetes access"],
    n_results=2,
    where={"topic": "k8s"}   # Filter by metadata
)
print(results["documents"])

# Using Pinecone (cloud)
from pinecone import Pinecone, ServerlessSpec

pc = Pinecone(api_key="YOUR_PINECONE_API_KEY")
index = pc.Index("knowledge-base")

# Upsert vectors
index.upsert(vectors=[
    {"id": "doc1", "values": embedding, "metadata": {"text": "content", "source": "file.md"}}
])

# Query
results = index.query(
    vector=query_embedding,
    top_k=5,
    include_metadata=True,
    filter={"source": {"$eq": "file.md"}}
)

Section 4: Prompt Engineering

Q11: What are key prompt engineering techniques?

Answer:

1. Zero-shot vs Few-shot:

# Zero-shot
prompt = "Classify sentiment: 'I love this product!'"

# Few-shot (examples guide the model)
prompt = """Classify sentiment as positive/negative/neutral:
Text: "I love this!" → positive
Text: "It's okay I guess" → neutral
Text: "Terrible experience" → negative
Text: "I love this product!" → """

2. Chain-of-Thought (CoT):

prompt = """Solve step by step:
Q: A store has 3 shelves with 12 items each. If 8 items are sold,
how many remain?

Think step by step:
- Total items = 3 × 12 = 36
- Items sold = 8
- Remaining = 36 - 8 = 28

Q: A server has 4 nodes with 8 pods each. If 5 pods crash,
how many pods remain?

Think step by step:"""

3. Role prompting:

system_prompt = """You are a senior DevOps engineer with 10 years of experience
in Kubernetes, Docker, and CI/CD. You give precise, production-ready answers
with actual code examples. You always mention potential pitfalls."""

4. ReAct prompting:

prompt = """Answer questions using this format:
Thought: Think about what to do
Action: tool_name[input]
Observation: result of action
... (repeat as needed)
Final Answer: your response

Question: What are the top 3 container orchestration platforms in 2024?
Thought:"""

Q12: How do you prevent prompt injection in AI Agents?

Answer:

# 1. Input validation
import re

def sanitize_input(user_input: str) -> str:
    # Remove common injection patterns
    dangerous_patterns = [
        r"ignore previous instructions",
        r"forget your system prompt",
        r"you are now",
        r"act as if",
    ]
    for pattern in dangerous_patterns:
        if re.search(pattern, user_input, re.IGNORECASE):
            raise ValueError("Potentially malicious input detected")
    return user_input

# 2. Separate system and user context
def build_prompt(system_instructions: str, user_input: str) -> list:
    return [
        {"role": "system", "content": system_instructions},
        # User input is NEVER injected into system prompt
        {"role": "user", "content": user_input}
    ]

# 3. Output validation
def validate_tool_call(tool_name: str, args: dict) -> bool:
    allowed_tools = {"search_web", "read_file", "calculate"}
    if tool_name not in allowed_tools:
        raise ValueError(f"Unauthorized tool: {tool_name}")
    return True

# 4. Sandboxed code execution
def execute_agent_code(code: str) -> str:
    # Use a sandboxed environment (Docker, subprocess with limits)
    import subprocess
    result = subprocess.run(
        ["python", "-c", code],
        capture_output=True,
        timeout=10,       # Kill if takes too long
        text=True
    )
    return result.stdout

Section 5: Agent Orchestration & Production

Q13: How do you deploy AI Agents in production?

Answer:

# FastAPI agent endpoint
from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel
import asyncio

app = FastAPI()

class AgentRequest(BaseModel):
    task: str
    session_id: str
    max_steps: int = 10

class AgentResponse(BaseModel):
    result: str
    steps: int
    tools_used: list[str]

@app.post("/agent/run", response_model=AgentResponse)
async def run_agent(request: AgentRequest):
    agent = create_agent(session_id=request.session_id)

    result = await asyncio.get_event_loop().run_in_executor(
        None,
        agent.run,
        request.task
    )

    return AgentResponse(
        result=result.output,
        steps=result.steps,
        tools_used=result.tools_used
    )

# Streaming responses
from fastapi.responses import StreamingResponse

@app.post("/agent/stream")
async def stream_agent(request: AgentRequest):
    async def generate():
        agent = create_streaming_agent()
        async for chunk in agent.astream(request.task):
            yield f"data: {chunk}\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream")

Q14: How do you evaluate AI Agent performance?

Answer:

# LangSmith for tracing and evaluation
from langsmith import Client
from langchain.callbacks.tracers import LangChainTracer

# Trace agent runs
tracer = LangChainTracer(project_name="my-agent-v1")
result = agent.run(task, callbacks=[tracer])

# Define evaluators
from langchain.evaluation import load_evaluator

# Correctness evaluator
evaluator = load_evaluator("qa", llm=ChatOpenAI())
eval_result = evaluator.evaluate_strings(
    input="What is Docker?",
    prediction=agent_response,
    reference="Docker is a containerization platform..."
)
print(eval_result["score"])  # 0-1

# Custom metric: tool call accuracy
def evaluate_tool_usage(trace: dict) -> float:
    """Check if agent used appropriate tools"""
    expected_tools = {"search_web"}
    actual_tools = {step["tool"] for step in trace["steps"] if "tool" in step}
    return len(expected_tools & actual_tools) / len(expected_tools)

# Benchmark suite
test_cases = [
    {"input": "What is Kubernetes?", "expected_tools": ["search_web"], "keywords": ["orchestration"]},
    {"input": "Calculate 15% of 200", "expected_tools": ["calculator"], "keywords": ["30"]},
]

scores = []
for test in test_cases:
    result = agent.run(test["input"])
    score = evaluate_response(result, test)
    scores.append(score)

print(f"Average score: {sum(scores)/len(scores):.2f}")

Q15: How does agent token management and cost optimization work?

Answer:

import tiktoken

def count_tokens(text: str, model: str = "gpt-4o") -> int:
    """Count tokens before sending to API"""
    encoder = tiktoken.encoding_for_model(model)
    return len(encoder.encode(text))

# Truncate conversation history to fit context window
def trim_history(messages: list, max_tokens: int = 100000) -> list:
    total = 0
    trimmed = []

    for msg in reversed(messages):
        tokens = count_tokens(msg["content"])
        if total + tokens > max_tokens:
            break
        trimmed.insert(0, msg)
        total += tokens

    return trimmed

# Cost tracking
PRICING = {
    "gpt-4o": {"input": 0.005, "output": 0.015},      # per 1K tokens
    "gpt-4o-mini": {"input": 0.00015, "output": 0.0006},
    "claude-3-5-sonnet": {"input": 0.003, "output": 0.015}
}

class CostTracker:
    def __init__(self, model: str):
        self.model = model
        self.total_input_tokens = 0
        self.total_output_tokens = 0

    def track(self, input_tokens: int, output_tokens: int):
        self.total_input_tokens += input_tokens
        self.total_output_tokens += output_tokens

    @property
    def total_cost(self) -> float:
        pricing = PRICING[self.model]
        return (
            self.total_input_tokens / 1000 * pricing["input"] +
            self.total_output_tokens / 1000 * pricing["output"]
        )

Section 6: Advanced AI Agent Concepts

Q16: What is the difference between ReAct, CoT, and Plan-and-Execute?

Answer:

Pattern	Description	Best for
CoT	Chain-of-thought; think step by step	Math, logic
ReAct	Interleave reasoning with actions	Tool-using tasks
Plan-and-Execute	Plan all steps upfront, then execute	Complex multi-step
Reflexion	Agent reflects on failures and retries	Reliability

# Plan-and-Execute pattern
from langchain_experimental.plan_and_execute import PlanAndExecute, load_agent_executor, load_chat_planner

planner = load_chat_planner(ChatOpenAI(temperature=0))
executor = load_agent_executor(ChatOpenAI(temperature=0), tools, verbose=True)

agent = PlanAndExecute(planner=planner, executor=executor)

# Agent will first create a full plan, then execute each step
result = agent.run("Research the top 5 container registries and create a comparison table")

Q17: How do you build a multi-agent pipeline for DevOps automation?

Answer:

# DevOps Automation Multi-Agent System

from autogen import AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager

config = {"model": "gpt-4o", "api_key": "YOUR_KEY"}

# Infrastructure agent
infra_agent = AssistantAgent(
    name="Infrastructure_Agent",
    system_message="""You manage cloud infrastructure using Terraform.
    When asked, generate Terraform configs and run terraform commands.
    Report any issues clearly.""",
    llm_config={"config_list": [config]}
)

# Deployment agent
deploy_agent = AssistantAgent(
    name="Deployment_Agent",
    system_message="""You handle Kubernetes deployments.
    Use kubectl commands to deploy applications and check health.
    Ensure zero-downtime deployments.""",
    llm_config={"config_list": [config]}
)

# Monitoring agent
monitor_agent = AssistantAgent(
    name="Monitoring_Agent",
    system_message="""You monitor system health using Prometheus/Grafana.
    Alert on anomalies and suggest fixes.
    Check metrics after every deployment.""",
    llm_config={"config_list": [config]}
)

# Security agent
security_agent = AssistantAgent(
    name="Security_Agent",
    system_message="""You perform security checks: scan images with Trivy,
    check IAM policies, validate network policies.
    Block deployments with CRITICAL vulnerabilities.""",
    llm_config={"config_list": [config]}
)

# Orchestrate
user_proxy = UserProxyAgent(name="DevOps_Orchestrator", human_input_mode="NEVER")

groupchat = GroupChat(
    agents=[user_proxy, infra_agent, deploy_agent, monitor_agent, security_agent],
    messages=[],
    max_round=30,
    speaker_selection_method="auto"
)

manager = GroupChatManager(groupchat=groupchat, llm_config={"config_list": [config]})

user_proxy.initiate_chat(
    manager,
    message="""Deploy myapp v2.0 to production:
    1. Security scan the image
    2. Update Terraform for any infra changes
    3. Deploy to Kubernetes with canary strategy
    4. Monitor for 10 minutes and confirm health
    5. Report summary"""
)

Q18: What are Agentic patterns you should know?

Answer:

1. Tool Selection Agent:

# Let LLM dynamically choose which tool to use
tools = [search_tool, calculator_tool, code_tool, database_tool]
agent = create_tool_calling_agent(llm, tools, prompt)

2. Subagent Delegation:

# Main agent spawns specialized subagents
@tool
def delegate_to_research_agent(task: str) -> str:
    """Delegate research tasks to a specialized research agent."""
    research_agent = create_research_agent()
    return research_agent.run(task)

3. Human-in-the-Loop:

# Pause and ask for approval
@tool
def request_human_approval(action: str, risk_level: str) -> str:
    """Request human approval before executing risky actions."""
    if risk_level == "high":
        user_input = input(f"Approve action '{action}'? (yes/no): ")
        return "approved" if user_input.lower() == "yes" else "rejected"
    return "auto-approved"

4. Self-Reflection/Critique:

def self_critique_agent(task: str, initial_answer: str) -> str:
    critique_prompt = f"""
    Task: {task}
    Answer: {initial_answer}

    Critique this answer:
    - Is it correct?
    - Is anything missing?
    - How can it be improved?
    """
    critique = llm.invoke(critique_prompt)

    improve_prompt = f"""
    Original answer: {initial_answer}
    Critique: {critique}

    Provide an improved answer addressing the critique:
    """
    return llm.invoke(improve_prompt)

Q19: How do you handle agent failures and retries?

Answer:

import time
from functools import wraps

def retry_with_backoff(max_retries: int = 3, base_delay: float = 1.0):
    """Retry decorator with exponential backoff"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_retries - 1:
                        raise
                    delay = base_delay * (2 ** attempt)
                    print(f"Attempt {attempt+1} failed: {e}. Retrying in {delay}s...")
                    time.sleep(delay)
        return wrapper
    return decorator

@retry_with_backoff(max_retries=3)
def call_llm_api(prompt: str) -> str:
    return llm.invoke(prompt)

# Fallback models
def call_with_fallback(prompt: str) -> str:
    models = ["gpt-4o", "gpt-4o-mini", "claude-3-haiku-20240307"]
    for model in models:
        try:
            return ChatOpenAI(model=model).invoke(prompt).content
        except Exception as e:
            print(f"Model {model} failed: {e}")
    raise RuntimeError("All models failed")

# Agent-level retry with different strategy
def agent_with_reflection(task: str, max_attempts: int = 3) -> str:
    history = []
    for i in range(max_attempts):
        result = agent.run(task + "\n\nPrevious attempts:\n" + "\n".join(history))

        if validate_result(result):
            return result

        feedback = get_failure_reason(result)
        history.append(f"Attempt {i+1} failed: {feedback}")

    return "Failed after max attempts"

Q20: What are the key safety considerations for AI Agents in production?

Answer:

# 1. Permission scoping — least privilege
class RestrictedToolkit:
    def __init__(self, allowed_paths: list, read_only: bool = True):
        self.allowed_paths = allowed_paths
        self.read_only = read_only

    def read_file(self, path: str) -> str:
        if not any(path.startswith(p) for p in self.allowed_paths):
            raise PermissionError(f"Access denied: {path}")
        with open(path) as f:
            return f.read()

    def write_file(self, path: str, content: str) -> str:
        if self.read_only:
            raise PermissionError("Write operations not permitted")
        # Additional checks...

# 2. Rate limiting
from collections import defaultdict
import time

class RateLimiter:
    def __init__(self, max_calls: int, period: float):
        self.max_calls = max_calls
        self.period = period
        self.calls = defaultdict(list)

    def check(self, key: str) -> bool:
        now = time.time()
        self.calls[key] = [t for t in self.calls[key] if now - t < self.period]

        if len(self.calls[key]) >= self.max_calls:
            return False

        self.calls[key].append(now)
        return True

# 3. Output sanitization
import re

def sanitize_agent_output(output: str) -> str:
    # Remove potential PII or sensitive data
    # Remove credit card patterns
    output = re.sub(r'\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b', '[REDACTED-CC]', output)
    # Remove potential passwords
    output = re.sub(r'(?i)password[s]?\s*[:=]\s*\S+', 'password: [REDACTED]', output)
    return output

# 4. Audit logging
import logging

agent_audit_log = logging.getLogger("agent.audit")

def log_agent_action(session_id: str, action: str, tool: str, input: str, output: str):
    agent_audit_log.info({
        "session_id": session_id,
        "action": action,
        "tool": tool,
        "input_hash": hash(input),  # Don't log raw sensitive inputs
        "output_preview": output[:100],
        "timestamp": time.time()
    })

Section 7: Advanced Topics & Production Patterns

Q21: What is the Model Context Protocol (MCP)?

Answer: MCP is an open standard by Anthropic that lets AI agents connect to external tools and data sources through a standardized interface — like USB-C but for AI tools.

Without MCP vs With MCP:

Without MCP:                     With MCP:
Each tool = custom integration   Any MCP-compatible tool works
Agent A tools ≠ Agent B tools    Agent A tools = Agent B tools
N×M integrations needed          N+M integrations needed

# MCP Server example (exposes tools to any MCP client)
from mcp import FastMCP

mcp = FastMCP("DevOps Tools Server")

@mcp.tool()
def get_kubernetes_pod_logs(namespace: str, pod_name: str, lines: int = 100) -> str:
    """Fetch logs from a Kubernetes pod."""
    import subprocess
    result = subprocess.run(
        ["kubectl", "logs", pod_name, "-n", namespace, f"--tail={lines}"],
        capture_output=True, text=True
    )
    return result.stdout

@mcp.tool()
def check_deployment_status(namespace: str, deployment: str) -> dict:
    """Check if a deployment is healthy."""
    import subprocess, json
    result = subprocess.run(
        ["kubectl", "get", "deployment", deployment, "-n", namespace, "-o", "json"],
        capture_output=True, text=True
    )
    data = json.loads(result.stdout)
    return {
        "ready": data["status"].get("readyReplicas", 0),
        "desired": data["spec"]["replicas"],
        "healthy": data["status"].get("readyReplicas", 0) == data["spec"]["replicas"]
    }

# Run the MCP server
if __name__ == "__main__":
    mcp.run()

# Client (Claude Desktop, Claude API, custom agent) connects automatically

Interview tip: MCP is becoming the standard for agent tool integration. Instead of writing custom tool code for each agent, you write one MCP server and any MCP-compatible client can use it. Claude, Cursor, and many IDEs already support MCP.

Q22: What is the OpenAI Assistants API and how does it differ from Chat Completions?

Answer: The Assistants API is a higher-level abstraction that manages threads, runs, and tool calls automatically — you don’t manage conversation history manually.

Feature	Chat Completions	Assistants API
History management	You manage	OpenAI manages (Threads)
File handling	Manual	Built-in (File Search tool)
Code execution	Manual	Built-in (Code Interpreter)
State	Stateless	Stateful (Threads persist)
Cost	Lower	Higher (storage fees)
Control	Full	Less (black box)

from openai import OpenAI

client = OpenAI()

# Create an Assistant (once, reuse across conversations)
assistant = client.beta.assistants.create(
    name="DevOps Helper",
    instructions="""You are a DevOps expert. Help with Kubernetes, Docker, and CI/CD.
    When given logs or configs, analyze them and suggest fixes.""",
    model="gpt-4o",
    tools=[
        {"type": "code_interpreter"},   # Run Python code
        {"type": "file_search"},        # Search uploaded docs
    ]
)

# Create a Thread (one per user/conversation)
thread = client.beta.threads.create()

# Add message to thread
client.beta.threads.messages.create(
    thread_id=thread.id,
    role="user",
    content="My nginx pod keeps CrashLoopBackOff. Here are the logs: [ERROR] ..."
)

# Run the assistant
run = client.beta.threads.runs.create_and_poll(
    thread_id=thread.id,
    assistant_id=assistant.id
)

# Get response
messages = client.beta.threads.messages.list(thread_id=thread.id)
print(messages.data[0].content[0].text.value)

Interview tip: Use Assistants API for quick prototypes and user-facing apps where you want built-in file handling. Use Chat Completions + manual orchestration (LangChain/LangGraph) for production systems where you need full control over costs, retries, and observability.

Q23: What is CrewAI and how do you build multi-agent teams?

Answer: CrewAI enables “crews” of specialized AI agents that collaborate on complex tasks with defined roles, goals, and hierarchical or sequential processes.

from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool, FileReadTool

# Define specialized agents with roles
researcher = Agent(
    role='Senior DevOps Researcher',
    goal='Research the latest Kubernetes security vulnerabilities and CVEs',
    backstory="""You are an expert security researcher specializing in container
    security and Kubernetes. You always verify findings from multiple sources.""",
    tools=[SerperDevTool()],    # Web search
    verbose=True,
    max_iter=5,
    llm="gpt-4o"
)

writer = Agent(
    role='Technical Documentation Writer',
    goal='Write clear, actionable security advisories',
    backstory="""You transform complex security findings into clear remediation
    guides that DevOps teams can act on immediately.""",
    verbose=True,
    llm="gpt-4o"
)

reviewer = Agent(
    role='Security Lead',
    goal='Review and validate security findings before publication',
    backstory="You ensure accuracy and completeness of all security documentation.",
    verbose=True,
    llm="gpt-4o"
)

# Define tasks
research_task = Task(
    description="Research top 5 critical Kubernetes CVEs in 2024. Include CVE IDs, severity, affected versions.",
    expected_output="A detailed list of 5 CVEs with technical details and impact assessment.",
    agent=researcher
)

write_task = Task(
    description="Write a security advisory based on the research. Include remediation steps.",
    expected_output="A formatted security advisory with actionable remediation steps.",
    agent=writer,
    context=[research_task]   # Depends on research_task output
)

review_task = Task(
    description="Review the advisory for accuracy. Approve or request changes.",
    expected_output="Reviewed and approved security advisory or list of corrections needed.",
    agent=reviewer,
    context=[write_task]
)

# Assemble the crew
crew = Crew(
    agents=[researcher, writer, reviewer],
    tasks=[research_task, write_task, review_task],
    process=Process.sequential,   # sequential | hierarchical
    verbose=True,
    memory=True   # Agents share memory
)

result = crew.kickoff()
print(result)

Interview tip: CrewAI uses role-playing to get specialized behavior from the same underlying LLM. The backstory is crucial — it sets the agent’s “personality” and expertise. Use Process.hierarchical with a manager LLM when you want dynamic task assignment instead of predefined order.

Q24: What is Microsoft Semantic Kernel and when would you use it?

Answer: Semantic Kernel is Microsoft’s enterprise-grade AI orchestration SDK (C#, Python, Java) that integrates AI into existing applications with plugins, planners, and memory.

import asyncio
from semantic_kernel import Kernel
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion
from semantic_kernel.functions import kernel_function

kernel = Kernel()

# Add AI service
kernel.add_service(OpenAIChatCompletion(
    service_id="openai",
    ai_model_id="gpt-4o",
))

# Define a plugin (collection of related functions)
class DevOpsPlugin:
    @kernel_function(
        name="analyze_logs",
        description="Analyze application logs and identify errors or anomalies"
    )
    def analyze_logs(self, logs: str) -> str:
        # This gets wrapped as an LLM-callable function
        return f"Analyzing: {logs[:500]}..."

    @kernel_function(
        name="suggest_fix",
        description="Suggest a fix for a given error message"
    )
    def suggest_fix(self, error: str) -> str:
        return f"For error '{error}', try: ..."

# Register plugin
kernel.add_plugin(DevOpsPlugin(), plugin_name="DevOps")

# Use Handlebars prompt template
prompt = """
{{$input}}
Based on this, {{DevOps.analyze_logs input}}
Then {{DevOps.suggest_fix input}}
"""

async def main():
    result = await kernel.invoke_prompt(
        prompt,
        input="ERROR: Pod nginx-xxx CrashLoopBackOff - OOMKilled"
    )
    print(result)

asyncio.run(main())

SK vs LangChain:

	Semantic Kernel	LangChain
Primary language	C# (best), Python	Python (best)
Enterprise focus	Very high (Microsoft)	Medium
Azure integration	Native	Via integrations
Learning curve	Higher	Medium
Ecosystem	Growing	Large/mature

Interview tip: Semantic Kernel is the go-to for .NET/Azure shops and enterprises already using Microsoft stack. LangChain is better for Python-first teams with broader ecosystem needs.

Q25: What is the difference between fine-tuning and RAG? When to use each?

Answer:

Aspect	Fine-tuning	RAG
How	Train model on your data	Retrieve relevant docs at query time
When to update	Retrain model (hours/days)	Update vector DB (minutes)
Cost	High (GPU training + inference)	Lower (inference + vector search)
Latency	Same as base model	Slightly higher (retrieval step)
Hallucination	Can still hallucinate	Grounded in retrieved context
Best for	Style/format/behavior changes	Factual knowledge, dynamic data

# Fine-tuning use case: teaching model a specific format
# e.g., "Always respond with valid JSON in our schema"

# Fine-tuning with OpenAI
from openai import OpenAI
client = OpenAI()

# Prepare training data (JSONL format)
training_data = [
    {"messages": [
        {"role": "system", "content": "You are a DevOps bot that outputs ONLY valid JSON."},
        {"role": "user", "content": "Check pod status"},
        {"role": "assistant", "content": '{"action": "kubectl_get", "resource": "pod", "status": "pending"}'}
    ]},
    # ... more examples
]

# Create fine-tuning job
job = client.fine_tuning.jobs.create(
    training_file="file-abc123",
    model="gpt-4o-mini",
    hyperparameters={"n_epochs": 3}
)

# Use fine-tuned model
response = client.chat.completions.create(
    model=job.fine_tuned_model,  # e.g., "ft:gpt-4o-mini:myorg::abc123"
    messages=[{"role": "user", "content": "Check pod status"}]
)

Decision guide:

Your data changes frequently?    → RAG
Need specific output format?     → Fine-tuning
Need domain knowledge?           → RAG (faster to update)
Need different personality/style? → Fine-tuning
Want to reduce prompt length?    → Fine-tuning (bakes in instructions)
On a budget?                     → RAG (no training cost)

Interview tip: “Fine-tuning teaches HOW to respond; RAG teaches WHAT to respond with. Most production systems combine both: fine-tune for consistent behavior/format, RAG for up-to-date knowledge.”

Q26: How do you implement streaming responses in AI Agents?

Answer: Streaming sends tokens as they’re generated instead of waiting for the full response — dramatically improves perceived latency for users.

from openai import OpenAI
from langchain_openai import ChatOpenAI
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler

client = OpenAI()

# --- Method 1: OpenAI streaming ---
def stream_openai_response(prompt: str):
    stream = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        stream=True  # Enable streaming
    )

    full_response = ""
    for chunk in stream:
        delta = chunk.choices[0].delta.content
        if delta:
            print(delta, end="", flush=True)  # Print as it arrives
            full_response += delta

    return full_response

# --- Method 2: LangChain streaming ---
llm = ChatOpenAI(
    model="gpt-4o",
    streaming=True,
    callbacks=[StreamingStdOutCallbackHandler()]
)
response = llm.invoke("Explain Kubernetes networking")

# --- Method 3: FastAPI SSE streaming endpoint ---
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from langchain_openai import ChatOpenAI
import json

app = FastAPI()

@app.post("/stream")
async def stream_agent(request: dict):
    llm = ChatOpenAI(model="gpt-4o", streaming=True)

    async def generate():
        async for chunk in llm.astream(request["message"]):
            # Server-Sent Events format
            data = json.dumps({"token": chunk.content, "done": False})
            yield f"data: {data}\n\n"

        yield f"data: {json.dumps({'done': True})}\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream")

# Frontend JavaScript:
# const es = new EventSource('/stream');
# es.onmessage = (e) => { const d = JSON.parse(e.data); console.log(d.token); }

Interview tip: Streaming is essential for chat interfaces — users see “thinking” instantly rather than waiting 10 seconds for a full response. Always stream for user-facing agents. For background batch processing, streaming is unnecessary overhead.

Q27: How do you implement guardrails and output validation for AI Agents?

Answer: Guardrails ensure agent outputs meet safety, format, and quality requirements before being shown to users or used in downstream systems.

from pydantic import BaseModel, validator, Field
from typing import Literal
import re

# --- Method 1: Pydantic structured output (format validation) ---
class DeploymentDecision(BaseModel):
    action: Literal["deploy", "rollback", "scale", "no_action"]
    target_environment: Literal["dev", "staging", "production"]
    reason: str = Field(min_length=10, max_length=500)
    risk_level: Literal["low", "medium", "high", "critical"]
    requires_human_approval: bool

    @validator('requires_human_approval', always=True)
    def high_risk_needs_approval(cls, v, values):
        if values.get('risk_level') in ['high', 'critical'] and not v:
            raise ValueError("High/critical risk actions must require human approval")
        return v

# Force LLM to output this structure
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import PydanticOutputParser

llm = ChatOpenAI(model="gpt-4o")
structured_llm = llm.with_structured_output(DeploymentDecision)

decision = structured_llm.invoke(
    "Should we deploy v2.1 to production? Current error rate is 0.1%"
)
print(decision.action)          # Validated enum
print(decision.requires_human_approval)  # Always True if high risk

# --- Method 2: Guardrails library ---
from guardrails import Guard
from guardrails.hub import ToxicLanguage, ValidLength

guard = Guard().use_many(
    ToxicLanguage(threshold=0.5, on_fail="exception"),
    ValidLength(min=10, max=1000, on_fail="reask")
)

validated_output, *rest = guard(
    llm_api=client.chat.completions.create,
    prompt="Summarize this deployment log: ...",
    model="gpt-4o"
)

# --- Method 3: Content safety checks ---
def check_output_safety(output: str) -> tuple[bool, str]:
    """Check agent output for sensitive data or harmful content."""
    issues = []

    # No hardcoded secrets in output
    if re.search(r'(?i)(password|secret|api[_-]?key)\s*[=:]\s*\S{8,}', output):
        issues.append("Output contains potential secret")

    # No internal IPs
    if re.search(r'10\.\d+\.\d+\.\d+|192\.168\.\d+\.\d+', output):
        issues.append("Output contains internal IP")

    return len(issues) == 0, "; ".join(issues)

Interview tip: Use Pydantic structured output to guarantee JSON schema compliance. Use content safety checks to prevent data leakage. For production, always validate BOTH the format (structure) AND the content (safety) of agent outputs.

Q28: How do you implement agent observability and tracing?

Answer: Observability lets you understand what your agent did, why it made decisions, and where it failed — essential for debugging and improving production agents.

# --- LangSmith (LangChain's tracing platform) ---
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-langsmith-key"
os.environ["LANGCHAIN_PROJECT"] = "my-devops-agent-prod"

# All LangChain calls are now automatically traced!
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor

agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# Add custom metadata to traces
result = agent_executor.invoke(
    {"input": "Check prod deployment"},
    config={
        "metadata": {
            "user_id": "user-123",
            "session_id": "sess-abc",
            "environment": "production",
            "agent_version": "v2.1.0"
        }
    }
)

# --- OpenTelemetry for custom tracing ---
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider

tracer = trace.get_tracer("ai-agent")

def traced_agent_run(task: str, session_id: str):
    with tracer.start_as_current_span("agent.run") as span:
        span.set_attribute("agent.task", task[:200])
        span.set_attribute("session.id", session_id)

        try:
            result = agent.run(task)
            span.set_attribute("agent.success", True)
            span.set_attribute("agent.result_length", len(result))
            return result
        except Exception as e:
            span.record_exception(e)
            span.set_attribute("agent.success", False)
            raise

# --- Structured logging for agent steps ---
import structlog

log = structlog.get_logger()

class ObservableAgentCallback:
    def on_tool_start(self, tool_name: str, tool_input: str, **kwargs):
        log.info("tool_called", tool=tool_name, input_preview=tool_input[:100])

    def on_tool_end(self, output: str, **kwargs):
        log.info("tool_completed", output_length=len(output))

    def on_llm_start(self, serialized: dict, prompts: list, **kwargs):
        log.info("llm_called", model=serialized.get("id", ["unknown"])[-1])

    def on_llm_end(self, response, **kwargs):
        usage = response.llm_output.get("token_usage", {})
        log.info("llm_completed",
                 prompt_tokens=usage.get("prompt_tokens"),
                 completion_tokens=usage.get("completion_tokens"))

Interview tip: For production agents, LangSmith is the easiest observability solution. Each “trace” shows the full agent reasoning chain — every LLM call, every tool call, every token used. This is how you debug “why did the agent do X?” in production.

Answer: Multi-modal agents can process and generate text, images, audio, and other media — enabling richer interactions beyond pure text.

import base64
from openai import OpenAI
from pathlib import Path

client = OpenAI()

# --- Vision: Analyze images ---
def analyze_dashboard_screenshot(image_path: str, question: str) -> str:
    """Let an agent analyze a monitoring dashboard screenshot."""
    with open(image_path, "rb") as f:
        image_data = base64.b64encode(f.read()).decode("utf-8")

    response = client.chat.completions.create(
        model="gpt-4o",  # Vision-capable model
        messages=[{
            "role": "user",
            "content": [
                {
                    "type": "image_url",
                    "image_url": {
                        "url": f"data:image/png;base64,{image_data}",
                        "detail": "high"  # high | low | auto
                    }
                },
                {
                    "type": "text",
                    "text": question
                }
            ]
        }],
        max_tokens=1000
    )
    return response.choices[0].message.content

# Usage: agent reads a Grafana screenshot
analysis = analyze_dashboard_screenshot(
    "grafana_dashboard.png",
    "Is there a spike in error rate? What time did it start? What metric is most concerning?"
)

# --- Tool that accepts images ---
from langchain.tools import tool

@tool
def analyze_error_screenshot(image_url: str, context: str = "") -> str:
    """Analyze a screenshot of an error or dashboard.
    Use when user shares an image of logs, errors, or metrics."""
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{
            "role": "user",
            "content": [
                {"type": "image_url", "image_url": {"url": image_url}},
                {"type": "text", "text": f"Analyze this image. Context: {context}. What issues do you see?"}
            ]
        }]
    )
    return response.choices[0].message.content

# --- Audio transcription → agent processing ---
def process_voice_command(audio_file: str) -> str:
    """Transcribe voice command and process with agent."""
    # 1. Transcribe
    with open(audio_file, "rb") as f:
        transcript = client.audio.transcriptions.create(
            model="whisper-1",
            file=f
        )

    # 2. Process with agent
    return agent.run(transcript.text)

Interview tip: Multi-modal agents are increasingly important for DevOps — imagine an agent that can look at a Grafana dashboard screenshot, identify anomalies, correlate with logs, and auto-create an incident ticket. GPT-4o natively supports vision; Claude 3.5 Sonnet also has strong vision capabilities.

Q30: How do you implement agent workflows with human-in-the-loop approval?

Answer: Human-in-the-loop (HITL) lets you pause agent execution and wait for human review before taking irreversible actions.

from langgraph.graph import StateGraph, END, interrupt
from langgraph.checkpoint.memory import MemorySaver
from typing import TypedDict

class AgentState(TypedDict):
    task: str
    plan: str
    action: str
    human_approved: bool
    result: str

def planning_node(state: AgentState) -> dict:
    """Agent creates a plan."""
    plan = llm.invoke(f"Create a deployment plan for: {state['task']}")
    return {"plan": plan.content}

def human_approval_node(state: AgentState) -> dict:
    """Pause and wait for human approval."""
    print(f"\n🔍 HUMAN REVIEW REQUIRED")
    print(f"Plan: {state['plan']}")

    # This PAUSES the graph and returns to the caller
    # Resume by calling: app.invoke(None, config={"configurable": {"thread_id": tid}})
    human_input = interrupt({
        "message": "Please review this plan",
        "plan": state["plan"],
        "requires_approval": True
    })

    return {"human_approved": human_input.get("approved", False)}

def execute_node(state: AgentState) -> dict:
    """Execute only if approved."""
    if not state["human_approved"]:
        return {"result": "Action rejected by human reviewer"}

    result = execute_deployment(state["plan"])
    return {"result": result}

def route_after_approval(state: AgentState) -> str:
    return "execute" if state["human_approved"] else "rejected"

# Build graph with checkpointing (required for interrupt)
workflow = StateGraph(AgentState)
workflow.add_node("plan", planning_node)
workflow.add_node("human_review", human_approval_node)
workflow.add_node("execute", execute_node)
workflow.set_entry_point("plan")
workflow.add_edge("plan", "human_review")
workflow.add_conditional_edges("human_review", route_after_approval, {
    "execute": "execute",
    "rejected": END
})
workflow.add_edge("execute", END)

# Compile with memory checkpointer
app = workflow.compile(checkpointer=MemorySaver())

thread_id = "deploy-session-001"
config = {"configurable": {"thread_id": thread_id}}

# Step 1: Run until interrupt
result = app.invoke({"task": "Deploy v2.0 to production"}, config=config)
print("Waiting for human approval...")

# Step 2: Human reviews, then resumes
# In a web app, this would be an API call from the approval UI
resume_result = app.invoke(
    {"approved": True},   # Human's decision
    config=config
)

Interview tip: LangGraph’s interrupt() is the cleanest way to implement HITL. The graph state is checkpointed, the workflow pauses, and you can resume it hours later after human review. Use this for any action that is: irreversible, expensive, or high-risk (production deployments, deletions, large API calls).

Q31: How do you test AI Agents?

Answer: Agent testing requires specialized strategies beyond unit tests — you need to verify reasoning, tool selection, and output quality.

import pytest
from unittest.mock import Mock, patch
from langchain.agents import AgentExecutor

# --- Level 1: Unit test individual tools ---
def test_k8s_tool_returns_pod_list():
    """Test the tool function in isolation."""
    with patch("subprocess.run") as mock_run:
        mock_run.return_value = Mock(
            stdout='{"items": [{"metadata": {"name": "nginx-pod"}}]}',
            returncode=0
        )
        result = get_kubernetes_pods("default")
        assert "nginx-pod" in result

# --- Level 2: Mock LLM for deterministic agent tests ---
def test_agent_calls_correct_tool_for_pod_query():
    """Verify agent selects the right tool."""
    from langchain_core.messages import AIMessage, ToolCall

    # Mock LLM response with predetermined tool call
    mock_llm = Mock()
    mock_llm.invoke.return_value = AIMessage(
        content="",
        tool_calls=[ToolCall(
            name="get_kubernetes_pods",
            args={"namespace": "production"},
            id="call_123"
        )]
    )

    agent = create_devops_agent(llm=mock_llm)
    # Test that agent routes "show pods" to correct tool
    agent.invoke({"input": "Show all pods in production"})

    assert mock_llm.invoke.called

# --- Level 3: Evaluation with LLM-as-judge ---
from langchain.evaluation import load_evaluator

def evaluate_agent_response(question: str, expected_keywords: list) -> float:
    """Use LLM to judge response quality."""
    response = agent.invoke({"input": question})["output"]

    evaluator = load_evaluator("criteria", criteria={
        "relevance": "Does the response directly answer the question?",
        "accuracy": "Is the information technically accurate?",
        "completeness": "Does it cover all important aspects?"
    })

    result = evaluator.evaluate_strings(
        input=question,
        prediction=response,
        reference=f"Should mention: {', '.join(expected_keywords)}"
    )
    return result["score"]

# --- Level 4: Regression test suite ---
test_cases = [
    {
        "input": "What is the difference between a Deployment and StatefulSet?",
        "must_contain": ["stateful", "persistent", "ordered"],
        "must_not_contain": ["I don't know", "I'm not sure"]
    },
    {
        "input": "How do I scale a deployment to 5 replicas?",
        "must_contain": ["kubectl scale", "replicas"],
    }
]

@pytest.mark.parametrize("case", test_cases)
def test_agent_regression(case):
    result = agent.invoke({"input": case["input"]})["output"].lower()
    for keyword in case.get("must_contain", []):
        assert keyword.lower() in result, f"Missing keyword: {keyword}"
    for keyword in case.get("must_not_contain", []):
        assert keyword.lower() not in result, f"Should not contain: {keyword}"

Interview tip: Agent testing has 4 levels: (1) unit test tools, (2) mock LLM for deterministic routing tests, (3) LLM-as-judge for quality evaluation, (4) regression suite to catch regressions as you update prompts or models. All 4 are needed in production.

Q32: How do you implement caching for AI Agents to reduce cost?

Answer: Caching eliminates redundant LLM calls for identical or similar inputs, cutting costs significantly.

import hashlib
import json
import redis
from langchain.cache import RedisSemanticCache
from langchain_openai import OpenAIEmbeddings
from langchain_core.globals import set_llm_cache

# --- Method 1: Exact match cache (Redis) ---
redis_client = redis.Redis(host="localhost", port=6379)

def cached_llm_call(prompt: str, model: str = "gpt-4o") -> str:
    """Cache exact prompt → response pairs."""
    cache_key = f"llm:{hashlib.md5(f'{model}:{prompt}'.encode()).hexdigest()}"

    # Check cache
    cached = redis_client.get(cache_key)
    if cached:
        return json.loads(cached)

    # Call LLM
    response = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}]
    )
    result = response.choices[0].message.content

    # Cache for 24 hours
    redis_client.setex(cache_key, 86400, json.dumps(result))
    return result

# --- Method 2: Semantic cache (similar questions hit same cache) ---
# "How do I scale K8s deployment?" and "How to scale kubernetes deployment?"
# → same semantic meaning → same cached response!
set_llm_cache(RedisSemanticCache(
    redis_url="redis://localhost:6379",
    embedding=OpenAIEmbeddings(),
    score_threshold=0.95   # 95% similarity = cache hit
))

# LangChain automatically checks semantic cache now
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o")
result = llm.invoke("How do I scale a Kubernetes deployment?")  # Cached if similar asked before

# --- Method 3: Tool result caching ---
from functools import lru_cache
import time

tool_cache = {}

def cached_tool(tool_name: str, args: dict, ttl_seconds: int = 300):
    """Cache tool results to avoid repeated API/DB calls."""
    cache_key = f"{tool_name}:{json.dumps(args, sort_keys=True)}"

    if cache_key in tool_cache:
        result, timestamp = tool_cache[cache_key]
        if time.time() - timestamp < ttl_seconds:
            return result  # Fresh cache hit

    # Execute tool
    result = execute_tool(tool_name, args)
    tool_cache[cache_key] = (result, time.time())
    return result

# Cost savings example:
# Without cache: 1000 users ask "What is Docker?" → 1000 LLM calls @ $0.01 = $10
# With cache:    1000 users ask "What is Docker?" → 1 LLM call + 999 cache hits = $0.01

Interview tip: Semantic caching is the most powerful — it catches paraphrased versions of the same question. Combine exact + semantic cache: exact cache for identical prompts (fast), semantic cache for similar prompts (intelligent). Always cache tool results too, not just LLM calls.

Q33: What are knowledge graphs and how do they enhance AI Agents?

Answer: Knowledge graphs store entities and relationships in a structured graph format, enabling agents to reason about connections that vector search alone can’t find.

from neo4j import GraphDatabase
from langchain_community.graphs import Neo4jGraph
from langchain.chains import GraphCypherQAChain
from langchain_openai import ChatOpenAI

# --- Build a DevOps knowledge graph ---
driver = GraphDatabase.driver("bolt://localhost:7687", auth=("neo4j", "password"))

with driver.session() as session:
    # Create nodes and relationships
    session.run("""
        CREATE (k8s:Technology {name: 'Kubernetes', type: 'orchestrator'})
        CREATE (docker:Technology {name: 'Docker', type: 'container-runtime'})
        CREATE (helm:Technology {name: 'Helm', type: 'package-manager'})
        CREATE (argocd:Technology {name: 'ArgoCD', type: 'gitops-tool'})

        CREATE (k8s)-[:USES]->(docker)
        CREATE (helm)-[:DEPLOYS_TO]->(k8s)
        CREATE (argocd)-[:MANAGES]->(k8s)
        CREATE (argocd)-[:USES]->(helm)
    """)

# --- Query knowledge graph with natural language ---
graph = Neo4jGraph(url="bolt://localhost:7687", username="neo4j", password="password")

chain = GraphCypherQAChain.from_llm(
    llm=ChatOpenAI(model="gpt-4o"),
    graph=graph,
    verbose=True
)

# LLM converts natural language → Cypher query → executes → natural language answer
result = chain.invoke("What tools does ArgoCD use?")
# LLM generates: MATCH (a:Technology {name: 'ArgoCD'})-[:USES]->(t) RETURN t.name
# Executes query, gets "Helm"
# Returns: "ArgoCD uses Helm for package management"

# Hybrid RAG + Knowledge Graph
def hybrid_search(question: str) -> str:
    """Combine vector search (facts) with graph search (relationships)."""
    # Vector search: find relevant documents
    docs = vector_retriever.get_relevant_documents(question)

    # Graph search: find related entities
    entities = extract_entities(question)  # e.g., ["ArgoCD", "Kubernetes"]
    graph_context = graph.query(f"""
        MATCH (n) WHERE n.name IN {entities}
        MATCH (n)-[r]-(m)
        RETURN n.name, type(r), m.name LIMIT 20
    """)

    # Combine both contexts for LLM
    return llm.invoke(f"""
        Documentation: {docs}
        Relationships: {graph_context}
        Question: {question}
    """)

Interview tip: Vector search finds “similar” content. Knowledge graphs find “related” content via explicit relationships. Example: “What breaks if I update the K8s version?” — a knowledge graph can traverse relationships to find all dependent tools; vector search cannot reason about dependencies.

Q34: How do you handle context window limits in long-running agents?

Answer: Context windows are finite. Long tasks accumulate history that exceeds the limit, causing failures or degraded quality.

import tiktoken
from langchain_openai import ChatOpenAI
from langchain.memory import ConversationSummaryBufferMemory

# --- Strategy 1: Sliding window (keep last N messages) ---
def sliding_window_messages(messages: list, max_messages: int = 20) -> list:
    """Keep system message + last N messages."""
    system_msgs = [m for m in messages if m["role"] == "system"]
    non_system = [m for m in messages if m["role"] != "system"]
    return system_msgs + non_system[-max_messages:]

# --- Strategy 2: Summary memory (compress old messages) ---
memory = ConversationSummaryBufferMemory(
    llm=ChatOpenAI(model="gpt-4o-mini"),  # Cheap model for summarization
    max_token_limit=2000,  # Keep recent messages up to 2000 tokens
    # Older messages are summarized: "Previously: user asked about pods, agent found 3 running..."
)

# --- Strategy 3: Token counting + truncation ---
def trim_to_token_limit(messages: list, model: str = "gpt-4o", max_tokens: int = 100000) -> list:
    encoder = tiktoken.encoding_for_model(model)

    total_tokens = 0
    result = []

    # Always keep system message
    system_msgs = [m for m in messages if m["role"] == "system"]
    for msg in system_msgs:
        total_tokens += len(encoder.encode(msg["content"]))
        result.append(msg)

    # Add recent messages until limit
    for msg in reversed([m for m in messages if m["role"] != "system"]):
        tokens = len(encoder.encode(msg["content"]))
        if total_tokens + tokens > max_tokens:
            break
        result.insert(len(system_msgs), msg)
        total_tokens += tokens

    return result

# --- Strategy 4: Hierarchical summarization for very long tasks ---
def hierarchical_memory(steps: list) -> str:
    """Summarize old steps in batches."""
    if len(steps) <= 10:
        return str(steps)

    # Summarize oldest batch
    old_summary = llm.invoke(f"Summarize these agent steps concisely: {steps[:5]}")

    # Keep summary + recent steps
    return f"[Earlier summary]: {old_summary.content}\n[Recent steps]: {steps[5:]}"

Context window comparison:

Model	Context	~Pages of text
GPT-4o	128K tokens	~200 pages
Claude 3.5 Sonnet	200K tokens	~320 pages
Gemini 1.5 Pro	1M tokens	~1600 pages
Gemini 1.5 Flash	1M tokens	~1600 pages

Interview tip: Even with 1M token windows, you should still implement context management — larger contexts = higher cost + slower responses. Use summary memory for ongoing conversations, sliding window for task-focused agents.

Q35: How do you implement agent security — preventing data exfiltration?

Answer: Agents with tool access can be tricked into leaking sensitive data through prompt injection or misconfigured permissions.

import re
from typing import Optional

# --- 1. Tool permission scoping ---
class SecureToolRegistry:
    def __init__(self, user_role: str):
        self.user_role = user_role
        self._tools = {}
        self._permissions = {
            "readonly": ["search_docs", "get_pod_status", "view_logs"],
            "operator": ["search_docs", "get_pod_status", "view_logs", "scale_deployment"],
            "admin": ["search_docs", "get_pod_status", "view_logs", "scale_deployment",
                     "delete_deployment", "create_secret"]
        }

    def get_allowed_tools(self) -> list:
        allowed_names = self._permissions.get(self.user_role, [])
        return [t for name, t in self._tools.items() if name in allowed_names]

    def register(self, name: str, tool):
        self._tools[name] = tool

# --- 2. Output sanitization ---
SENSITIVE_PATTERNS = [
    (r'(?i)(password|secret|token|api[_-]?key)\s*[=:]\s*[^\s]{6,}', '[REDACTED]'),
    (r'\b\d{4}[- ]?\d{4}[- ]?\d{4}[- ]?\d{4}\b', '[CARD-REDACTED]'),
    (r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', '[EMAIL-REDACTED]'),
    (r'10\.\d+\.\d+\.\d+', '[INTERNAL-IP]'),
]

def sanitize_output(text: str, allow_internal: bool = False) -> str:
    patterns = SENSITIVE_PATTERNS if not allow_internal else SENSITIVE_PATTERNS[:3]
    for pattern, replacement in patterns:
        text = re.sub(pattern, replacement, text)
    return text

# --- 3. Prompt injection detection ---
INJECTION_SIGNALS = [
    "ignore previous instructions",
    "forget your system prompt",
    "you are now",
    "act as if you are",
    "disregard all prior",
    "new instructions:",
    "override:",
]

def detect_injection(user_input: str) -> Optional[str]:
    lower = user_input.lower()
    for signal in INJECTION_SIGNALS:
        if signal in lower:
            return f"Potential prompt injection detected: '{signal}'"
    return None

# --- 4. Data access logging ---
import logging
audit_logger = logging.getLogger("agent.security")

def secure_tool_call(tool_name: str, args: dict, user_id: str):
    """Log all tool calls for security audit."""
    audit_logger.info(
        "tool_access",
        extra={
            "user_id": user_id,
            "tool": tool_name,
            "args_hash": hash(str(sorted(args.items()))),
            "timestamp": time.time()
        }
    )
    return tools[tool_name](**args)

Interview tip: Security for agents has 4 pillars: (1) Least-privilege tool access — agents only get the tools they need. (2) Output sanitization — scrub secrets before showing to users. (3) Injection detection — validate all user inputs. (4) Audit logging — log every tool call for forensics.

Q36: What is Retrieval-Augmented Generation vs. Agentic RAG?

Answer:

	Basic RAG	Agentic RAG
Retrieval	One-shot retrieve → answer	Iterative: retrieve → reason → re-retrieve
Query	Original question only	Agent reformulates queries
Sources	Single vector DB	Multiple sources (web, DB, APIs)
Verification	None	Agent verifies answer quality
Use case	Simple Q&A	Complex research, multi-hop questions

from langchain.tools import tool
from langchain_community.vectorstores import Chroma
from langchain_openai import ChatOpenAI, OpenAIEmbeddings

vectorstore = Chroma(embedding_function=OpenAIEmbeddings())

# --- Basic RAG (one-shot) ---
def basic_rag(question: str) -> str:
    docs = vectorstore.similarity_search(question, k=3)
    context = "\n".join([d.page_content for d in docs])
    return llm.invoke(f"Answer using this context:\n{context}\n\nQuestion: {question}").content

# --- Agentic RAG (iterative, multi-step) ---
@tool
def search_knowledge_base(query: str) -> str:
    """Search the internal knowledge base. Use specific, targeted queries for best results."""
    docs = vectorstore.similarity_search(query, k=3)
    return "\n---\n".join([f"Source: {d.metadata.get('source', 'unknown')}\n{d.page_content}" for d in docs])

@tool
def search_web(query: str) -> str:
    """Search the web for current information not in the knowledge base."""
    return web_searcher.run(query)

@tool
def verify_answer(answer: str, sources: str) -> str:
    """Verify if an answer is supported by the provided sources."""
    verification = llm.invoke(f"""
    Answer: {answer}
    Sources: {sources}

    Is this answer fully supported by the sources?
    What's missing or unverified?
    """)
    return verification.content

# Agentic RAG agent automatically:
# 1. Searches KB → insufficient info
# 2. Reformulates query, searches again
# 3. Searches web for latest info
# 4. Verifies answer against sources
# 5. Returns verified, sourced answer
agentic_rag = AgentExecutor(
    agent=create_tool_calling_agent(
        ChatOpenAI(model="gpt-4o"),
        [search_knowledge_base, search_web, verify_answer],
        prompt
    ),
    tools=[search_knowledge_base, search_web, verify_answer]
)

Interview tip: “Basic RAG is like asking a search engine — one query, static results. Agentic RAG is like a researcher — it searches, reads, realizes it needs more info, searches again with a better query, verifies the answer. Use Agentic RAG for complex, multi-hop questions where simple search fails.”

Q37: How do you implement A/B testing for AI Agents?

Answer: A/B testing lets you compare different agent versions, prompts, or models objectively using real traffic.

import random
import time
from dataclasses import dataclass
from typing import Callable

@dataclass
class AgentVariant:
    name: str
    agent: Callable
    weight: float   # Traffic allocation (0.0-1.0)

class AgentABTest:
    def __init__(self, variants: list[AgentVariant]):
        self.variants = variants
        self.results = {v.name: {"success": 0, "failure": 0, "latency": [], "scores": []}
                       for v in variants}
        assert abs(sum(v.weight for v in variants) - 1.0) < 0.01, "Weights must sum to 1.0"

    def select_variant(self, user_id: str = None) -> AgentVariant:
        """Sticky assignment: same user always gets same variant."""
        if user_id:
            # Hash user_id for consistent assignment
            seed = int(hashlib.md5(user_id.encode()).hexdigest(), 16) % 100
            cumulative = 0
            for variant in self.variants:
                cumulative += variant.weight * 100
                if seed < cumulative:
                    return variant

        # Random assignment
        return random.choices(self.variants, weights=[v.weight for v in self.variants])[0]

    def run(self, task: str, user_id: str = None) -> tuple[str, str]:
        variant = self.select_variant(user_id)

        start = time.time()
        try:
            result = variant.agent(task)
            latency = time.time() - start

            self.results[variant.name]["success"] += 1
            self.results[variant.name]["latency"].append(latency)

            return result, variant.name
        except Exception as e:
            self.results[variant.name]["failure"] += 1
            raise

    def get_stats(self) -> dict:
        stats = {}
        for name, data in self.results.items():
            total = data["success"] + data["failure"]
            stats[name] = {
                "success_rate": data["success"] / total if total > 0 else 0,
                "avg_latency": sum(data["latency"]) / len(data["latency"]) if data["latency"] else 0,
                "avg_score": sum(data["scores"]) / len(data["scores"]) if data["scores"] else 0,
                "total_runs": total
            }
        return stats

# Usage
ab_test = AgentABTest([
    AgentVariant("gpt-4o-baseline", create_agent("gpt-4o", PROMPT_V1), weight=0.5),
    AgentVariant("gpt-4o-new-prompt", create_agent("gpt-4o", PROMPT_V2), weight=0.3),
    AgentVariant("gpt-4o-mini-fast", create_agent("gpt-4o-mini", PROMPT_V1), weight=0.2),
])

result, used_variant = ab_test.run("Analyze the deployment logs", user_id="user-123")
print(ab_test.get_stats())

Q38: How do you implement agent versioning and CI/CD for AI Agents?

Answer: AI agents need version control for prompts, models, and tools — just like software code.

# --- Prompt versioning with git-like tracking ---
from datetime import datetime
import json

class PromptRegistry:
    def __init__(self, storage_path: str = "./prompt_registry.json"):
        self.storage_path = storage_path
        self.registry = self._load()

    def register(self, name: str, prompt: str, metadata: dict = None) -> str:
        version = f"v{len(self.registry.get(name, [])) + 1}"

        if name not in self.registry:
            self.registry[name] = []

        self.registry[name].append({
            "version": version,
            "prompt": prompt,
            "created_at": datetime.now().isoformat(),
            "metadata": metadata or {},
            "active": False
        })
        self._save()
        return version

    def promote(self, name: str, version: str):
        """Set a version as the active production version."""
        for entry in self.registry.get(name, []):
            entry["active"] = (entry["version"] == version)
        self._save()

    def get_active(self, name: str) -> str:
        for entry in self.registry.get(name, []):
            if entry["active"]:
                return entry["prompt"]
        raise ValueError(f"No active prompt for {name}")

name: AI Agent CI/CD

on:
  push:
    branches: [main]
    paths:
      - 'agents/**'
      - 'prompts/**'
      - 'tools/**'

jobs:
  test-agents:
    runs-on: ubuntu-latest
    steps:
    - uses: actions/checkout@v4

    - name: Run agent unit tests
      run: pytest tests/test_tools.py -v

    - name: Run agent integration tests (with mocked LLM)
      run: pytest tests/test_agent_routing.py -v

    - name: Evaluate prompt quality
      env:
        OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
        LANGSMITH_API_KEY: ${{ secrets.LANGSMITH_API_KEY }}
      run: |
        python scripts/evaluate_prompts.py \
          --prompt-file prompts/devops_agent_v2.txt \
          --test-cases tests/eval_cases.json \
          --min-score 0.85

    - name: Compare against baseline
      run: |
        python scripts/compare_agents.py \
          --baseline-version v1.2 \
          --new-version v1.3 \
          --test-cases tests/regression.json

  deploy-agent:
    needs: test-agents
    runs-on: ubuntu-latest
    environment: production
    steps:
    - name: Deploy agent service
      run: |
        docker build -t agent-service:${{ github.sha }} .
        docker push registry.example.com/agent-service:${{ github.sha }}
        kubectl set image deployment/agent-service \
          agent=registry.example.com/agent-service:${{ github.sha }}

Interview tip: Treat prompts as code — version them, test them, review changes in PRs. A single prompt change can degrade agent quality more than a code bug. Your CI pipeline should automatically evaluate prompt quality against a test set before deploying.

Q39: What are the key differences between major LLM providers for agents?

Answer:

Feature	OpenAI (GPT-4o)	Anthropic (Claude 3.5)	Google (Gemini 1.5)
Context window	128K	200K	1M
Tool calling	Excellent	Excellent	Good
Code generation	Excellent	Excellent	Very Good
Following instructions	Very Good	Excellent	Good
Vision	Yes	Yes	Yes
Pricing (approx)	$$$	$$$	$$
Self-hosted option	No	No	Vertex AI
API stability	High	High	Medium
Function calling	`tool_calls`	`tool_use`	`function_call`

# --- OpenAI ---
from openai import OpenAI
client = OpenAI()
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello"}],
    tools=tools
)

# --- Anthropic Claude ---
import anthropic
claude = anthropic.Anthropic()
response = claude.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    tools=tools,  # Same format as OpenAI!
    messages=[{"role": "user", "content": "Hello"}]
)

# --- Google Gemini ---
import google.generativeai as genai
genai.configure(api_key="YOUR_KEY")
model = genai.GenerativeModel("gemini-1.5-pro")
chat = model.start_chat()
response = chat.send_message("Hello")

# --- Provider-agnostic with LiteLLM ---
import litellm
# Switch providers by just changing model name!
response = litellm.completion(
    model="gpt-4o",          # Or "claude-3-5-sonnet", "gemini/gemini-1.5-pro"
    messages=[{"role": "user", "content": "Hello"}]
)

Interview tip: Use LiteLLM to build provider-agnostic agents. You can switch from GPT-4o to Claude to Gemini by changing one string. This gives you fallback capability (if OpenAI is down, switch to Claude), cost optimization (use cheaper models for simple tasks), and freedom from vendor lock-in.

Q40: How do you implement self-improving agents?

Answer: Self-improving agents learn from their mistakes and user feedback to get better over time without manual prompt engineering.

from langsmith import Client
from langsmith.evaluation import evaluate

client_ls = Client()

# --- 1. Collect feedback on agent outputs ---
class FeedbackCollector:
    def __init__(self):
        self.langsmith = Client()

    def record_thumbs(self, run_id: str, score: int, comment: str = ""):
        """Record user thumbs up/down."""
        self.langsmith.create_feedback(
            run_id=run_id,
            key="user_rating",
            score=score,          # 1 = thumbs up, 0 = thumbs down
            comment=comment
        )

    def get_low_quality_runs(self, min_score: float = 0.5) -> list:
        """Find runs where agent performed poorly."""
        runs = self.langsmith.list_runs(
            project_name="my-agent",
            filter=f"feedback_key = 'user_rating' and feedback_score < {min_score}"
        )
        return list(runs)

# --- 2. Automatic prompt optimization ---
def optimize_prompt_from_failures(failed_runs: list, current_prompt: str) -> str:
    """Use LLM to improve prompt based on failure cases."""
    failure_examples = "\n".join([
        f"Input: {run.inputs['input']}\nBad Output: {run.outputs['output']}\nFeedback: {run.feedback_stats}"
        for run in failed_runs[:10]
    ])

    improved_prompt = llm.invoke(f"""
    Current system prompt:
    {current_prompt}

    These cases failed (user gave negative feedback):
    {failure_examples}

    Analyze what went wrong and rewrite the system prompt to handle these cases better.
    Return ONLY the improved system prompt, nothing else.
    """)

    return improved_prompt.content

# --- 3. Few-shot learning from good examples ---
class FewShotMemory:
    """Store good examples and include them in future prompts."""

    def __init__(self, vectorstore):
        self.vectorstore = vectorstore

    def save_good_example(self, question: str, answer: str, score: float):
        if score > 0.8:  # Only save high-quality examples
            self.vectorstore.add_texts(
                texts=[f"Question: {question}\nAnswer: {answer}"],
                metadatas=[{"score": score, "type": "example"}]
            )

    def get_relevant_examples(self, question: str, k: int = 3) -> str:
        docs = self.vectorstore.similarity_search(
            question, k=k,
            filter={"type": "example"}
        )
        return "\n\n".join([d.page_content for d in docs])

    def build_prompt_with_examples(self, question: str, system_prompt: str) -> str:
        examples = self.get_relevant_examples(question)
        return f"{system_prompt}\n\nHere are similar successful examples:\n{examples}"

Q41: How do you implement rate limiting and quota management for agents?

Answer:

import time
import asyncio
from collections import defaultdict
from dataclasses import dataclass, field

@dataclass
class RateLimiter:
    requests_per_minute: int = 60
    tokens_per_minute: int = 100000
    _request_times: list = field(default_factory=list)
    _token_counts: list = field(default_factory=list)

    def can_make_request(self, estimated_tokens: int) -> tuple[bool, float]:
        """Check if we can make a request. Returns (allowed, wait_seconds)."""
        now = time.time()
        minute_ago = now - 60

        # Clean old entries
        self._request_times = [t for t in self._request_times if t > minute_ago]
        self._token_counts = [(t, c) for t, c in self._token_counts if t > minute_ago]

        recent_requests = len(self._request_times)
        recent_tokens = sum(c for _, c in self._token_counts)

        if recent_requests >= self.requests_per_minute:
            wait = 60 - (now - self._request_times[0])
            return False, max(0, wait)

        if recent_tokens + estimated_tokens > self.tokens_per_minute:
            wait = 60 - (now - self._token_counts[0][0])
            return False, max(0, wait)

        return True, 0

    def record_request(self, tokens_used: int):
        now = time.time()
        self._request_times.append(now)
        self._token_counts.append((now, tokens_used))

# Per-user quota management
class QuotaManager:
    def __init__(self):
        self.daily_limits = {"free": 100, "pro": 1000, "enterprise": 10000}
        self.usage = defaultdict(lambda: defaultdict(int))

    def check_quota(self, user_id: str, plan: str) -> tuple[bool, int]:
        today = time.strftime("%Y-%m-%d")
        used = self.usage[user_id][today]
        limit = self.daily_limits.get(plan, 100)
        return used < limit, limit - used

    def record_usage(self, user_id: str, tokens: int):
        today = time.strftime("%Y-%m-%d")
        self.usage[user_id][today] += tokens

# Retry with exponential backoff for rate limit errors
async def call_llm_with_retry(messages: list, max_retries: int = 5):
    for attempt in range(max_retries):
        try:
            return await async_client.chat.completions.create(
                model="gpt-4o", messages=messages
            )
        except Exception as e:
            if "rate_limit" in str(e).lower():
                wait = (2 ** attempt) + random.random()  # Exponential backoff
                await asyncio.sleep(wait)
            else:
                raise
    raise Exception("Max retries exceeded")

Q42: What is DSPy and how does it differ from prompt engineering?

Answer: DSPy (Declarative Self-improving Python) replaces hand-written prompts with automatic optimization — you declare WHAT you want, and DSPy figures out HOW to prompt the model.

import dspy

# Configure LLM
lm = dspy.LM("openai/gpt-4o")
dspy.configure(lm=lm)

# --- DSPy Signatures (declare input/output, not the prompt!) ---
class DevOpsQA(dspy.Signature):
    """Answer DevOps questions with technical accuracy."""
    question: str = dspy.InputField(desc="A DevOps or infrastructure question")
    answer: str = dspy.OutputField(desc="Detailed technical answer with examples")

class AnalyzeLogs(dspy.Signature):
    """Analyze application logs and identify the root cause."""
    logs: str = dspy.InputField(desc="Raw application or system logs")
    error_type: str = dspy.OutputField(desc="Type of error detected")
    root_cause: str = dspy.OutputField(desc="Most likely root cause")
    fix: str = dspy.OutputField(desc="Recommended fix")

# --- Modules (how to process the signature) ---
class ChainOfThoughtQA(dspy.Module):
    def __init__(self):
        self.qa = dspy.ChainOfThought(DevOpsQA)   # Auto-adds reasoning steps

    def forward(self, question: str) -> str:
        return self.qa(question=question).answer

# --- Compile (auto-optimize prompts using examples) ---
training_data = [
    dspy.Example(
        question="What is a Kubernetes Pod?",
        answer="A Pod is the smallest deployable unit in K8s..."
    ).with_inputs("question"),
]

teleprompter = dspy.BootstrapFewShot(metric=lambda x, y, _: 1.0)
optimized_qa = teleprompter.compile(ChainOfThoughtQA(), trainset=training_data)

# Use optimized module
result = optimized_qa(question="How do I debug a CrashLoopBackOff?")
print(result)

Interview tip: DSPy is gaining popularity because it treats prompts as optimizable parameters, not static strings. Instead of spending hours prompt-engineering, you give DSPy examples and a metric, and it finds the best prompts automatically. It’s particularly useful when you have labeled training data.

Q43: How do you implement agent monitoring and alerting in production?

Answer:

from prometheus_client import Counter, Histogram, Gauge, start_http_server
import time

# --- Prometheus metrics ---
agent_requests_total = Counter(
    "agent_requests_total",
    "Total agent requests",
    ["status", "agent_version", "environment"]
)

agent_latency_seconds = Histogram(
    "agent_latency_seconds",
    "Agent response latency",
    ["agent_version"],
    buckets=[0.5, 1.0, 2.0, 5.0, 10.0, 30.0]
)

agent_token_usage = Counter(
    "agent_token_usage_total",
    "Total tokens consumed",
    ["model", "type"]  # type: input|output
)

active_agent_sessions = Gauge(
    "active_agent_sessions",
    "Currently running agent sessions"
)

# --- Instrumented agent wrapper ---
class MonitoredAgent:
    def __init__(self, agent, version: str = "v1.0"):
        self.agent = agent
        self.version = version
        start_http_server(8080)  # Prometheus scrape endpoint

    def run(self, task: str, environment: str = "production") -> str:
        active_agent_sessions.inc()
        start_time = time.time()

        try:
            result = self.agent.invoke({"input": task})

            agent_requests_total.labels(
                status="success",
                agent_version=self.version,
                environment=environment
            ).inc()

            return result["output"]

        except Exception as e:
            agent_requests_total.labels(
                status="error",
                agent_version=self.version,
                environment=environment
            ).inc()
            raise

        finally:
            latency = time.time() - start_time
            agent_latency_seconds.labels(agent_version=self.version).observe(latency)
            active_agent_sessions.dec()

# Grafana alert rules (YAML):
"""
groups:
- name: agent_alerts
  rules:
  - alert: AgentHighErrorRate
    expr: rate(agent_requests_total{status="error"}[5m]) > 0.1
    for: 2m
    annotations:
      summary: "Agent error rate > 10%"

  - alert: AgentHighLatency
    expr: histogram_quantile(0.95, agent_latency_seconds) > 10
    for: 5m
    annotations:
      summary: "Agent P95 latency > 10 seconds"

  - alert: AgentHighTokenCost
    expr: rate(agent_token_usage_total[1h]) * 0.01 > 100
    for: 10m
    annotations:
      summary: "Agent spending >$100/hour on tokens"
"""

Q44: What is the difference between synchronous and asynchronous agent execution?

Answer:

import asyncio
from langchain_openai import ChatOpenAI

# --- Synchronous (blocking) ---
def sync_agent_run(tasks: list[str]) -> list[str]:
    """Runs tasks one by one. Slow for multiple tasks."""
    results = []
    for task in tasks:
        result = agent.invoke({"input": task})  # Blocks until complete
        results.append(result["output"])
    return results  # Total time = sum of all task times

# --- Asynchronous (non-blocking) ---
async def async_agent_run(tasks: list[str]) -> list[str]:
    """Runs all tasks concurrently. Much faster."""
    llm = ChatOpenAI(model="gpt-4o")

    async def process_one(task: str) -> str:
        result = await agent.ainvoke({"input": task})  # Non-blocking
        return result["output"]

    # All tasks start immediately, run in parallel
    results = await asyncio.gather(*[process_one(t) for t in tasks])
    return results  # Total time ≈ longest single task time

# Performance comparison:
tasks = ["Task 1", "Task 2", "Task 3", "Task 4", "Task 5"]
# Sync: 5 × 3s = 15s total
# Async: max(3s, 3s, 3s, 3s, 3s) = 3s total  → 5x faster!

# --- Async with rate limiting ---
from asyncio import Semaphore

async def rate_limited_batch(tasks: list[str], max_concurrent: int = 5) -> list[str]:
    """Process many tasks with concurrency limit to avoid rate limits."""
    semaphore = Semaphore(max_concurrent)

    async def process_with_limit(task: str) -> str:
        async with semaphore:  # Max 5 concurrent at any time
            return await async_process(task)

    return await asyncio.gather(*[process_with_limit(t) for t in tasks])

asyncio.run(rate_limited_batch(tasks=["task"]*100, max_concurrent=5))

Interview tip: Always use async for agent endpoints that handle multiple users or batch processing. Synchronous agents block the entire thread — one slow LLM call starves all other requests. Async allows one server to handle hundreds of concurrent agent sessions efficiently.

Q45: How do you implement agent chaining — multiple agents in sequence?

Answer:

from langchain_core.runnables import RunnableSequence, RunnableLambda
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

llm = ChatOpenAI(model="gpt-4o")

# --- Method 1: Simple LCEL chain ---
analyze_prompt = ChatPromptTemplate.from_template(
    "Analyze these Kubernetes logs and identify the problem:\n{logs}"
)

fix_prompt = ChatPromptTemplate.from_template(
    "Given this problem: {problem}\nProvide step-by-step fix commands:"
)

document_prompt = ChatPromptTemplate.from_template(
    "Create an incident report for:\nProblem: {problem}\nFix: {fix}"
)

# Chain: logs → analyze → fix → document
pipeline = (
    analyze_prompt | llm | (lambda x: {"problem": x.content})
    | fix_prompt | llm | (lambda x: {"fix": x.content})
    | document_prompt | llm
)

result = pipeline.invoke({"logs": "ERROR: OOMKilled, pod restarted 5 times"})

# --- Method 2: Explicit agent chain with context passing ---
class AgentPipeline:
    def __init__(self):
        self.steps = []

    def add_step(self, name: str, agent_fn, input_key: str = None, output_key: str = None):
        self.steps.append({
            "name": name,
            "fn": agent_fn,
            "input_key": input_key,
            "output_key": output_key or name
        })
        return self  # Fluent interface

    def run(self, initial_input: dict) -> dict:
        context = initial_input.copy()

        for step in self.steps:
            print(f"Running step: {step['name']}")

            # Get input (from context or use whole context)
            if step["input_key"]:
                step_input = context[step["input_key"]]
            else:
                step_input = context

            # Run the agent
            result = step["fn"](step_input)

            # Store output in context
            context[step["output_key"]] = result

        return context

# Usage
pipeline = (
    AgentPipeline()
    .add_step("analysis", analyze_agent.run, input_key="logs", output_key="analysis")
    .add_step("fix", fix_agent.run, input_key="analysis", output_key="fix_plan")
    .add_step("validation", validate_agent.run, input_key="fix_plan", output_key="validated_fix")
    .add_step("ticket", jira_agent.run, output_key="ticket_id")
)

result = pipeline.run({"logs": "ERROR: CrashLoopBackOff in pod nginx-abc"})
print(result["ticket_id"])  # JRA-1234

Q46: What is the OpenAI Swarm framework and agent handoffs?

Answer: Swarm (now called OpenAI Agents SDK) is a lightweight framework for orchestrating multiple specialized agents with clean handoff mechanisms.

from openai import OpenAI

client = OpenAI()

# Define specialized agents
def triage_agent(context: dict) -> dict:
    """First point of contact — routes to specialist."""
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": """You are a DevOps triage agent.
            Classify the issue as: kubernetes | docker | cicd | aws | general
            Respond with ONLY the category."""},
            {"role": "user", "content": context["issue"]}
        ]
    )
    category = response.choices[0].message.content.strip()

    # Handoff to specialist
    return {"handoff_to": category, "context": context}

def kubernetes_specialist(context: dict) -> str:
    """Handles Kubernetes-specific issues."""
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "You are a Kubernetes expert. Provide detailed K8s solutions."},
            {"role": "user", "content": context["issue"]}
        ]
    )
    return response.choices[0].message.content

# With OpenAI Agents SDK (newer approach)
from agents import Agent, handoff, Runner

kubernetes_agent = Agent(
    name="Kubernetes Expert",
    instructions="You solve Kubernetes problems. Provide kubectl commands and YAML fixes.",
    model="gpt-4o",
)

docker_agent = Agent(
    name="Docker Expert",
    instructions="You solve Docker and container problems.",
    model="gpt-4o",
)

triage_agent = Agent(
    name="Triage Agent",
    instructions="""You are a DevOps triage agent.
    Route Kubernetes issues to the Kubernetes Expert.
    Route Docker issues to the Docker Expert.""",
    model="gpt-4o",
    handoffs=[handoff(kubernetes_agent), handoff(docker_agent)]
)

result = Runner.run_sync(
    triage_agent,
    "My pod keeps getting OOMKilled. Memory usage spikes every hour."
)
print(result.final_output)

Interview tip: Agent handoffs are the key to scalable multi-agent systems. Instead of one giant agent that knows everything, you build specialized experts and a triage agent that routes to them. Each expert can have different tools, prompts, and even different models (cheap model for triage, powerful model for complex fixes).

Q47: How do you implement long-running background agent tasks?

Answer:

from celery import Celery
from fastapi import FastAPI, BackgroundTasks
import asyncio
import uuid

app = FastAPI()
celery = Celery("agent_tasks", broker="redis://localhost:6379/0")

# Task status store
task_status = {}

# --- Method 1: FastAPI Background Tasks (simple, same process) ---
@app.post("/agent/run-background")
async def run_agent_background(request: dict, background_tasks: BackgroundTasks):
    task_id = str(uuid.uuid4())
    task_status[task_id] = {"status": "pending", "result": None}

    async def run():
        try:
            task_status[task_id]["status"] = "running"
            result = await agent.ainvoke({"input": request["task"]})
            task_status[task_id] = {"status": "done", "result": result["output"]}
        except Exception as e:
            task_status[task_id] = {"status": "error", "error": str(e)}

    background_tasks.add_task(run)
    return {"task_id": task_id}

@app.get("/agent/status/{task_id}")
def get_task_status(task_id: str):
    return task_status.get(task_id, {"status": "not_found"})

# --- Method 2: Celery (distributed, survives restarts) ---
@celery.task(bind=True, max_retries=3)
def run_agent_task(self, task: str, user_id: str):
    """Long-running agent task, retryable."""
    try:
        result = agent.invoke({"input": task})
        return {"status": "done", "result": result["output"]}
    except Exception as exc:
        # Retry with exponential backoff
        raise self.retry(exc=exc, countdown=2 ** self.request.retries)

# Submit task
task = run_agent_task.delay("Analyze 10,000 log files", "user-123")
print(task.id)  # Task ID to poll later

# Poll status
from celery.result import AsyncResult
result = AsyncResult(task.id)
print(result.state)   # PENDING | STARTED | SUCCESS | FAILURE | RETRY
print(result.get())   # Block until done (or timeout=60)

Q48: What are the emerging trends in AI Agents for 2025?

Answer:

🔥 Top AI Agent Trends 2025:

1. MCP (Model Context Protocol)
   → Standard for tool integration
   → Write one MCP server, works with all agents

2. Computer Use / Browser Agents
   → Claude Computer Use, GPT-4o browser tools
   → Agents that control real browsers/computers

3. Voice Agents
   → Real-time voice with <500ms latency
   → OpenAI Realtime API, ElevenLabs conversational AI

4. Reasoning Models in Agents
   → o3, Claude 3.7 (extended thinking)
   → Multi-step reasoning before acting

5. Multi-Agent Standards
   → Google A2A (Agent-to-Agent) protocol
   → Agents communicating with agents

6. Edge/Local Agents
   → Ollama, LM Studio for local model inference
   → Privacy-preserving, offline-capable agents

7. Agentic RAG
   → Agents that dynamically query multiple sources
   → Self-evaluating retrieval quality

8. Long-horizon task completion
   → Agents that work for hours/days on complex tasks
   → Devin, SWE-agent, AutoCodeRover

# Computer Use (Claude can control a real browser)
import anthropic

client = anthropic.Anthropic()

response = client.beta.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=4096,
    tools=[{"type": "computer_20241022", "name": "computer", "display_width_px": 1024, "display_height_px": 768}],
    messages=[{"role": "user", "content": "Go to grafana.example.com and take a screenshot of the error rate dashboard"}]
)

# Real-time Voice Agent
from openai import OpenAI
import websocket

# Connect to OpenAI Realtime API
ws = websocket.create_connection("wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview")
# Agent speaks and listens in real-time with <200ms latency

Interview tip: For 2025 interviews, mention MCP as the emerging standard for tool integration, and note that agents are moving from text-only to truly multi-modal (voice, vision, browser control). The shift from single-agent to agent swarms with standardized communication protocols (A2A, MCP) is the biggest architectural trend.

Q49: How do you migrate from a rules-based system to an AI Agent?

Answer:

Migration Strategy: Rules → Hybrid → Full Agent

PHASE 1: Identify automation candidates
- Rules with many exceptions → Good for AI
- Rules that require "judgment" → Good for AI
- Simple, exact-match rules → Keep as code
- Regulatory/compliance rules → Keep as code (auditable)

PHASE 2: Hybrid approach (safest)
- Keep existing rules engine
- Add AI for exception handling
- AI routes ambiguous cases

PHASE 3: Full agent (gradual rollout)
- Start with low-risk use cases
- A/B test agent vs rules
- Roll out to 5% → 25% → 100% traffic

class HybridAutomation:
    """Combines deterministic rules with AI agent for edge cases."""

    def __init__(self, rules_engine, ai_agent):
        self.rules = rules_engine
        self.agent = ai_agent
        self.ai_usage_counter = 0

    def process(self, request: dict) -> dict:
        # 1. Try rules engine first (fast, deterministic, auditable)
        rule_result = self.rules.evaluate(request)

        if rule_result.confidence == "high":
            # Rule matched clearly → use it
            return {"result": rule_result.action, "method": "rules", "confidence": "high"}

        elif rule_result.confidence == "medium":
            # Ambiguous → consult AI but log for review
            ai_result = self.agent.run(str(request))
            self.ai_usage_counter += 1

            return {
                "result": ai_result,
                "method": "ai-assisted",
                "rule_suggestion": rule_result.action,
                "needs_review": True  # Flag for human review
            }

        else:
            # No matching rule → full AI
            ai_result = self.agent.run(str(request))
            return {"result": ai_result, "method": "ai-full"}

    def retrain_rules(self):
        """Use AI decisions to generate new rules automatically."""
        ai_decisions = get_recent_ai_decisions()
        patterns = extract_patterns(ai_decisions)  # Cluster similar decisions
        new_rules = generate_rules_from_patterns(patterns)
        self.rules.add(new_rules)

Q50: What is Agentic AI vs Traditional AI — and what’s next?

Answer:

Evolution of AI Systems:

Level 1: Traditional AI (2010s)
  → Fixed rules, ML models, narrow tasks
  → "Is this email spam?" → yes/no

Level 2: LLM-powered apps (2022-2023)
  → Chatbots, question answering, code generation
  → Single prompt → single response

Level 3: AI Agents (2023-2024)
  → Multi-step reasoning, tool use, autonomy
  → "Research this topic and write a report"

Level 4: Agentic AI / Agent Swarms (2024-2025)
  → Multiple agents collaborating, long-horizon tasks
  → "Build, test, and deploy this feature"

Level 5: Autonomous AI Systems (Future)
  → Self-improving, self-directing
  → AI that creates and manages other AIs

Practical Summary for Interviews:

Question	Answer
Best agent framework?	LangGraph for complex workflows, LangChain for simple agents
Best model for agents?	GPT-4o or Claude 3.5 Sonnet (both excellent for tool use)
RAG vs Fine-tuning?	RAG for dynamic knowledge, Fine-tuning for behavior/style
How to test agents?	Unit test tools + LLM-as-judge + regression suite
How to reduce cost?	Semantic caching + smaller models for simple tasks + tool result caching
Production monitoring?	LangSmith traces + Prometheus metrics + structured logging
Security?	Least-privilege tools + output sanitization + injection detection
Scaling?	Async execution + Celery for background tasks + rate limiting

# The 10-line agent that impresses interviewers
from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_community.tools import DuckDuckGoSearchRun

llm = ChatOpenAI(model="gpt-4o", temperature=0)
tools = [DuckDuckGoSearchRun()]
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful DevOps assistant. Use tools to give accurate, current answers."),
    ("human", "{input}"),
    MessagesPlaceholder("agent_scratchpad")
])
agent = AgentExecutor(agent=create_tool_calling_agent(llm, tools, prompt), tools=tools, verbose=True)
print(agent.invoke({"input": "What is the latest Kubernetes version?"})["output"])

Interview tip: The best answer to “Where is AI Agents heading?” is: from single-agent (one LLM doing everything) to multi-agent swarms (specialized agents collaborating), from text-only to truly multi-modal (vision, voice, browser control), and from stateless to persistent agents that remember context across weeks/months of work.

Back to DevOps Q&A Index

AI Agents Interview Questions

AI Agents Interview Questions & Answers (1–40)

Section 1: AI Agents Fundamentals

Q1: What is an AI Agent and how does it differ from a regular LLM call?

Q2: What are the core components of an AI Agent?

Q3: What is RAG (Retrieval-Augmented Generation)?

Q4: What are tool/function calling in LLMs?

Section 2: LangChain & Agent Frameworks

Q5: What is LangChain and how do you build an agent with it?

Q6: What is LangGraph and when should you use it over LangChain agents?

Q7: What is AutoGen and how does it enable multi-agent systems?

Section 3: Agent Memory & State

Q8: How do you implement memory in AI Agents?

Q9: What are vector embeddings and how do they work?

Q10: How do you implement a vector database for an agent?

Section 4: Prompt Engineering

Q11: What are key prompt engineering techniques?

Q12: How do you prevent prompt injection in AI Agents?

Section 5: Agent Orchestration & Production

Q13: How do you deploy AI Agents in production?

Q14: How do you evaluate AI Agent performance?

Q15: How does agent token management and cost optimization work?

Section 6: Advanced AI Agent Concepts

Q16: What is the difference between ReAct, CoT, and Plan-and-Execute?

Q17: How do you build a multi-agent pipeline for DevOps automation?

Q18: What are Agentic patterns you should know?

Q19: How do you handle agent failures and retries?

Q20: What are the key safety considerations for AI Agents in production?

Section 7: Advanced Topics & Production Patterns

Q21: What is the Model Context Protocol (MCP)?

Q22: What is the OpenAI Assistants API and how does it differ from Chat Completions?

Q23: What is CrewAI and how do you build multi-agent teams?

Q24: What is Microsoft Semantic Kernel and when would you use it?

Q25: What is the difference between fine-tuning and RAG? When to use each?

Q26: How do you implement streaming responses in AI Agents?

Q27: How do you implement guardrails and output validation for AI Agents?

Q28: How do you implement agent observability and tracing?

Q29: How do you implement multi-modal agents (images, audio, video)?

Q30: How do you implement agent workflows with human-in-the-loop approval?

Q31: How do you test AI Agents?

Q32: How do you implement caching for AI Agents to reduce cost?

Q33: What are knowledge graphs and how do they enhance AI Agents?

Q34: How do you handle context window limits in long-running agents?

Q35: How do you implement agent security — preventing data exfiltration?

Q36: What is Retrieval-Augmented Generation vs. Agentic RAG?

Q37: How do you implement A/B testing for AI Agents?

Q38: How do you implement agent versioning and CI/CD for AI Agents?

Q39: What are the key differences between major LLM providers for agents?

Q40: How do you implement self-improving agents?

Q41: How do you implement rate limiting and quota management for agents?

Q42: What is DSPy and how does it differ from prompt engineering?

Q43: How do you implement agent monitoring and alerting in production?

Q44: What is the difference between synchronous and asynchronous agent execution?

Q45: How do you implement agent chaining — multiple agents in sequence?

Q46: What is the OpenAI Swarm framework and agent handoffs?

Q47: How do you implement long-running background agent tasks?

Q48: What are the emerging trends in AI Agents for 2025?

Q49: How do you migrate from a rules-based system to an AI Agent?

Q50: What is Agentic AI vs Traditional AI — and what’s next?