5.1 LangChain i LangGraph en Profunditat

Versió de Referència

LangChain 0.3.x i LangGraph 0.2.x. LangChain va fer un refactoring important de 0.1 a 0.2 i 0.3 — molts tutorials antics (pre-2024) usen l'API antiga. Consulta sempre la documentació oficial.

---

🦜 LangChain 0.3: L'Ecosistema

LangChain no és una sola llibreria sinó un ecosistema de paquets:

langchain-core          → Interfaces base (LLM, Tool, Memory...)
langchain               → Implementació de chains i agents
langchain-openai        → Integració amb OpenAI
langchain-anthropic     → Integració amb Anthropic/Claude
langchain-community     → 200+ integracions de tercers
langchain-experimental  → Eines experimentals (Python REPL, etc.)
langgraph               → Agents basats en grafs d'estat
langsmith               → Observabilitat i tracing

# Instal·lació completa per al curs
pip install langchain==0.3.7 \
            langchain-openai==0.2.6 \
            langchain-anthropic==0.3.0 \
            langchain-community==0.3.7 \
            langgraph==0.2.45 \
            langsmith \
            chromadb \
            duckduckgo-search

---

🔗 LCEL: LangChain Expression Language

LangChain 0.3 usa LCEL per composar components via l'operador | (pipe):

from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.output_parsers import StructuredOutputParser, ResponseSchema

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

# ── Chain bàsica: Prompt | LLM ─────────────────────────────────

prompt = ChatPromptTemplate.from_template(
    "Explica '{concepte}' de forma clara per a alumnes de CFGS."
)

chain_basica = prompt | llm  # ← Composició via pipe!

result = chain_basica.invoke({"concepte": "subnetting IPv4"})
print(result.content)

# ── Chain amb parser d'output ──────────────────────────────────

from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from pydantic import BaseModel, Field

class ExplicacioConcepte(BaseModel):
    concepte: str = Field(description="Nom del concepte")
    definicio: str = Field(description="Definició en 2 frases")
    exemple: str = Field(description="Exemple pràctic")
    dificultat: str = Field(description="baixa/mitja/alta")

parser = JsonOutputParser(pydantic_object=ExplicacioConcepte)

prompt_json = ChatPromptTemplate.from_messages([
    ("system", "Respon SEMPRE en format JSON vàlid."),
    ("human", "Explica el concepte: {concepte}\n{format_instructions}"),
]).partial(format_instructions=parser.get_format_instructions())

chain_json = prompt_json | llm | parser

result = chain_json.invoke({"concepte": "VLAN"})
print(result)
# → {'concepte': 'VLAN', 'definicio': '...', 'exemple': '...', 'dificultat': 'mitja'}

# ── Chain Seqüencial (encadenament de steps) ───────────────────

prompt_resum = ChatPromptTemplate.from_template(
    "Resumeix en 3 punts: {text}"
)
prompt_traduccio = ChatPromptTemplate.from_template(
    "Tradueix al català: {text}"
)

chain_seq = (
    {"text": prompt | llm | StrOutputParser()}
    | prompt_resum | llm | StrOutputParser()
)

---

🕸️ LangGraph: Agents basats en Grafs d'Estat

LangGraph permet crear agents amb lògica de control complexa: branching, cicles, human-in-the-loop, i col·laboració multi-agent.

Conceptes Fonamentals

NODES   → Funcions que processen l'estat i el modifiquen
EDGES   → Connexions entre nodes (condicionals o directes)
ESTAT   → Un diccionari TypedDict que flueix per tot el graf
CICLES  → Permeten iteració (el que fan els agents ReAct)

Agent ReAct amb LangGraph

from typing import TypedDict, Annotated, Sequence
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode, tools_condition
from langchain_openai import ChatOpenAI
from langchain_core.messages import BaseMessage, HumanMessage
import operator

# ── 1. Definir l'Estat del Graf ────────────────────────────────

class AgentState(TypedDict):
    # La llista de missatges s'acumula amb operator.add
    messages: Annotated[Sequence[BaseMessage], operator.add]

# ── 2. Definir els Nodes ───────────────────────────────────────

llm = ChatOpenAI(model="gpt-4o", temperature=0)
tools = [dns_lookup, get_weather, calculadora]  # Eines definides prèviament
llm_with_tools = llm.bind_tools(tools)

def call_model(state: AgentState) -> AgentState:
    """Node que crida el LLM."""
    messages = state["messages"]
    response = llm_with_tools.invoke(messages)
    return {"messages": [response]}

tool_node = ToolNode(tools)  # Node pre-construït per executar eines

# ── 3. Construir el Graf ───────────────────────────────────────

workflow = StateGraph(AgentState)

# Afegir nodes
workflow.add_node("agent",  call_model)
workflow.add_node("tools",  tool_node)

# Punt d'entrada
workflow.set_entry_point("agent")

# Edge condicional: si l'agent vol usar eines → "tools", sinó → END
workflow.add_conditional_edges(
    "agent",
    tools_condition,  # Funció pre-construïda que detecta tool calls
    {
        "tools": "tools",  # Si hi ha tool calls → anar a "tools"
        END: END            # Si no → finalitzar
    }
)

# Després d'executar eines, tornar a l'agent
workflow.add_edge("tools", "agent")

# Compilar el graf
app = workflow.compile()

# ── 4. Executar ────────────────────────────────────────────────

result = app.invoke({
    "messages": [HumanMessage(content="Quina IP té www.google.com i quin temps fa a la seva seu?")]
})

for msg in result["messages"]:
    print(f"{msg.__class__.__name__}: {msg.content[:100]}")

Human-in-the-Loop amb LangGraph

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, END
import uuid

# Checkpointer per a persistència i human-in-the-loop
checkpointer = MemorySaver()

# ── Agent amb aprovació humana per a accions crítiques ─────────

def check_needs_approval(state: AgentState) -> str:
    """Decideix si cal aprovació humana per a l'acció."""
    last_msg = state["messages"][-1]

    # Si l'agent vol usar eines destructives → demanar aprovació
    if hasattr(last_msg, "tool_calls"):
        for tc in last_msg.tool_calls:
            if tc["name"] in {"esborrar_fitxer", "modificar_config", "reiniciar_servei"}:
                return "human_approval"  # → Aturar i esperar humà

    return tools_condition(state)  # → Comportament normal

workflow_hitl = StateGraph(AgentState)
workflow_hitl.add_node("agent", call_model)
workflow_hitl.add_node("tools", tool_node)
workflow_hitl.add_node("human_approval", lambda state: state)  # Node de pausa

workflow_hitl.set_entry_point("agent")
workflow_hitl.add_conditional_edges("agent", check_needs_approval)
workflow_hitl.add_edge("tools", "agent")
workflow_hitl.add_edge("human_approval", "tools")  # Continua si l'humà aprova

# Compilar amb checkpointer per poder reprendre
app_hitl = workflow_hitl.compile(
    checkpointer=checkpointer,
    interrupt_before=["human_approval"]  # Aturar ABANS d'aprovar
)

# Primera execució (s'atura a human_approval)
thread_id = str(uuid.uuid4())
config = {"configurable": {"thread_id": thread_id}}

for event in app_hitl.stream(
    {"messages": [HumanMessage("Reinicia el servidor Apache")]},
    config, stream_mode="values"
):
    print(event)

# Inspecció de l'estat (per a l'aprovació humana)
snapshot = app_hitl.get_state(config)
print("Pending:", snapshot.next)  # → ('human_approval',)

# Reprendre l'execució (l'humà ha aprovat)
for event in app_hitl.stream(None, config, stream_mode="values"):
    print(event)

---

🔭 LangSmith: Observabilitat i Debugging

LangSmith és la plataforma d'observabilitat per a LangChain. Permet: - Veure les traces completes de cada execució - Debugar errors en el raonament de l'agent - Avaluar la qualitat de les respostes - Monitorar el cost i la latència

import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_ENDPOINT"]   = "https://api.smith.langchain.com"
os.environ["LANGCHAIN_API_KEY"]    = "lsv2_..."  # Obtenir a smith.langchain.com
os.environ["LANGCHAIN_PROJECT"]    = "curs-agents-ia"

# A partir d'aquí, TOTES les execucions de LangChain s'envien a LangSmith
# sense cap canvi al codi!

result = executor.invoke({"input": "Quin temps fa a Barcelona?"})
# → La traça completa (prompt, eines, respostes) es pot veure a LangSmith UI

---

✅ Activitats

Exercici 5.1.1 — LCEL Pipeline

Construeix una pipeline LCEL que: (1) prengui un topic, (2) generi 3 preguntes d'examen, (3) generi les respostes model, (4) retorni un JSON estructurat.

Exercici 5.1.2 — Agent LangGraph amb Cicles

Implementa un agent LangGraph per a diagnòstic de xarxa que: (1) faci ping a un host, (2) si falla, consulti DNS, (3) si DNS falla, consulti si el host existeix a Wikipedia. Dibuixa el graf primer, després implementa'l.

Exercici 5.1.3 — Human-in-the-Loop

Modifica l'agent de l'exercici anterior per demanar aprovació humana ABANS d'executar qualsevol diagnòstic sobre IPs de xarxa privada (192.168.x.x, 10.x.x.x). Usa interrupt_before.