Building Production-Ready AI Agents in 2026: A Hands-On Implementation Guide

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage

# 1. Define state schema
from typing import TypedDict, List, Annotated
import operator

class AgentState(TypedDict):
    messages: Annotated[List, operator.add]
    research_topic: str
    needs_review: bool
    approved_by: str | None

# 2. Initialize with checkpointing (critical for production)
memory = MemorySaver()  # In production: use PostgresSaver or RedisSaver

# 3. Create graph
workflow = StateGraph(AgentState)

# 4. Define nodes
def researcher_node(state: AgentState):
    llm = ChatOpenAI(model="gpt-4o", temperature=0.1)
    system_prompt = """You are a research analyst. Provide factual, cited information."""
    response = llm.invoke([
        SystemMessage(content=system_prompt),
        HumanMessage(content=state["research_topic"])
    ])
    return {
        "messages": [response],
        "needs_review": True  # Flag for human review
    }

def reviewer_node(state: AgentState):
    """Human-in-the-loop approval node"""
    # In production: this would trigger a UI notification
    print(f"Research output requires review: {state['messages'][-1].content[:100]}...")
    return {"approved_by": "human_operator"}  # Simulated approval

# 5. Build graph with conditional edges
workflow.add_node("researcher", researcher_node)
workflow.add_node("reviewer", reviewer_node)

workflow.set_entry_point("researcher")
workflow.add_conditional_edges(
    "researcher",
    lambda state: "reviewer" if state.get("needs_review") else END
)
workflow.add_edge("reviewer", END)

# 6. Compile with persistence
agent = workflow.compile(checkpointer=memory)

# 7. Execute with thread ID for checkpointing
config = {"configurable": {"thread_id": "research-123"}}
result = agent.invoke(
    {"research_topic": "Impact of quantum computing on cryptography", "messages": []},
    config=config
)
print(f"Final output: {result['messages'][-1].content}")

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

# 1. Define agents with clear roles and backstory
researcher = Agent(
    role='Senior Research Analyst',
    goal='Uncover cutting-edge developments in AI agents',
    backstory="""You're a seasoned research analyst with 10 years experience in enterprise AI.
    You specialize in identifying practical patterns and anti-patterns.""",
    verbose=True,
    tools=[SerperDevTool(), ScrapeWebsiteTool()],
    llm=ChatOpenAI(model="gpt-4o")
)

writer = Agent(
    role='Technical Content Strategist',
    goal='Craft compelling content on AI agents for engineers',
    backstory="""You're a renowned technical writer who translates complex concepts
    into actionable insights. You write for DevOps and backend developers.""",
    verbose=True,
    llm=ChatOpenAI(model="gpt-4o")
)

editor = Agent(
    role='Lead Editor',
    goal='Ensure content quality, accuracy, and readability',
    backstory="""You're a meticulous editor with a sharp eye for technical accuracy
    and clarity. You enforce style guidelines and catch hallucinations.""",
    verbose=True,
    llm=ChatOpenAI(model="gpt-4o")
)

# 2. Define tasks with explicit outputs and context
research_task = Task(
    description="""Research the latest production patterns for AI agents in 2026.
    Focus on: deployment strategies, cost optimization, security patterns.
    Include specific tool names and configuration examples.""",
    expected_output="""A structured research report with:
    1. Framework comparison (LangGraph vs CrewAI vs Microsoft)
    2. 3-5 concrete deployment patterns
    3. Cost optimization techniques with numbers""",
    agent=researcher,
    output_file='research_report.md'  # Persists to disk
)

write_task = Task(
    description="""Using the research report, write a comprehensive guide
    for engineers building production AI agents. Include code examples.""",
    expected_output="""A 2000-word technical article with:
    - Clear sections and code blocks
    - Python examples for each framework
    - Production tips and warnings""",
    agent=writer,
    context=[research_task]  # Explicit dependency
)

edit_task = Task(
    description="""Review the article for:
    - Technical accuracy (verify all code examples)
    - Hallucinations and unsupported claims
    - Readability for target audience (DevOps/backend)""",
    expected_output="""Edited article with:
    - Track changes and comments
    - Final approval or revision requests""",
    agent=editor,
    context=[write_task]
)

# 3. Assemble crew with hierarchical process
crew = Crew(
    agents=[researcher, writer, editor],
    tasks=[research_task, write_task, edit_task],
    process=Process.hierarchical,  # Manager-style delegation
    manager_llm=ChatOpenAI(model="gpt-4o"),
    verbose=2
)

# 4. Execute
result = crew.kickoff()
print(f"Article delivered: {result}")

using Microsoft.SemanticKernel;
using Microsoft.SemanticKernel.Agents;
using Microsoft.SemanticKernel.Agents.Chat;

// 1. Configure kernel with Azure OpenAI
var builder = Kernel.CreateBuilder();
builder.AddAzureOpenAIChatCompletion(
    deploymentName: "gpt-4o",
    endpoint: Environment.GetEnvironmentVariable("AZURE_OPENAI_ENDPOINT"),
    apiKey: Environment.GetEnvironmentVariable("AZURE_OPENAI_API_KEY")
);
var kernel = builder.Build();

// 2. Create agent with Azure AI Studio plugin
var agent = new ChatCompletionAgent()
{
    Name = "SupportAgent",
    Instructions = "You are a support agent. Use the knowledge base to answer accurately.",
    Kernel = kernel,
    // Azure AI Studio plugin with RBAC
    Plugins = { new AzureAIPlugin("support-knowledge-base") }
};

// 3. Create chat with thread persistence
var chat = new ChatHistory();
var threadId = await agent.CreateThreadAsync(chat);

// 4. Add user message with context
chat.AddUserMessage("How do I reset my MFA token?");
var response = await agent.InvokeAsync(chat, threadId);

// 5. Production: Add monitoring and safety
var monitoredAgent = agent
    .WithLogging(new AzureMonitorLogger())  // Azure Monitor integration
    .WithPromptFiltering(new AzureContentSafetyFilter())  // Content safety
    .WithApprovalPolicy(new HumanInTheLoopApprovalPolicy());  // Approval gate

Console.WriteLine($"Response: {response}");

# Multi-stage build for minimal attack surface
FROM python:3.11-slim as builder

WORKDIR /app
COPY requirements.txt .
RUN pip install --user --no-cache-dir -r requirements.txt

# Runtime stage
FROM python:3.11-slim

# Create non-root user
RUN adduser --disabled-password --gecos '' appuser
USER appuser

WORKDIR /app

# Copy only installed packages from builder
COPY --from=builder /home/appuser/.local /home/appuser/.local
COPY . .

# Ensure Python finds user-installed packages
ENV PATH=/home/appuser/.local/bin:$PATH

# Security: Read-only filesystem, no root
RUN chmod -R 755 /app && \
    find /app -type f -exec chmod 644 {} \; && \
    find /app -type d -exec chmod 755 {} \;

EXPOSE 8000

# Use gunicorn with workers based on CPU cores
CMD ["gunicorn", "--bind", "0.0.0.0:8000", "--workers", "4", "agent_service:app"]

apiVersion: apps/v1
kind: Deployment
metadata:
  name: ai-agent-service
  namespace: production
spec:
  replicas: 3
  strategy:
    type: RollingUpdate
    rollingUpdate:
      maxSurge: 1
      maxUnavailable: 0  # Zero-downtime deployments
  selector:
    matchLabels:
      app: ai-agent
  template:
    metadata:
      labels:
        app: ai-agent
        tier: backend
    spec:
      containers:
      - name: agent
        image: registry.example.com/ai-agent:v1.2.0
        ports:
        - containerPort: 8000
        env:
        - name: REDIS_URL
          valueFrom:
            secretKeyRef:
              name: agent-secrets
              key: redis-url
        - name: DATABASE_URL
          valueFrom:
            secretKeyRef:
              name: agent-secrets
              key: db-url
        resources:
          requests:
            memory: "2Gi"
            cpu: "1000m"
          limits:
            memory: "4Gi"
            cpu: "2000m"
        readinessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 30
          periodSeconds: 10
        livenessProbe:
          httpGet:
            path: /live
            port: 8000
          initialDelaySeconds: 60
          periodSeconds: 30
        volumeMounts:
        - name: checkpoint-storage
          mountPath: /checkpoints
      volumes:
      - name: checkpoint-storage
        persistentVolumeClaim:
          claimName: agent-checkpoints-pvc
      securityContext:
        runAsNonRoot: true
        runAsUser: 1000
        fsGroup: 1000

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

# Structured output reduces tokens by ~30%
from pydantic import BaseModel, Field
from typing import List

class AgentResponse(BaseModel):
    action: str = Field(description="Next action: 'continue', 'ask_clarification', 'complete'")
    reasoning: str = Field(description="Brief reasoning for the action")
    content: str = Field(description="Response content if action is 'continue'")
    questions: List[str] = Field(default=[], description="Clarification questions if action is 'ask_clarification'")

# Use with OpenAI's JSON mode
llm = ChatOpenAI(model="gpt-4o", response_format={"type": "json_object"})
structured_llm = llm.with_structured_output(AgentResponse)

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant. Respond in the specified JSON format."),
    ("human", "{input}")
])

chain = prompt | structured_llm
result = chain.invoke({"input": "What's the weather?"})
print(result.action)  # "ask_clarification" (needs location)

import re
from typing import Tuple

def sanitize_user_input(user_input: str) -> Tuple[bool, str]:
    """Basic prompt injection sanitization"""
    
    # 1. Length limit
    if len(user_input) > 2000:
        return False, "Input too long"
    
    # 2. Detect common injection patterns
    injection_patterns = [
        r"ignore\s+(previous|all|above).*instructions",
        r"system\s*:",
        r"<\s*script",
        r"exec\s*\(",
        r"__import__",
        r"os\.system"
    ]
    
    for pattern in injection_patterns:
        if re.search(pattern, user_input, re.IGNORECASE):
            return False, "Suspicious pattern detected"
    
    # 3. Strip dangerous characters
    sanitized = re.sub(r"[^\w\s.,!?'\-()\[\]]", "", user_input)
    
    return True, sanitized

# Usage in agent
is_safe, clean_input = sanitize_user_input(user_message)
if not is_safe:
    return {"error": "Input rejected", "reason": clean_input}

# Proceed with clean_input

# Producer: API endpoint
from fastapi import FastAPI, BackgroundTasks
import redis
import json
import uuid

app = FastAPI()
r = redis.Redis(host='redis', port=6379, db=0)

@app.post("/agent/run")
async def run_agent(request: AgentRequest, background_tasks: BackgroundTasks):
    # 1. Validate and sanitize
    is_safe, clean_input = sanitize_user_input(request.input)
    if not is_safe:
        raise HTTPException(400, "Invalid input")
    
    # 2. Create job
    job_id = str(uuid.uuid4())
    job = {
        "id": job_id,
        "input": clean_input,
        "user_id": request.user_id,
        "agent_type": request.agent_type,
        "priority": request.priority or "normal"
    }
    
    # 3. Push to queue (Redis Streams)
    r.xadd("agent_jobs", {"data": json.dumps(job)})
    
    # 4. Return job ID for polling
    return {"job_id": job_id, "status": "queued"}

# Consumer: Worker process (separate container)
import asyncio
from langgraph.graph import StateGraph

async def worker():
    while True:
        # 1. Read from stream (blocking)
        messages = r.xread({"agent_jobs": "$"}, count=1, block=5000)
        if not messages:
            continue
            
        for stream, message_list in messages:
            for message_id, message in message_list:
                job = json.loads(message[b"data"])
                
                try:
                    # 2. Execute agent (LangGraph with checkpoint)
                    result = await run_agent_graph(job["input"], job_id=job["id"])
                    
                    # 3. Store result
                    r.set(f"result:{job['id']}", json.dumps({
                        "status": "completed",
                        "output": result,
                        "completed_at": time.time()
                    }))
                    
                    # 4. Notify via WebSocket/push
                    notify_completion(job["user_id"], job["id"])
                    
                except Exception as e:
                    # 5. Error handling with retry logic
                    r.xadd("agent_errors", {"job_id": job["id"], "error": str(e)})
                    if should_retry(e):
                        r.xadd("agent_jobs", {"data": json.dumps(job)})  # Re-queue
                
                # Acknowledge
                r.xdel("agent_jobs", message_id)

if __name__ == "__main__":
    asyncio.run(worker())

# LangSmith integration (works with LangGraph)
from langchain.callbacks import LangSmithTracer

tracer = LangSmithTracer(
    project_name="production-agent",
    tags=["production", "v1.2"]
)

# Pass to any LangChain/LangGraph call
llm = ChatOpenAI(
    callbacks=[tracer],
    tags=["gpt-4o", "research"]
)

# Or for LangGraph
agent = workflow.compile(
    checkpointer=memory,
    callback_manager=tracer.get_callback_manager()
)

# Traces appear in LangSmith UI with:
# - Full graph visualization
# - Token usage per node
# - Latency breakdown
# - Input/output for each step