LangGraph + LangSmith: crear y monitorizar sistemas multiagente en producción (2026)

pip install langgraph langsmith langchain-openai langchain-anthropic

# Set environment variables
export LANGCHAIN_TRACING_V2=true
export LANGCHAIN_API_KEY=ls__your_api_key_here
export LANGCHAIN_PROJECT=production-agents
export OPENAI_API_KEY=sk-your-key

from typing import TypedDict, Annotated, Sequence
from langchain_core.messages import BaseMessage
import operator

class SupportAgentState(TypedDict):
    # All messages in the conversation
    messages: Annotated[Sequence[BaseMessage], operator.add]
    # Classified ticket category
    category: str
    # Retrieved knowledge base docs
    kb_results: list[dict]
    # Generated resolution
    resolution: str | None
    # Whether ticket needs escalation
    needs_escalation: bool
    # Escalation reason if applicable
    escalation_reason: str | None
    # Final response sent to customer
    final_response: str | None
    # Confidence score (0-1)
    confidence: float

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage
from langchain_core.prompts import ChatPromptTemplate

llm = ChatOpenAI(model="gpt-4o", temperature=0)

# Node 1: Classify the incoming ticket
def classifier_node(state: SupportAgentState) -> dict:
    prompt = ChatPromptTemplate.from_messages([
        ("system", """Classify this support ticket into one of:
        billing, technical, account, shipping, general
        Also assess confidence (0-1) that you can auto-resolve it.
        Return JSON: {{"category": "...", "confidence": 0.0}}"""),
        ("human", "{ticket}"),
    ])
    chain = prompt | llm
    ticket_text = state["messages"][-1].content
    result = chain.invoke({"ticket": ticket_text})
    parsed = parse_json(result.content)
    return {
        "category": parsed["category"],
        "confidence": parsed["confidence"],
    }

# Node 2: Query the RAG knowledge base
def rag_node(state: SupportAgentState) -> dict:
    query = state["messages"][-1].content
    # Your vector store query here (Pinecone, pgvector, etc.)
    results = vector_store.similarity_search(
        query, k=5, filter={"category": state["category"]}
    )
    return {"kb_results": [r.page_content for r in results]}

# Node 3: Generate resolution using retrieved context
def resolver_node(state: SupportAgentState) -> dict:
    context = "\n\n".join(state["kb_results"])
    prompt = ChatPromptTemplate.from_messages([
        ("system", f"""You are a support agent. Use this context to
        resolve the ticket. If you cannot resolve with high confidence,
        return null for resolution.\n\nContext:\n{context}"""),
        ("human", "{ticket}"),
    ])
    chain = prompt | llm
    result = chain.invoke({"ticket": state["messages"][-1].content})
    return {"resolution": result.content}

# Node 4: Human escalation node
def escalation_node(state: SupportAgentState) -> dict:
    return {
        "needs_escalation": True,
        "escalation_reason": (
            f"Low confidence ({state['confidence']:.0%}) — "
            f"routed to human agent"
        ),
        "final_response": (
            "We've escalated your ticket to a specialist "
            "who will follow up within 2 business hours."
        ),
    }

# Node 5: Format and send final response
def responder_node(state: SupportAgentState) -> dict:
    return {
        "final_response": state["resolution"],
        "messages": [AIMessage(content=state["resolution"])],
    }

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.sqlite import SqliteSaver

# Conditional routing function
def route_after_classification(state: SupportAgentState) -> str:
    """Route based on category and confidence score."""
    if state["confidence"] < 0.7:
        return "escalate"  # Low confidence → human
    if state["category"] == "billing":
        return "escalate"  # Always escalate billing
    return "rag"           # High confidence → try auto-resolve

def route_after_resolution(state: SupportAgentState) -> str:
    """Check if resolution is good enough to send."""
    if state["resolution"] is None:
        return "escalate"
    if len(state["resolution"]) < 50:
        return "escalate"  # Too short, probably incomplete
    return "respond"

# Build the graph
builder = StateGraph(SupportAgentState)

# Add all nodes
builder.add_node("classifier", classifier_node)
builder.add_node("rag", rag_node)
builder.add_node("resolver", resolver_node)
builder.add_node("escalation", escalation_node)
builder.add_node("responder", responder_node)

# Add edges
builder.add_edge(START, "classifier")
builder.add_conditional_edges(
    "classifier",
    route_after_classification,
    {
        "rag": "rag",
        "escalate": "escalation",
    },
)
builder.add_edge("rag", "resolver")
builder.add_conditional_edges(
    "resolver",
    route_after_resolution,
    {
        "respond": "responder",
        "escalate": "escalation",
    },
)
builder.add_edge("responder", END)
builder.add_edge("escalation", END)

# Compile with checkpointer for persistent memory
memory = SqliteSaver.from_conn_string(":memory:")  # Use PostgresSaver in prod
graph = builder.compile(
    checkpointer=memory,
    interrupt_before=["escalation"],  # Human-in-the-loop checkpoint
)

# Run the graph
config = {"configurable": {"thread_id": "ticket-48291"}}
result = graph.invoke(
    {"messages": [HumanMessage(content="My invoice is wrong")]},
    config=config,
)
print(result["final_response"])

# These two variables are all you need to start tracing
export LANGCHAIN_TRACING_V2=true
export LANGCHAIN_API_KEY=ls__your_api_key

# Optional but recommended
export LANGCHAIN_PROJECT=customer-support-prod
export LANGCHAIN_ENDPOINT=https://api.smith.langchain.com

# That's it. Your LangGraph runs are now fully traced.
# No code changes required.

from langsmith import Client
from langsmith.evaluation import evaluate, LangChainStringEvaluator

client = Client()

# Create or use an existing dataset
dataset = client.create_dataset(
    dataset_name="support-tickets-v2",
    description="Golden set of 200 resolved support tickets",
)

# Add examples: input ticket → expected resolution
client.create_examples(
    inputs=[
        {"messages": [{"role": "user", "content": "How do I cancel my subscription?"}]},
        {"messages": [{"role": "user", "content": "My payment failed but I was charged."}]},
    ],
    outputs=[
        {"final_response": "To cancel, go to Settings → Billing → Cancel Plan."},
        {"final_response": "We've issued a refund. It will appear in 3-5 business days."},
    ],
    dataset_id=dataset.id,
)

# Define evaluators
correctness_evaluator = LangChainStringEvaluator(
    "criteria",
    config={"criteria": "correctness"},
    prepare_output_fn=lambda run, example: {
        "prediction": run.outputs.get("final_response", ""),
        "reference": example.outputs.get("final_response", ""),
    },
)

# Run evaluation against your compiled graph
def run_agent(inputs: dict) -> dict:
    config = {"configurable": {"thread_id": f"eval-{id(inputs)}"}}
    result = graph.invoke(inputs, config=config)
    return {"final_response": result.get("final_response", "")}

results = evaluate(
    run_agent,
    data=dataset.name,
    evaluators=[correctness_evaluator],
    experiment_prefix="gpt-4o-support-v3",
    num_repetitions=1,
)

print(f"Average correctness: {results.to_pandas()['feedback.correctness'].mean():.2f}")

# Compile with human checkpoint before escalation
graph = builder.compile(
    checkpointer=memory,
    interrupt_before=["escalation"],  # Pause here for human review
)

config = {"configurable": {"thread_id": "ticket-48291"}}

# First run — graph pauses at "escalation" node
result = graph.invoke(
    {"messages": [HumanMessage(content="My invoice is wrong")]},
    config=config,
)
print("Graph paused. Current state:", result)
# → Graph stopped before escalation node

# Human reviews in your dashboard / Slack / UI...
# When approved, resume from the same thread_id:
graph.update_state(
    config,
    {"escalation_reason": "Human approved escalation — billing dispute"},
    as_node="escalation",
)

# Resume — graph continues from where it paused
final_result = graph.invoke(None, config=config)
print("Final response:", final_result["final_response"])

# Stream tokens as they're generated (for chat UIs)
async for event in graph.astream_events(
    {"messages": [HumanMessage(content="Help me understand my bill")]},
    config=config,
    version="v2",
):
    kind = event["event"]

    if kind == "on_chat_model_stream":
        # Token-level streaming — send to frontend via SSE/WebSocket
        chunk = event["data"]["chunk"]
        if chunk.content:
            print(chunk.content, end="", flush=True)

    elif kind == "on_chain_start":
        # Node started — useful for progress indicators
        node_name = event["name"]
        print(f"\n[{node_name} started]")

    elif kind == "on_tool_start":
        # Tool call started
        tool_name = event["name"]
        inputs = event["data"]["input"]
        print(f"\n🔧 Using tool: {tool_name}({inputs})")

    elif kind == "on_chain_end":
        # Node completed — log duration
        node_name = event["name"]
        print(f"\n[{node_name} completed]")