4.3 Sistemes Multi-Agent

Per Què Multi-Agent?

Un sol agent té limitacions: context finit, una única especialitat, un únic fil d'execució. Els sistemes multi-agent permeten dividir tasques complexes entre agents especialitzats que col·laboren, revisen i combinen resultats, igual que un equip humà.

---

🤔 El Problema d'un Únic Agent

Un agent generalista es queda curt quan la tasca és massa llarga o requereix habilitats molt diverses:

Tasca: "Desenvolupa una API REST, escriu els tests, documenta-la
        i desplega-la a AWS amb CI/CD"

Agent únic:
→ Context window desbordat
→ Errors al mesclar rols (dissenyador, programador, tester, devops)
→ No pot paral·lelitzar feina
→ Un error al pas 3 invalida tot el treball anterior

Sistema multi-agent:
→ Agent Arquitecte: dissenya l'API i els endpoints
→ Agent Programador: implementa el codi
→ Agent Tester: escriu i executa els tests
→ Agent DevOps: genera el Dockerfile i el pipeline CI/CD
→ Agent Supervisor: coordina, revisa i integra resultats

---

🏗️ Arquitectures Multi-Agent

Arquitectura 1: Supervisor + Treballadors

El patró més comú. Un agent supervisor rep la tasca, la descompon, i delega subtasques a agents especialitzats.

Usuari → [Supervisor] → [Agent A: Cerca]
                      → [Agent B: Anàlisi]
                      → [Agent C: Redacció]
                      ← Resultats integrats

# Arquitectura Supervisor amb LangGraph
# pip install langgraph langchain-openai python-dotenv

from typing import Annotated, Literal, TypedDict
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
import operator

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

# ── ESTAT COMPARTIT ────────────────────────────────────────────
class EstatEquip(TypedDict):
    tasca: str
    resultats: Annotated[list[str], operator.add]  # acumula resultats
    pas_actual: str
    finalitzat: bool

# ── AGENTS ESPECIALITZATS ──────────────────────────────────────
def agent_investigador(estat: EstatEquip) -> dict:
    """Agent especialitzat en cerca i recollida d'informació."""
    resposta = llm.invoke([
        SystemMessage(content="Ets un expert investigador. "
                              "Busca i sintetitza informació rellevant."),
        HumanMessage(content=f"Investiga aquest tema: {estat['tasca']}")
    ])
    return {
        "resultats": [f"INVESTIGACIÓ:\n{resposta.content}"],
        "pas_actual": "investigació completada"
    }

def agent_analista(estat: EstatEquip) -> dict:
    """Agent especialitzat en anàlisi i extracció d'insights."""
    context = "\n".join(estat["resultats"])
    resposta = llm.invoke([
        SystemMessage(content="Ets un expert analista. "
                              "Analitza la informació i extreu conclusions clau."),
        HumanMessage(content=f"Analitza:\nTasca: {estat['tasca']}\n"
                             f"Informació recollida:\n{context}")
    ])
    return {
        "resultats": [f"ANÀLISI:\n{resposta.content}"],
        "pas_actual": "anàlisi completada"
    }

def agent_redactor(estat: EstatEquip) -> dict:
    """Agent especialitzat en redacció d'informes finals."""
    context = "\n\n".join(estat["resultats"])
    resposta = llm.invoke([
        SystemMessage(content="Ets un expert redactor. "
                              "Genera informes clars, estructurats i professionals."),
        HumanMessage(content=f"Redacta un informe final sobre: {estat['tasca']}\n"
                             f"Basa't en:\n{context}")
    ])
    return {
        "resultats": [f"INFORME FINAL:\n{resposta.content}"],
        "pas_actual": "informe completat",
        "finalitzat": True
    }

# ── CONSTRUCCIÓ DEL GRAF ───────────────────────────────────────
graf = StateGraph(EstatEquip)
graf.add_node("investigador", agent_investigador)
graf.add_node("analista", agent_analista)
graf.add_node("redactor", agent_redactor)

graf.set_entry_point("investigador")
graf.add_edge("investigador", "analista")
graf.add_edge("analista", "redactor")
graf.add_edge("redactor", END)

equip = graf.compile()

# ── EXECUCIÓ ───────────────────────────────────────────────────
resultat = equip.invoke({
    "tasca": "Tendències actuals en agents d'IA per a l'educació",
    "resultats": [],
    "pas_actual": "inici",
    "finalitzat": False
})

print(resultat["resultats"][-1])  # Informe final

---

Arquitectura 2: Xarxa d'Iguals (Peer-to-Peer)

Agents que es comuniquen entre ells sense jerarquia. Útil per a debat, revisió per parells o negociació.

# Exemple: Agents de debat — un proposa, l'altre critica, un tercer arbitra
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage

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

def agent_proposador(tema: str, historial: list) -> str:
    """Proposa o defensa una posició."""
    resposta = llm.invoke([
        SystemMessage(content="Ets un agent que DEFENSA i proposa idees. "
                              "Sigues concret i argumenta amb fets."),
        HumanMessage(content=f"Tema: {tema}"),
        *historial,
        HumanMessage(content="Exposa o defensa el teu punt de vista.")
    ])
    return resposta.content

def agent_critic(tema: str, proposta: str) -> str:
    """Critica i troba debilitats en la proposta."""
    resposta = llm.invoke([
        SystemMessage(content="Ets un agent crític constructiu. "
                              "Identifica punts febles i proposa millores."),
        HumanMessage(content=f"Tema: {tema}\nProposta: {proposta}\n"
                             "Critica aquesta proposta de forma constructiva.")
    ])
    return resposta.content

def agent_arbitre(tema: str, debat: str) -> str:
    """Sintetitza el debat i extreu conclusions equilibrades."""
    resposta = llm.invoke([
        SystemMessage(content="Ets un àrbitre imparcial. "
                              "Sintetitza el debat i extreu les millors conclusions."),
        HumanMessage(content=f"Tema: {tema}\n\nDebat:\n{debat}\n"
                             "Genera una conclusió equilibrada.")
    ])
    return resposta.content

# ── EXECUCIÓ DEL DEBAT ─────────────────────────────────────────
tema = "Python és millor que JavaScript per al backend d'agents d'IA"

proposta = agent_proposador(tema, [])
critica = agent_critic(tema, proposta)

# Torn 2: el proposador respon a la crítica
historial = [
    AIMessage(content=proposta),
    HumanMessage(content=f"Crítica rebuda: {critica}")
]
replica = agent_proposador(tema, historial)

debat_complet = (f"PROPOSTA INICIAL:\n{proposta}\n\n"
                 f"CRÍTICA:\n{critica}\n\n"
                 f"RÈPLICA:\n{replica}")

conclusio = agent_arbitre(tema, debat_complet)
print(f"CONCLUSIÓ FINAL:\n{conclusio}")

---

Arquitectura 3: Pipeline Seqüencial amb Validació

Els agents treballen en cadena, on cada agent valida la sortida de l'anterior abans de continuar.

[Agent 1: Generació] → [Agent 2: Revisió] → [Agent 3: Millora] → Sortida
                              ↓ (si falla)
                         [Torna a Agent 1]

# Pipeline amb validació i reintents
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
from pydantic import BaseModel
import json

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

class ResultatValidacio(BaseModel):
    aprovat: bool
    puntuacio: int  # 0-10
    problemes: list[str]
    suggeriments: list[str]

def agent_generador(tasca: str, feedback: str = "") -> str:
    """Genera codi Python per a la tasca donada."""
    prompt_feedback = f"\nFeedback anterior a corregir:\n{feedback}" if feedback else ""
    resposta = llm.invoke([
        SystemMessage(content="Ets un programador Python expert. "
                              "Genera codi net, eficient i ben documentat."),
        HumanMessage(content=f"Escriu una funció Python per a: {tasca}"
                             f"{prompt_feedback}\nRetorna NOMÉS el codi.")
    ])
    return resposta.content

def agent_revisor(codi: str, tasca: str) -> ResultatValidacio:
    """Revisa la qualitat del codi i retorna feedback estructurat."""
    prompt = (
        f'Revisa aquest codi per a la tasca: "{tasca}"\n\n'
        f"CODI:\n{codi}\n\n"
        'Respon en JSON:\n'
        '{"aprovat": true/false, "puntuacio": 0-10, '
        '"problemes": [...], "suggeriments": [...]}'
    )
    resposta = llm.invoke([
        SystemMessage(content="Ets un revisor de codi expert. "
                              "Avalua el codi amb criteri professional. "
                              "Respon en JSON amb el format especificat."),
        HumanMessage(content=prompt)
    ])
    text = resposta.content.strip()
    if "```json" in text:
        text = text.split("```json")[1].split("```")[0].strip()
    elif "```" in text:
        text = text.split("```")[1].split("```")[0].strip()
    dades = json.loads(text)
    return ResultatValidacio(**dades)

# ── PIPELINE AMB REINTENTS ─────────────────────────────────────
def pipeline_generar_i_revisar(tasca: str, max_intents: int = 3) -> str:
    feedback = ""
    for intent in range(1, max_intents + 1):
        print(f"\n🔄 Intent {intent}/{max_intents}")

        codi = agent_generador(tasca, feedback)
        print(f"✍️  Codi generat ({len(codi)} caràcters)")

        validacio = agent_revisor(codi, tasca)
        print(f"🔍 Revisió: {validacio.puntuacio}/10 — "
              f"{'✅ Aprovat' if validacio.aprovat else '❌ Rebutjat'}")

        if validacio.aprovat and validacio.puntuacio >= 7:
            print("✅ Codi acceptat!")
            return codi

        feedback = (f"Problemes detectats: {', '.join(validacio.problemes)}\n"
                    f"Suggeriments: {', '.join(validacio.suggeriments)}")
        print(f"📝 Feedback: {feedback[:100]}...")

    print("⚠️  Màxim d'intents assolit. Retornant la millor versió.")
    return codi

codi_final = pipeline_generar_i_revisar(
    "calcular la distància entre dos punts GPS (lat/lon) en quilòmetres"
)
print(f"\nCODI FINAL:\n{codi_final}")

---

🔗 Patrons de Comunicació entre Agents

Memòria Compartida (Shared State)

El mecanisme més simple: tots els agents llegeixen i escriuen un estat global.

# LangGraph gestiona l'estat compartit automàticament
class EstatCompartit(TypedDict):
    missatges: Annotated[list, operator.add]  # acumula missatges
    documents: list[str]                       # documents processats
    metadades: dict                            # informació addicional
    decisio_final: str

Cua de Missatges (Message Queue)

Agents asíncrons que s'envien missatges a través d'una cua. Útil per a processament paral·lel.

import asyncio
from asyncio import Queue

async def agent_productor(cua: Queue, tasques: list[str]):
    """Genera tasques i les posa a la cua."""
    for tasca in tasques:
        await cua.put(tasca)
        print(f"📤 Tasca enviada: {tasca[:40]}...")
    await cua.put(None)  # Senyal de finalització

async def agent_consumidor(cua: Queue, id_agent: int):
    """Processa tasques de la cua."""
    while True:
        tasca = await cua.get()
        if tasca is None:
            await cua.put(None)  # Propaga la senyal als altres consumidors
            break
        await asyncio.sleep(0.1)  # Simular processament
        print(f"✅ Agent {id_agent} ha processat: {tasca[:40]}...")
        cua.task_done()

async def sistema_parallel():
    cua = Queue()
    tasques = [f"Analitzar document {i}" for i in range(10)]

    # Iniciar 3 agents consumidors en paral·lel
    await asyncio.gather(
        agent_productor(cua, tasques),
        agent_consumidor(cua, 1),
        agent_consumidor(cua, 2),
        agent_consumidor(cua, 3),
    )

asyncio.run(sistema_parallel())

---

🌐 Implementació Completa: Equip de Desenvolupament

Exemple realista: un equip d'agents que treballa com un equip de dev real.

# Equip multi-agent per crear una funcionalitat de software
# pip install langgraph langchain-openai

from typing import Annotated, TypedDict, Literal
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
import operator

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

class EstatProjecte(TypedDict):
    requisit: str
    disseny: str
    codi: str
    tests: str
    revisio: str
    aprovacio: bool
    iteracio: int

# ── AGENTS DEL EQUIP ───────────────────────────────────────────

def agent_dissenyador(estat: EstatProjecte) -> dict:
    """Crea el disseny tècnic del requisit."""
    resposta = llm.invoke([
        SystemMessage(content="Ets un arquitecte de software. "
                              "Dissenya la solució tècnica de forma clara i concreta."),
        HumanMessage(content=f"Dissenya la solució per a: {estat['requisit']}\n"
                             "Inclou: estructura de dades, funcions principals, "
                             "dependències i casos edge.")
    ])
    print("🎨 Dissenyador: disseny creat")
    return {"disseny": resposta.content}

def agent_programador(estat: EstatProjecte) -> dict:
    """Implementa el codi basat en el disseny."""
    feedback = (f"\nRevisions anteriors:\n{estat['revisio']}"
                if estat.get("revisio") else "")
    resposta = llm.invoke([
        SystemMessage(content="Ets un programador Python sènior. "
                              "Implementa codi de qualitat producció."),
        HumanMessage(content=f"Implementa: {estat['requisit']}\n"
                             f"Disseny tècnic: {estat['disseny']}"
                             f"{feedback}\nRetorna NOMÉS el codi Python.")
    ])
    print(f"💻 Programador: codi escrit (iteració {estat.get('iteracio', 1)})")
    return {"codi": resposta.content}

def agent_tester(estat: EstatProjecte) -> dict:
    """Escriu tests per al codi generat."""
    resposta = llm.invoke([
        SystemMessage(content="Ets un expert en testing. "
                              "Escriu tests exhaustius amb pytest."),
        HumanMessage(content=f"Escriu tests per al codi:\n{estat['codi']}\n"
                             f"Requisit original: {estat['requisit']}\n"
                             "Inclou: tests unitaris, casos límit i casos d'error.")
    ])
    print("🧪 Tester: tests escrits")
    return {"tests": resposta.content}

def agent_revisor_final(estat: EstatProjecte) -> dict:
    """Revisa el codi i els tests i decideix si s'aprova."""
    resposta = llm.invoke([
        SystemMessage(content="Ets un tech lead experimentat. "
                              "Revisa el codi i els tests amb rigor. "
                              "Respon amb APROVAT o NECESSITA_MILLORES, "
                              "seguit del teu feedback detallat."),
        HumanMessage(content=f"Revisa per al requisit: {estat['requisit']}\n\n"
                             f"CODI:\n{estat['codi']}\n\n"
                             f"TESTS:\n{estat['tests']}")
    ])
    text = resposta.content
    aprovat = "APROVAT" in text and "NECESSITA_MILLORES" not in text
    print(f"👀 Revisor: {'✅ Aprovat' if aprovat else '🔄 Necessita millores'}")
    return {"revisio": text, "aprovacio": aprovat}

def decidir_si_continuar(estat: EstatProjecte) -> Literal["programador", "fi"]:
    """Decideix si cal iterar o finalitzar."""
    if estat["aprovacio"] or estat.get("iteracio", 1) >= 3:
        return "fi"
    return "programador"

def incrementar_iteracio(estat: EstatProjecte) -> dict:
    return {"iteracio": estat.get("iteracio", 1) + 1}

# ── CONSTRUCCIÓ DEL GRAF ───────────────────────────────────────
graf = StateGraph(EstatProjecte)
graf.add_node("dissenyador", agent_dissenyador)
graf.add_node("programador", agent_programador)
graf.add_node("tester", agent_tester)
graf.add_node("revisor", agent_revisor_final)
graf.add_node("incrementar", incrementar_iteracio)

graf.set_entry_point("dissenyador")
graf.add_edge("dissenyador", "programador")
graf.add_edge("programador", "tester")
graf.add_edge("tester", "revisor")
graf.add_conditional_edges("revisor", decidir_si_continuar,
                           {"programador": "incrementar", "fi": END})
graf.add_edge("incrementar", "programador")

equip_dev = graf.compile()

# ── EXECUCIÓ ───────────────────────────────────────────────────
resultat = equip_dev.invoke({
    "requisit": "Funció que donada una llista de transaccions bancàries, "
                "retorni les 3 categories on s'ha gastat més diners",
    "disseny": "", "codi": "", "tests": "",
    "revisio": "", "aprovacio": False, "iteracio": 1
})

print(f"\nRESULTAT FINAL (iteració {resultat['iteracio']}):")
print(f"Aprovació: {'✅' if resultat['aprovacio'] else '⚠️'}")
print(f"\nCODI:\n{resultat['codi']}")

---

⚖️ Quan Usar Multi-Agent

Criteri	Un Sol Agent	Multi-Agent
Complexitat	Tasques simples i centrades	Tasques complexes multi-domini
Longitud del context	Cap a les 10-20 passes	Tasques llargues que desborden el context
Especialització	Generalista suficient	Cal expertise en àrees diverses
Paral·lelisme	No cal	Cal velocitat o processar en paral·lel
Qualitat	Acceptable en un pas	Cal revisió i validació creuada
Cost	Menor	Major (múltiples crides a l'API)

Complexitat Emergent

Els sistemes multi-agent poden tenir comportaments inesperats. Afegiu límits d'iteració, timeouts i logging exhaustiu per evitar bucles infinits i costos descontrolats.

---

✅ Activitats de Consolidació

Exercici 4.3.1 — Equip de Revisió de Codi

Crea un sistema de 2 agents on: 1. Agent A genera una funció Python per a un requisit donat 2. Agent B revisa el codi i retorna feedback 3. Si el feedback indica millores, Agent A torna a generar 4. El cicle es repeteix màxim 3 vegades — prova'l amb: "Funció per validar DNI espanyol".

Exercici 4.3.2 — Arquitectura de Debat

Implementa un sistema de 3 agents (Defensor, Crític, Moderador) per debatre: - Agent Defensor: argumenta a favor - Agent Crític: identifica riscos i problemes - Agent Moderador: sintetitza i conclou el debat sobre "SQLite vs PostgreSQL per a un projecte educatiu petit".

Exercici 4.3.3 — Anàlisi de Rendiment

Compara el sistema multi-agent de l'exemple 1 amb un agent únic per a la mateixa tasca: 1. Mesura el temps d'execució de cada enfocament 2. Compara la qualitat dels resultats 3. Reflexiona: en quin cas val la pena la complexitat afegida?

---

📚 Referències

• Wu, Q. et al. (2023). "AutoGen: Enabling Next-Gen LLM Applications via Multi-Agent Conversation". arXiv:2308.08155
• LangGraph Multi-Agent Documentation. langchain-ai.github.io/langgraph
• Park, J. et al. (2023). "Generative Agents: Interactive Simulacra of Human Behavior". arXiv:2304.03442
• Hong, S. et al. (2023). "MetaGPT: Meta Programming for Multi-Agent Collaborative Framework". arXiv:2308.00352