advanced⏱50 minutesLección 10 de 10

Proyecto Final: Agente de Soporte al Cliente de Producción

Construye un agente de soporte al cliente de nivel de producción con enrutamiento multi-agente, RAG, transferencia a humano y monitoreo — integrando todo del curso Avanzado.

Proyecto Final: Agente de Soporte al Cliente de Producción

This capstone project integrates everything from the Advanced course: hierarchical agents, RAG, SQL agents, interrupts, streaming, LangSmith monitoring, and deployment. The result is a production-ready customer support system.

Arquitectura del Sistema

Customer Query
    ↓
┌─────────────────────────────────────────────┐
│              Router Agent                    │
│  Classifies intent, routes to specialist     │
└────┬─────────┬──────────┬───────────────────┘
     ↓         ↓          ↓
┌─────────┐ ┌────────┐ ┌──────────┐
│  FAQ     │ │  RAG   │ │  SQL     │
│  Agent   │ │ Agent  │ │  Agent   │
└────┬─────┘ └───┬────┘ └────┬─────┘
     ↓          ↓           ↓
┌─────────────────────────────────────────────┐
│              Escalation Check                │
│  → Human agent if confidence < threshold     │
└─────────────────────────────────────────────┘
    ↓
┌─────────────────────────────────────────────┐
│           Quality Review                     │
│  → LangSmith tracing + evaluation            │
└─────────────────────────────────────────────┘
    ↓
    Response

Paso 1: Estado y Configuración

python

from langgraph.graph import StateGraph, START, END, add_messages
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.types import interrupt, Command
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_core.tools import tool
from langchain_core.messages import SystemMessage, ToolMessage, HumanMessage
from langgraph.prebuilt import ToolExecutor
from typing_extensions import TypedDict, Annotated
from typing import List, Any, Optional
from operator import add
from datetime import datetime
import os

# Environment setup
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_PROJECT"] = "customer-support-prod"

# Configuration
class Config:
    # Models
    ROUTER_MODEL = "gpt-4o-mini"
    SPECIALIST_MODEL = "gpt-4o"
    REVIEWER_MODEL = "gpt-4o-mini"

    # RAG
    VECTOR_STORE_PATH = "./customer_kb"
    RETRIEVAL_K = 5

    # Limits
    MAX_RETRIES = 3
    CONFIDENCE_THRESHOLD = 0.7
    MAX_TOKENS = 8000

# LLMs
router_llm = ChatOpenAI(model=Config.ROUTER_MODEL, temperature=0.0)
specialist_llm = ChatOpenAI(model=Config.SPECIALIST_MODEL, temperature=0.3)
reviewer_llm = ChatOpenAI(model=Config.REVIEWER_MODEL, temperature=0.0)

Paso 2: Definición de Estado

python

class CustomerState(TypedDict):
    # Core
    messages: Annotated[List[Any], add_messages]
    customer_id: str
    query: str

    # Routing
    intent: str                      # "faq", "knowledge_base", "data_query", "escalate"
    confidence: float

    # Specialists
    faq_answer: str
    rag_results: List[str]
    sql_result: str

    # Escalation
    needs_human: bool
    human_resolved: bool
    human_notes: str

    # Quality
    review_score: float
    reviewed: bool

    # Operations
    status: str
    retries: int
    errors: Annotated[List[str], add]
    token_usage: Annotated[List[int], add]
    start_time: str

Paso 3: Agente Enrutador

python

def router_agent(state: CustomerState) -> dict:
    prompt = f"""Classify this customer support query:

Query: {state['query']}

Categories:
- 'faq': Simple questions about policies, hours, shipping
- 'knowledge_base': Questions needing detailed product/service info
- 'data_query': Questions requiring looking up customer data
- 'escalate': Angry, complex, or queries the agent cannot handle

Respond with JSON: {{"intent": "category", "confidence": 0.0-1.0}}"""

    response = router_llm.invoke(prompt)
    import json
    try:
        result = json.loads(response.content)
        return {
            "intent": result.get("intent", "faq"),
            "confidence": result.get("confidence", 0.5),
            "status": "routed"
        }
    except (json.JSONDecodeError, KeyError):
        return {"intent": "faq", "confidence": 0.5, "status": "routed"}

Paso 4: Agentes Especialistas

Agente FAQ

python

def faq_agent(state: CustomerState) -> dict:
    prompt = f"""You are a customer support FAQ agent. Answer concisely.

Common FAQs:
- Shipping: Free shipping on orders over $50, 5-7 business days
- Returns: 30-day return policy, items must be unused
- Hours: Monday-Friday 9AM-6PM EST

Query: {state['query']}

Provide a helpful, concise answer based on these policies."""

    answer = specialist_llm.invoke(prompt).content
    return {"faq_answer": answer, "status": "faq_answered"}

Agente RAG (Base de Conocimiento)

python

# Initialize vector store
embeddings = OpenAIEmbeddings()
vectorstore = Chroma(
    collection_name="support_kb",
    embedding_function=embeddings,
    persist_directory=Config.VECTOR_STORE_PATH
)
retriever = vectorstore.as_retriever(search_kwargs={"k": Config.RETRIEVAL_K})

@tool
def search_knowledge_base(query: str) -> str:
    """Search the customer support knowledge base."""
    docs = retriever.invoke(query)
    return "\n\n".join(f"[{d.metadata.get('source', 'KB')}]: {d.page_content}" for d in docs)

rag_tools = [search_knowledge_base]
rag_llm = specialist_llm.bind_tools(rag_tools)
rag_executor = ToolExecutor(rag_tools)

def rag_agent(state: CustomerState) -> dict:
    system = """You are a support specialist with access to the knowledge base.
    Search for relevant information, then provide a detailed answer with citations."""
    messages = [SystemMessage(system), HumanMessage(state["query"])]
    response = rag_llm.invoke(messages)

    if response.tool_calls:
        results = []
        for tc in response.tool_calls:
            result = rag_executor.invoke(tc)
            results.append(str(result))

        # Generate final answer with context
        context = "\n".join(results)
        final = specialist_llm.invoke(
            f"Context:\n{context}\n\nAnswer the query: {state['query']}"
        )
        return {"rag_results": results, "messages": [HumanMessage(state["query"]), response, final],
                "status": "rag_answered"}

    return {"rag_results": [response.content], "status": "rag_answered"}

Agente SQL (Datos del Cliente)

python

from sqlalchemy import create_engine, text

# Read-only database connection
DB_URL = os.getenv("CUSTOMER_DB_URL", "postgresql://readonly:pass@localhost:5432/support")
db_engine = create_engine(DB_URL)

@tool
def query_customer_data(sql: str) -> str:
    """Execute a read-only SQL query for customer data."""
    try:
        with db_engine.connect() as conn:
            result = conn.execute(text(sql))
            rows = result.fetchmany(10)
            return "\n".join(str(row) for row in rows)
    except Exception as e:
        return f"Query error: {e}"

sql_tools = [query_customer_data]
sql_llm = specialist_llm.bind_tools(sql_tools)
sql_executor = ToolExecutor(sql_tools)

def sql_agent(state: CustomerState) -> dict:
    schema_prompt = specialist_llm.invoke(
        "Get the database table names and columns"
    )
    schema = schema_prompt.content

    messages = [
        SystemMessage(f"Database schema:\n{schema}\n\n"
                      "Convert the customer's question to a SQL query. "
                      "Use only SELECT statements. Show results clearly."),
        HumanMessage(state["query"])
    ]
    response = sql_llm.invoke(messages)

    if response.tool_calls:
        for tc in response.tool_calls:
            result = sql_executor.invoke(tc)
            return {"sql_result": str(result), "status": "sql_answered"}

    return {"sql_result": response.content, "status": "sql_answered"}

Paso 5: Escalada y Transferencia a Humano

python

def escalation_check(state: CustomerState) -> dict:
    needs_human = (
        state["confidence"] < Config.CONFIDENCE_THRESHOLD
        or state.get("intent") == "escalate"
        or state.get("retries", 0) >= Config.MAX_RETRIES
    )
    return {"needs_human": needs_human, "status": "checked"}

def human_handoff(state: CustomerState) -> dict:
    # Get context for the human agent
    context = f"""
Customer ID: {state['customer_id']}
Query: {state['query']}
Intent: {state.get('intent', 'unknown')}
Agent Confidence: {state.get('confidence', 0)}
Attempts: {state.get('retries', 0)}
Errors: {state.get('errors', [])}
"""
    response = interrupt({
        "type": "human_handoff",
        "context": context,
        "prompt": "Please handle this escalated ticket."
    })

    return {
        "human_resolved": response.get("resolved", True),
        "human_notes": response.get("notes", ""),
        "status": "human_handled"
    }

[!NOTA] The human handoff uses interrupt() to pause and wait for a human agent to take over. The human's response (resolved + notes) is captured in state.

Paso 6: Revisión de Calidad

python

def quality_review(state: CustomerState) -> dict:
    answer = (state.get("faq_answer") or state.get("sql_result") or
              (state.get("rag_results") or [""])[-1])
    if not answer:
        answer = state["messages"][-1].content if state["messages"] else ""

    prompt = f"""Review this customer support interaction:

Query: {state['query']}
Answer: {answer[:500]}
Confidence: {state.get('confidence', 0)}

Rate 0-10 on:
1. Did the answer address the query?
2. Was it accurate based on available info?
3. Was the tone appropriate?

Return JSON: {{"score": 0-10, "issues": ["issue1"], "suggestions": ["suggestion1"]}}"""

    review = reviewer_llm.invoke(prompt)
    import json
    try:
        result = json.loads(review.content)
        return {"review_score": result.get("score", 5), "reviewed": True, "status": "reviewed"}
    except json.JSONDecodeError:
        return {"review_score": 5, "reviewed": True, "status": "reviewed"}

Paso 7: Lógica del Enrutador

python

def route_intent(state: CustomerState) -> str:
    if state["intent"] == "faq":
        return "faq"
    elif state["intent"] == "knowledge_base":
        return "rag"
    elif state["intent"] == "data_query":
        return "sql"
    else:
        return "escalate"

def after_specialist(state: CustomerState) -> str:
    if state.get("needs_human"):
        return "human"
    return "review"

def after_human(state: CustomerState) -> str:
    return "review" if state.get("human_resolved") else "retry"

Paso 8: Construir el Grafo

python

builder = StateGraph(CustomerState)

# Nodes
builder.add_node("router", router_agent)
builder.add_node("faq_agent", faq_agent)
builder.add_node("rag_agent", rag_agent)
builder.add_node("sql_agent", sql_agent)
builder.add_node("escalation", escalation_check)
builder.add_node("human_handoff", human_handoff)
builder.add_node("quality_review", quality_review)

# Edges
builder.add_edge(START, "router")
builder.add_conditional_edges("router", route_intent, {
    "faq": "faq_agent",
    "rag": "rag_agent",
    "sql": "sql_agent",
    "escalate": "human_handoff"
})

# After specialists, check if escalation needed
builder.add_edge("faq_agent", "escalation")
builder.add_edge("rag_agent", "escalation")
builder.add_edge("sql_agent", "escalation")

# Escalation or review
builder.add_conditional_edges("escalation", after_specialist, {
    "human": "human_handoff",
    "review": "quality_review"
})

# After human
builder.add_conditional_edges("human_handoff", after_human, {
    "review": "quality_review",
    "retry": "router"  # Retry with different approach
})

# Final edge
builder.add_edge("quality_review", END)

# Compile with persistence
checkpointer = PostgresSaver.from_conn_string(
    os.getenv("DATABASE_URL", "postgresql://user:pass@localhost:5432/langgraph")
)
checkpointer.setup()  # One-time initialization

app = builder.compile(checkpointer=checkpointer)

Paso 9: Configuración de Despliegue

json

{
  "name": "customer-support-agent",
  "version": "2.0.0",
  "graphs": {
    "support": "./src/support_agent.py:app"
  },
  "dependencies": {
    "pip": [
      "langchain-openai",
      "langchain-community",
      "chromadb",
      "sqlalchemy",
      "psycopg2-binary",
      "tiktoken"
    ],
    "python": "3.11"
  },
  "env": {
    "OPENAI_API_KEY": "${OPENAI_API_KEY}",
    "DATABASE_URL": "${DATABASE_URL}",
    "CUSTOMER_DB_URL": "${CUSTOMER_DB_URL}"
  },
  "checkpointer": {
    "type": "postgres",
    "url": "${DATABASE_URL}"
  }
}

Paso 10: Configuración de Monitoreo

python

# In your application entry point:
import langsmith

client = langsmith.Client()

# Create monitors
client.create_monitor(
    project_name="customer-support-prod",
    metric="error_rate",
    threshold=0.02,  # Alert if > 2% errors
    interval_minutes=15
)

client.create_monitor(
    project_name="customer-support-prod",
    metric="latency_p95",
    threshold=30.0,  # Alert if > 30 seconds
    interval_minutes=15
)

Resumen Completo del Sistema

python

# Usage
support = CustomerSupportAgent()

# Customer asks a question
result = support.handle_query(
    customer_id="cust_12345",
    query="What's your return policy for electronics?"
)

# System automatically:
# 1. Routes to FAQ agent
# 2. Answers based on policy knowledge
# 3. Checks confidence (high enough, no escalation)
# 4. Reviews quality
# 5. Returns answer with tracing

print(result["messages"][-1].content)

# For complex queries:
result = support.handle_query(
    customer_id="cust_12345",
    query="How many orders have I placed this year?"
)
# Routes to SQL agent → queries database → returns data

[!ÉXITO] This production-grade system combines routing, RAG, SQL queries, human handoff, quality review, and comprehensive monitoring — all the patterns from the Advanced course working together.

Preguntas Prácticas

Practice Question

What is the first node executed in the customer support graph?

Practice Question

What triggers a human handoff?

Practice Question

What mechanism allows human agents to handle escalated tickets?

Practice Question

What does the quality_review node evaluate?

Practice Question

What database is recommended for production state persistence?

Practice Question

What routing happens after a specialist agent completes?

Practice Question

What system monitors error rates and latency in production?

Practice Question

What does the RAG agent use to find relevant information?

Practice Question

What happens after a human resolves an escalated ticket?

Practice Question

What environment variable enables LangSmith tracing for this system?

[!ÉXITO]

Conclusiones Clave

Production support system integrates: routing, RAG, SQL, human handoff, quality review
Router classifies intent and dispatches to specialist agents
Escalation check triggers human handoff for low-confidence or complex queries
Human handoff uses interrupt() for pause/resume workflow
Quality review evaluates each interaction before returning to customer
LangSmith monitors error rates, latency, and other production metrics
PostgreSQL for production state persistence
All Advanced course patterns combined into a single cohesive system

Progreso100%

Optimización

Volver a Cursos

Contenido del Curso

1. Despliegue 2. Observabilidad con LangSmith 3. Agentes Jerárquicos 4. RAG Agentic 5. Agentes SQL 6. Patrón de Agente Reflexivo 7. Patrón Planificar y Ejecutar 8. Ejecución Paralela 9. Optimización 10. Proyecto Final: Agente de Soporte al Cliente de Producción