Tutorial LangGraph: tu primer agente multi-paso para automatizar soporte SaaS

~
# crear venv y dependencias
python -m venv .venv
source .venv/bin/activate
pip install langgraph langchain-anthropic psycopg python-dotenv

# variables de entorno
cat > .env <<'EOF'
ANTHROPIC_API_KEY=sk-ant-...
DATABASE_URL=postgresql://localhost/agent_dev
EOF

~
# agent/state.py
from typing import TypedDict, Annotated, Sequence
from langchain_core.messages import BaseMessage
import operator

class TicketCategory(TypedDict):
  category: str  # "billing" | "technical" | "general" | "urgent"
  confidence: float
  reasoning: str

class SupportAgentState(TypedDict):
  # Input
  ticket_id: str
  ticket_text: str
  customer_email: str

  # Computado por el agente
  classification: TicketCategory | None
  action: str | None  # "auto_respond" | "escalate_human" | "ask_clarification"
  response_draft: str | None
  response_sent: bool

  # Historial
  messages: Annotated[Sequence[BaseMessage], operator.add]

  # Safety
  iterations: int

~
# agent/nodes/classify.py
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage, SystemMessage
import json

from agent.state import SupportAgentState, TicketCategory

CLASSIFY_LLM = ChatAnthropic(
  model="claude-3-5-haiku-20241022",
  temperature=0,
  max_tokens=300,
)

SYSTEM = """Sos un clasificador de tickets de soporte. Dado el texto del ticket, devolvé JSON con:
- category: una de "billing", "technical", "general", "urgent"
- confidence: 0.0 a 1.0
- reasoning: explicación corta (max 30 palabras)

Solo JSON, sin texto adicional."""

def classify_ticket(state: SupportAgentState) -> dict:
  msg = CLASSIFY_LLM.invoke([
      SystemMessage(content=SYSTEM),
      HumanMessage(content=state["ticket_text"]),
  ])
  parsed: TicketCategory = json.loads(msg.content)
  return {
      "classification": parsed,
      "iterations": state.get("iterations", 0) + 1,
  }

~
# agent/routing.py
from agent.state import SupportAgentState

def route_after_classification(state: SupportAgentState) -> str:
  classification = state["classification"]
  cat = classification["category"]
  confidence = classification["confidence"]

  # Tickets urgentes siempre a humano
  if cat == "urgent":
      return "escalate"

  # Baja confianza en clasificación → escalar
  if confidence < 0.7:
      return "escalate"

  # Billing complejo → humano (compliance)
  if cat == "billing":
      return "escalate"

  # Resto: auto-respond
  return "respond"

~
# agent/nodes/respond.py
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage, SystemMessage

from agent.state import SupportAgentState
from agent.knowledge_base import retrieve_relevant_docs

RESPONSE_LLM = ChatAnthropic(
  model="claude-3-5-sonnet-20241022",
  temperature=0.3,
  max_tokens=600,
)

SYSTEM_TEMPLATE = """Sos un agente de soporte de {company}. Tono: amable, claro, conciso.
Usá ÚNICAMENTE la información del contexto siguiente para responder. Si no podés responder con lo que tenés, decí que vas a escalarlo a un humano y NO inventes información.

Contexto:
{context}"""

def generate_response(state: SupportAgentState) -> dict:
  docs = retrieve_relevant_docs(state["ticket_text"], top_k=5)
  context = "\n\n".join([d.content for d in docs])

  system = SYSTEM_TEMPLATE.format(company="TuSaaS", context=context)

  msg = RESPONSE_LLM.invoke([
      SystemMessage(content=system),
      HumanMessage(content=state["ticket_text"]),
  ])

  return {
      "response_draft": msg.content,
      "action": "auto_respond",
  }

~
# agent/graph.py
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.postgres import PostgresSaver

from agent.state import SupportAgentState
from agent.nodes.classify import classify_ticket
from agent.nodes.respond import generate_response
from agent.nodes.escalate import escalate_to_human
from agent.nodes.send import send_response
from agent.routing import route_after_classification

def build_graph(checkpointer=None):
  builder = StateGraph(SupportAgentState)

  # Registrar nodos
  builder.add_node("classify", classify_ticket)
  builder.add_node("respond", generate_response)
  builder.add_node("escalate", escalate_to_human)
  builder.add_node("send", send_response)

  # Entry point
  builder.set_entry_point("classify")

  # Edges
  builder.add_conditional_edges(
      "classify",
      route_after_classification,
      {"respond": "respond", "escalate": "escalate"},
  )
  builder.add_edge("respond", "send")
  builder.add_edge("escalate", END)  # humano se hará cargo
  builder.add_edge("send", END)

  return builder.compile(checkpointer=checkpointer)

~
# agent/graph.py (modificado)
def build_graph(checkpointer=None):
  # ... mismo código de arriba ...

  return builder.compile(
      checkpointer=checkpointer,
      interrupt_before=["send"],  # pausar antes de enviar
  )

~
# api/main.py
from fastapi import FastAPI
from langgraph.checkpoint.postgres import PostgresSaver

from agent.graph import build_graph

app = FastAPI()
checkpointer = PostgresSaver.from_conn_string(os.environ["DATABASE_URL"])
graph = build_graph(checkpointer=checkpointer)

@app.post("/tickets/{ticket_id}/process")
async def process_ticket(ticket_id: str, body: dict):
  config = {"configurable": {"thread_id": ticket_id}}

  # Correr hasta el interrupt
  result = await graph.ainvoke(
      {"ticket_id": ticket_id, "ticket_text": body["text"], ...},
      config=config,
  )

  if result.get("response_draft"):
      return {"status": "pending_approval", "draft": result["response_draft"]}
  return {"status": "escalated"}

@app.post("/tickets/{ticket_id}/approve")
async def approve_response(ticket_id: str, body: dict):
  config = {"configurable": {"thread_id": ticket_id}}

  # Si el humano edita la respuesta, actualizamos el state
  if body.get("edited_draft"):
      await graph.aupdate_state(config, {"response_draft": body["edited_draft"]})

  # Continuar desde el interrupt
  result = await graph.ainvoke(None, config=config)
  return {"status": "sent"}

~
# agent/observability.py
import structlog

logger = structlog.get_logger()

def log_transition(state, node_name, decision=None):
  logger.info(
      "agent_transition",
      ticket_id=state["ticket_id"],
      node=node_name,
      decision=decision,
      iterations=state.get("iterations", 0),
      classification=state.get("classification"),
  )

~
export LANGCHAIN_TRACING_V2=true
export LANGCHAIN_API_KEY=ls__...
export LANGCHAIN_PROJECT=support-agent