Projeto Final: Agente de Suporte ao Cliente de Produção

Construa um agente de suporte ao cliente de nível de produção com roteamento multi-agente, RAG, transferência para humano e monitoramento, integrando tudo do curso Avançado.

Este projeto de integração reúne tudo do curso Avançado: agentes hierárquicos, RAG, agentes SQL, interrupts, streaming, monitoramento com LangSmith e implantação. O resultado é um sistema de suporte ao cliente pronto para produção.

Arquitetura do 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

Passo 1: Estado e Configuração

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)

Passo 2: Definição 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

Passo 3: Agente Roteador

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"}

Passo 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 Conhecimento)

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 (Dados do 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"}

Passo 5: Escalonamento e Transferência para 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

O repasse para humano usa interrupt() para pausar e esperar que um agente humano assuma. A resposta do humano (resolvido + observações) é capturada no estado.

Passo 6: Revisão de Qualidade

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"}

Passo 7: Lógica do Roteador

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"

Passo 8: Construir o 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)

Passo 9: Configuração de Implantação

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}"
  }
}

Passo 10: Configuração de Monitoramento

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
)

Resumo Completo do 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

✅Sucesso

Este sistema de nível de produção combina roteamento, RAG, consultas SQL, repasse para humano, revisão de qualidade e monitoramento abrangente: todos os padrões do curso Avançado trabalhando juntos.