Una guía de codificación para diseñar y organizar flujos de trabajo avanzados de múltiples agentes basados ​​en ReAct con AgentScope y OpenAI

En este tutorial, creamos un sistema avanzado de respuesta a incidentes de múltiples agentes utilizando AgentScope. Orquestamos múltiples agentes de ReAct, cada uno con una función claramente definida, como enrutamiento, clasificación, análisis, redacción y revisión, y los conectamos a través de enrutamiento estructurado y un centro de mensajes compartido. Al integrar modelos OpenAI, llamadas de herramientas livianas y un runbook interno simple, demostramos cómo se pueden componer flujos de trabajo agentes complejos y del mundo real en Python puro sin infraestructura pesada ni código frágil. Consulta los CÓDIGOS COMPLETOS aquí.

!pip -q install “agentscope>=0.1.5” pydantic nest_asyncio import os, json, re de getpass import getpass de escribir import Literal de pydantic import BaseModel, importación de campo nest_asyncio nest_asyncio.apply() de agentscope.agent import ReActAgent de agentscope.message import Msg, TextBlock de agentscope.model import OpenAIChatModel de agentscope.formatter import OpenAIChatFormatter desde agentscope.memory importar InMemoryMemory desde agentscope.tool importar Toolkit, ToolResponse, ejecutar_python_code desde agentscope.pipeline importar MsgHub, sequential_pipeline si no os.environ.get(“OPENAI_API_KEY”): os.environ[“OPENAI_API_KEY”] = getpass(“Ingrese OPENAI_API_KEY (oculto): “) OPENAI_MODEL = os.environ.get(“OPENAI_MODEL”, “gpt-4o-mini”)

Configuramos el entorno de ejecución e instalamos todas las dependencias necesarias para que el tutorial se ejecute de manera confiable en Google Colab. Cargamos de forma segura la clave API de OpenAI e inicializamos los componentes principales de AgentScope que se compartirán entre todos los agentes. Consulta los CÓDIGOS COMPLETOS aquí.

LIBRO DE EJECUCIÓN = [
{“id”: “P0”, “title”: “Severity Policy”, “text”: “P0 critical outage, P1 major degradation, P2 minor issue”},
{“id”: “IR1”, “title”: “Incident Triage Checklist”, “text”: “Assess blast radius, timeline, deployments, errors, mitigation”},
{“id”: “SEC7”, “title”: “Phishing Escalation”, “text”: “Disable account, reset sessions, block sender, preserve evidence”},
]

def _score(q, d): q = set(re.findall(r”[a-z0-9]+”, q.inferior())) d = re.findall(r”[a-z0-9]+”, d.lower()) return suma(1 para w en d si w en q) / max(1, len(d)) async def search_runbook(query: str, top_k: int = 2) -> ToolResponse: clasificado = ordenado(RUNBOOK, key=lambda r: _score(query, r[“title”] +r[“text”]), inversa = Verdadero)[: max(1, int(top_k))]
texto = “\n\n”.join(f”[{r[‘id’]}]{r[‘title’]}\n{r[‘text’]}” para r en clasificación) return ToolResponse(content=[TextBlock(type=”text”, text=text)]) kit de herramientas = kit de herramientas() kit de herramientas.register_tool_function(search_runbook) kit de herramientas.register_tool_function(execute_python_code)

Definimos un runbook interno liviano e implementamos sobre él una herramienta de búsqueda simple basada en relevancia. Registramos esta función junto con una herramienta de ejecución de Python, lo que permite a los agentes recuperar conocimiento de políticas o calcular resultados de forma dinámica. Demuestra cómo aumentamos los agentes con capacidades externas más allá del razonamiento del lenguaje puro. Consulta los CÓDIGOS COMPLETOS aquí.

def make_model(): devuelve OpenAIChatModel( model_name=OPENAI_MODEL, api_key=os.environ[“OPENAI_API_KEY”]generate_kwargs={“temperatura”: 0.2}, ) clase Ruta(BaseModel): carril: Literal[“triage”, “analysis”, “report”, “unknown”] = Field(…) objetivo: str = Field(…) router = ReActAgent( name=”Router”, sys_prompt=”Enrutar la solicitud para clasificación, análisis o informe y salida JSON estructurado solamente.”, model=make_model(), formatter=OpenAIChatFormatter(), memoria=InMemoryMemory(), ) triager = ReActAgent( name=”Triager”, sys_prompt=”Clasificar la gravedad y las acciones inmediatas usando runbook buscar cuando sea útil.”, model=make_model(), formatter=OpenAIChatFormatter(), memoria=InMemoryMemory(), toolkit=toolkit, ) analista = ReActAgent( name=”Analyst”, sys_prompt=”Analizar registros y calcular resúmenes usando la herramienta Python cuando sea útil.”, model=make_model(), formatter=OpenAIChatFormatter(), memoria=InMemoryMemory(), toolkit=toolkit, ) escritor = ReActAgent( nombre=”Escritor”, sys_prompt=”Escribe un informe de incidente conciso con una estructura clara.”, model=make_model(), formatter=OpenAIChatFormatter(), memoria=InMemoryMemory(), ) revisor = ReActAgent( nombre=”Revisor”, sys_prompt=”Critica y mejora el informe con correcciones concretas.”, model=make_model(), formateador=OpenAIChatFormatter(), memoria=InMemoryMemory(), )

Construimos múltiples agentes ReAct especializados y un enrutador estructurado que decide cómo se debe manejar cada solicitud de usuario. Asignamos responsabilidades claras a los agentes de clasificación, análisis, redacción y revisión, asegurando la separación de inquietudes. Consulta los CÓDIGOS COMPLETOS aquí.

LOGS = “””marca de tiempo, servicio, estado, latencia_ms, error 2025-12-18T12:00:00Z, pago, 200,180, falso 2025-12-18T12: 00: 05Z, pago, 500,900, verdadero 2025-12-18T12:00:10Z,auth,200,120,false 2025-12-18T12:00:12Z,checkout,502,1100,true 2025-12-18T12:00:20Z,búsqueda,200,140,false 2025-12-18T12:00:25Z,checkout,500,950,true “”” def msg_text(m: Msg) -> str: bloques = m.get_content_blocks(“text”) si los bloques es Ninguno: devolver “” si isinstance(blocks, str): devolver bloques if isinstance(blocks, list): devolver “\n”.join(str(x) para x en bloques) return str (bloques)

Introducimos datos de registro de muestra y una función de utilidad que normaliza los resultados del agente en texto limpio. Nos aseguramos de que los agentes posteriores puedan consumir y refinar de forma segura respuestas anteriores sin problemas de formato. Se centra en hacer que la comunicación entre agentes sea sólida y predecible. Consulta los CÓDIGOS COMPLETOS aquí.

async def run_demo(user_request: str): route_msg = esperar enrutador(Msg(“usuario”, user_request, “usuario”), estructurado_modelo=Ruta) carril = (route_msg.metadata o {}).get(“carril”, “desconocido”) if carril == “triage”: primero = esperar triager(Msg(“usuario”, user_request, “usuario”)) elif carril == “análisis”: primero = esperar analista(Msg(“usuario”, solicitud_usuario + “\n\nRegistros:\n” + REGISTROS, “usuario”)) elif lane == “informe”: borrador = esperar escritor(Msg(“usuario”, solicitud_usuario, “usuario”)) primero = esperar revisor(Msg(“usuario”, “Revisar y mejorar:\n\n” + msg_text(borrador), “usuario”)) else: primero = Msg(“sistema”, “No se pudo enrutar la solicitud.”, “sistema”) asíncrono con MsgHub(participantes=[triager, analyst, writer, reviewer]anuncio=Msg(“Anfitrión”, “Refinar la respuesta final en colaboración.”, “asistente”),): await sequential_pipeline([triager, analyst, writer, reviewer]) return {“route”: route_msg.metadata, “initial_output”: msg_text(first)} resultado = await run_demo( “Vemos errores 5xx repetidos en el proceso de pago. Clasifique la gravedad, analice los registros y genere un informe de incidentes”. ) print(json.dumps(resultado, sangría=2))

Orquestamos el flujo de trabajo completo enrutando la solicitud, ejecutando el agente apropiado y ejecutando un ciclo de refinamiento colaborativo utilizando un centro de mensajes. Coordinamos múltiples agentes en secuencia para mejorar el resultado final antes de devolvérselo al usuario. Reúne todos los componentes anteriores en un proceso de agente coherente y de extremo a extremo.

En conclusión, mostramos cómo AgentScope nos permite diseñar sistemas de agentes robustos, modulares y colaborativos que van más allá de las interacciones de un solo mensaje. Enrutamos tareas de forma dinámica, invocamos herramientas solo cuando era necesario y refinamos los resultados mediante la coordinación de múltiples agentes, todo dentro de una configuración de Colab limpia y reproducible. Este patrón ilustra cómo podemos escalar desde simples experimentos con agentes hasta procesos de razonamiento de estilo producción, manteniendo al mismo tiempo la claridad, el control y la extensibilidad en nuestras aplicaciones de IA agente.

Consulta los CÓDIGOS COMPLETOS aquí. Además, no dude en seguirnos en Twitter y no olvide unirse a nuestro SubReddit de más de 100.000 ML y suscribirse a nuestro boletín. ¡Esperar! estas en telegrama? Ahora también puedes unirte a nosotros en Telegram.

Asif Razzaq es el director ejecutivo de Marktechpost Media Inc.. Como empresario e ingeniero visionario, Asif está comprometido a aprovechar el potencial de la inteligencia artificial para el bien social. Su esfuerzo más reciente es el lanzamiento de una plataforma de medios de inteligencia artificial, Marktechpost, que se destaca por su cobertura en profundidad del aprendizaje automático y las noticias sobre aprendizaje profundo que es técnicamente sólida y fácilmente comprensible para una amplia audiencia. La plataforma cuenta con más de 2 millones de visitas mensuales, lo que ilustra su popularidad entre el público.