Agente de vigilancia de mercado con LangGraph y Strands en AgentCore

A medida que las aplicaciones de inteligencia artificial evolucionan desde simples chatbots hasta sofisticados sistemas autónomos, las organizaciones enfrentan nuevos desafíos al orquestar flujos de trabajo complejos de múltiples agentes que puedan manejar escenarios de producción del mundo real. Los enfoques tradicionales de un solo agente a menudo se quedan cortos cuando se trata de procesos de negocio complejos que requieren experiencia especializada, toma de decisiones dinámica y mecanismos sólidos de recuperación de errores. La industria de servicios financieros es un ejemplo de este desafío. Los sistemas de vigilancia del mercado deben coordinar múltiples agentes especializados para analizar patrones comerciales, investigar actividades sospechosas y generar informes completos manteniendo estrictos estándares de cumplimiento y confiabilidad.

La solución combina dos marcos: LangGraph para la orquestación del flujo de trabajo a nivel macro y Strands para el razonamiento inteligente de los agentes. LangGraph se destaca en la gestión de gráficos dirigidos y de estado para la coordinación de múltiples agentes. Le brinda un control detallado tanto sobre la ejecución del flujo de trabajo como sobre el estado que se puede compartir entre los agentes. Su capa de persistencia central admite características críticas para la producción, incluidas las interacciones humanas en el circuito y una sólida recuperación de fallas basada en puntos de control. Mientras tanto, Strands Agent sirve como motor de razonamiento dentro de los nodos de flujo de trabajo individuales. Ofrece capacidades independientes del modelo que se integran con varios proveedores de modelos de lenguaje grandes (LLM) al tiempo que mantiene una integración de herramientas flexible y una observabilidad integral.

Con el lanzamiento de Amazon Bedrock AgentCore el año pasado, la producción de una solución agente podría simplificarse para muchos casos de uso. La combinación proporciona una base sólida para sistemas de IA agentes listos para producción que pueden manejar casos de uso complejos y al mismo tiempo ayudar a brindar la confiabilidad y observabilidad de la infraestructura que exigen las aplicaciones empresariales.

En esta publicación, demostramos cómo diseñar e implementar un sistema de IA multiagente utilizando LangGraph y Strands en la infraestructura de AWS. Aprenderá cómo implementar la orquestación del flujo de trabajo impulsado por el estado con el sistema de puntos de control de LangGraph, integrar agentes de Strands para tareas de razonamiento especializadas y utilizar AgentCore para una implementación de producción escalable. La solución completa está disponible en GitHub.

Hilos: Razonamiento de agentes inteligentes

Strands Agent opera en una arquitectura independiente del modelo que se adapta a su infraestructura existente sin imponer restricciones arquitectónicas. El agente implementa un bucle de razonamiento agente que evalúa continuamente los resultados de la herramienta y toma decisiones basadas en resultados intermedios, para que pueda crear flujos de trabajo de análisis sofisticados de varios pasos. El marco incluye gestión integral de sesiones y estados y múltiples administradores de conversaciones para evitar que la ventana de contexto se desborde.

Con Strands, puede configurar interacciones de herramientas externas definiendo esquemas de herramientas y patrones de acceso. Para nuestro agente de vigilancia, separamos el descubrimiento de datos de la recuperación para evitar alucinaciones y fortalecer la solución contra ataques de inyección. Usamos herramientas como get_report_list y get_report_schema para buscar informes y run_report para crear la consulta SQL con parámetros validados y ejecutarla.

Creamos un agente security_monitor con las siguientes herramientas y un indicador del sistema:

from strands import Agente, herramienta de strands.models.bedrock import BedrockModel modelo = BedrockModel( model_id="us.anthropic.claude-sonnet-4-6", region_name="us-east-1", max_tokens=16000, adicional_request_fields={ "thinking": {"type": "adaptive", "budget_tokens": 8000}, }, cache_prompt="default", ) @tool def get_report_list(agent_name: str) -> str: """Carga la lista de informes disponibles para un agente específico. Args: agent_name: Nombre del agente (por ejemplo, 'security_monitor'). Devuelve: str: matriz JSON de objetos de informe con nombre y descripción. """ reports_data = load_agent_reports(agent_name) return json.dumps(reports_data["reports"], indent=2) @tool def get_report_schema(report_name: str, query_intent: str) -> str: """Cargue definiciones de columnas para un informe para que pueda crear consultas. Args: report_name: nombre del informe (por ejemplo, 'TradeActivity'). query_intent: descripción de qué datos extraer. Devuelve: str: objeto JSON con parámetros y definiciones de columna. """ return json.dumps(load_json_report_definition(report_name), indent=2) @tool def run_report( report_name: str, filters: Dict[str, Any], limit: Opcional[int] = Ninguno, ) -> Dict[str, Any]: """Ejecuta un informe predefinido. LLM nunca escribe SQL sin formato, por lo que los valores de filtro no se pueden inyectar en la consulta. Args: report_name: un informe de `get_report_list` (por ejemplo, 'TradeActivity': filtros de igualdad codificados por nombre de columna, por ejemplo, {"symbol": "AAPL", "date": "2024-03-15"}. bool, 'data': str (CSV), 'error': str o Ninguno} """ esquema = load_json_report_definition(report_name) Allow_columns = {c["name"] for c in esquema["columnas"]} # Rechazar cualquier campo de filtro que no esté en la lista de columnas permitidas del informe. desconocido = set(filtros) – columnas_permitidas si se desconoce: elevar ValueError( f"Campos de filtro desconocidos {ordenados(desconocidos)} para {nombre_informe}. " f"Permitido: {ordenados(columnas_permitidas)}" ) # Construya SQL con parámetros de enlace con nombre. donde = " AND ".join(f"{campo} = :{campo}" para el campo en filtros) sql = f"SELECT * FROM {schema['reportName']}" if donde: sql += f" WHERE {where}" si el límite no es Ninguno: si no es instancia(límite, int) o no 1 <= límite <= 10_000: elevar ValueError("el límite debe ser un número entero en [1, 10000]") sql += f" LIMIT {límite}" return query_market_data(report_name=report_name, sql=sql, bind=filters) security_monitor = Agente( modelo=modelo, system_prompt=SECURITY_MONITOR_PROMPT, herramientas=[get_report_list, get_report_schema, run_report], nombre="security_monitor", )

LangGraph: orquestación de flujo de trabajo macro

LangGraph proporciona orquestación de nivel de producción para sistemas multiagente a través de tres capacidades principales que lo hacen ideal para flujos de trabajo complejos de IA.

Máquinas de estado basadas en gráficos: LangGraph modela los flujos de trabajo de los agentes como gráficos dirigidos donde los nodos representan funciones que contienen la lógica del agente y los bordes determinan el flujo de ejecución. Este enfoque declarativo transforma un razonamiento complejo de varios pasos en código legible y mantenible. Los gráficos admiten ramificaciones condicionales, ejecución paralela y enrutamiento dinámico. Estas capacidades son importantes para escenarios del mundo real donde los flujos de trabajo se adaptan en función de resultados intermedios.

Gestión de estado persistente: el sistema de puntos de control del marco captura automáticamente el estado completo del flujo de trabajo después de la ejecución de cada nodo. Con estos puntos de control, puede recuperarse sin problemas de las fallas y respaldar las interacciones humanas en el circuito. Cuando los analistas necesitan revisar resultados intermedios o se producen errores, el sistema se restaura desde el punto de control exacto. Luego se reanuda sin perder el trabajo anterior. Esta arquitectura con estado admite patrones de uso común, como conversaciones de varios turnos, refinamiento iterativo e investigaciones de larga duración que abarcan horas o días.

Confiabilidad de la producción: LangGraph incluye estrategias de reintento integradas con retroceso exponencial para manejar la limitación u otras fallas que puedan ocurrir en su sistema. También proporciona observabilidad integral a través de OpenTelemetry para una integración sencilla con la mayoría de las aplicaciones de observabilidad.

Construyamos una capa de orquestación que enrute las consultas entre nuestros agentes especializados de Strands. El siguiente código define nuestro gráfico de flujo de trabajo: un estado compartido, un orquestador que selecciona qué agentes especializados invocar, enrutamiento condicional entre ellos y persistencia respaldada por puntos de control para recuperación y revisión humana en el circuito.

desde escribir import TypedDict, Opcional, Lista, Dict, Cualquiera desde langgraph.graph import END, StateGraph desde langgraph_checkpoint_aws import AgentCoreMemorySaver clase AgentState(TypedDict): query_text: str session_id: Opcional[str] agent_task_map: Opcional[Dict[str, str]] require_agents: Opcional[List[str]] current_agent_index: Opcional[int] # Cada especialista escribe sus ideas aquí security_monitor_insights: Opcional[Dict[str, Any]] broker_monitor_insights: Opcional[Dict[str, Any]] Risk_monitor_insights: Opcional[Dict[str, Any]] intel_analyst_insights: Opcional[Dict[str, Any]] synthesizer_insights: Opcional[str] SPECIALIST_NODES = { "security_monitor": security_monitor_node, "broker_monitor": broker_monitor_node, "risk_monitor": Risk_monitor_node, "intel_analyst": intel_analyst_node, } def route_analysts(state: AgentState) -> str: """Enrutamiento dinámico — recorra la lista de agentes_requeridos por índice.""" requerido = state.get("agentes_requeridos",[]) index = state.get("current_agent_index", 0) si no es necesario: devuelve END si índice < len(obligatorio): devuelve requerido[index] devuelve "synthesizer" # Construye el gráfico flujo de trabajo = StateGraph(AgentState) flujo de trabajo.add_node("orchestrator", Orchestrator_node) para nombre, node_fn en SPECIALIST_NODES.items(): flujo de trabajo.add_node(nombre, node_fn) flujo de trabajo.add_node("synthesizer", synthesizer_node) flujo de trabajo.set_entry_point("orchestrator") # Bordes condicionales — el orquestador y cada especialista enrutan a través de route_analysts, que pueden pasar a cualquier especialista o al sintetizador. ALL_TARGETS = {nombre: nombre para nombre en SPECIALIST_NODES} | { "sintetizador": "sintetizador", FIN: FIN, } flujo de trabajo.add_conditional_edges("orchestrator", route_analysts, ALL_TARGETS) para el nombre en SPECIALIST_NODES: flujo de trabajo.add_conditional_edges(nombre, route_analysts, ALL_TARGETS) flujo de trabajo.add_edge("sintetizador", FIN) # AgentCoreMemorySaver comprueba el estado después de cada nodo.

Orquestación del flujo de trabajo LangGraph para el agente de análisis de mercado

Por qué LangGraph y Strands juntos

Muchos casos de uso empresarial deben basarse en flujos de trabajo estrictos y predefinidos. Depender únicamente de la naturaleza no determinista de un LLM para ejecutar los pasos correctos es un riesgo. Sin embargo, los pasos específicos dentro de esos flujos de trabajo requieren un razonamiento ágil con la inteligencia del LLM.

La combinación de LangGraph y Strands cierra esta brecha. Puede usarlo para construir sistemas donde la orquestación determinista albergue inteligencia dinámica y localizada.

Así es como esta combinación de arquitecturas resuelve desafíos complejos de flujo de trabajo:

Inteligencia a nivel de "nodo": LangGraph define la orquestación de nivel superior para el flujo de trabajo. Los agentes de Strands se colocan en nodos específicos, donde un flujo de trabajo complejo puede requerir un análisis de LLM o navegar por la ambigüedad. Aplican razonamiento autónomo y uso de herramientas sólo cuando esa flexibilidad es estrictamente necesaria.

Aislamiento de contexto con agentes nodales: los agentes monolíticos pierden fácilmente la pista de sus instrucciones. Al colocar agentes de Strands individuales dentro de nodos LangGraph separados, se compartimenta la memoria. Cada agente gestiona su propio contexto hipercentrado y su historial de herramientas, mientras que LangGraph mantiene un estado de sesión estructurado y general que estos diferentes agentes pueden actualizar y consultar de forma independiente.

Orquestación potenciada: en esencia, LangGraph es una herramienta de orquestación y enrutamiento de bajo nivel. Al incorporar Strands, conecta instantáneamente un marco integral de agentes empresariales a su gráfico. Obtiene el enrutamiento sólido de LangGraph junto con las integraciones nativas del Protocolo de contexto de modelo (MCP), controles de dirección, barandillas de seguridad y evaluación de Strands.

En concreto, cada especialista es un nodo LangGraph. El nodo activa un nuevo agente de Strands con su propio sistema, herramientas y contexto aislado. Ejecuta la tarea que asignó el orquestador y devuelve una actualización de estado. LangGraph fusiona esa actualización parcial en el estado compartido desde el que leen los otros nodos. Aquí está el nodo del monitor de seguridad:

async def security_monitor_node(state: AgentState) -> AgentState: """ Agente analista de actividad de un solo día que evalúa el precio, el volumen y las operaciones a nivel de tick """ agente = Agente( name="security_monitor", model=analyst_model, system_prompt=SECURITY_MONITOR_PROMPT, tools=[get_report_list, get_report_schema, run_report], callback_handler=Ninguno, ) # Extrae la tarea de este nodo del estado compartido que completó el orquestador. task = state.get("agent_task_map", {}).get("security_monitor", state["query_text"]) # Strands ejecuta su propio razonamiento + bucle de herramientas; recoger el texto final del agente. trozos =[]async para evento en agent.stream_async(task): if "data" in event: chunks.append(event["data"]) result = "".join(chunks) # Devolver actualizaciones de estado compartidas return { "security_monitor_insights": {"task": tarea, "business_insights": resultado}, "current_agent_index": state.get("current_agent_index", 0) + 1, }

Infraestructura e implementación de AWS con AgentCore

Amazon Bedrock AgentCore proporciona un servicio totalmente administrado para implementar y operar agentes a escala, lo que alivia la administración de la infraestructura y al mismo tiempo ofrece capacidades de nivel de producción.

Implementación en tiempo de ejecución: el tiempo de ejecución de AgentCore, una capacidad de Amazon Bedrock AgentCore, transforma el código del agente local en implementaciones nativas de la nube con una configuración mínima. El servicio es independiente del marco y funciona con LangGraph y Strands desde el primer momento. Proporciona una infraestructura diseñada específicamente para cargas de trabajo de agentes dinámicos, incluidos tiempos de ejecución extendidos para investigaciones de larga duración, ejecución de baja latencia para flujos de trabajo interactivos y escalamiento automático según la demanda.

Implemente su orquestador LangGraph y agentes de Strands utilizando el SDK de Python de AgentCore, una capacidad de Amazon Bedrock AgentCore y el kit de herramientas de inicio. El tiempo de ejecución maneja automáticamente la contenedorización, las redes y el aprovisionamiento informático. AgentCore se encarga del trabajo pesado indiferenciado de la orquestación, el escalado y la gestión de sesiones de contenedores.

# api.py — Punto de entrada del tiempo de ejecución de AgentCore desde bedrock_agentcore.runtime import BedrockAgentCoreApp desde src.agents import Workflow app = BedrockAgentCoreApp() flujo de trabajo = Workflow() @app.entrypoint async def market_surveillance_workflow(payload): """Invocado por AgentCore para cada solicitud. Produce fragmentos de transmisión.""" Prompt = payload.get("prompt") session_id = payload.get("session_id", "default-session") actor_id = payload.get("actor_id", "default-actor") async para fragmento en flujo de trabajo.stream_query( session_id=session_id, Prompt=prompt, actor_id=actor_id ): produce fragmento si __name__ == "__main__": app.run() # Implementar con el kit de herramientas de inicio AgentCore desde bedrock_agentcore_starter_toolkit import Runtime runtime = Runtime() runtime.configure( Entrypoint="api.py", auto_create_execution_role=True, auto_create_ecr=True, requisitos_file="requirements.txt", region="us-east-1", agent_name="market_surveillance_workflow", ) resultado = runtime.launch() print(f"ARN del agente: {result.agent_arn}") # Invocar el agente implementado import boto3, json client = boto3.client("bedrock-agentcore", region_name="us-east-1") respuesta = client.invoke_agent_runtime( agentRuntimeArn=result.agent_arn, qualifier="DEFAULT", payload=json.dumps({ "prompt": "¿Qué causó el aumento del precio de AAPL en ¿11:00 a. m. del 15 de marzo de 2024?", "session_id": "session-001", "actor_id": "analyst-jane", }), ) # AgentCore devuelve una secuencia de eventos enviados por el servidor. Analícelo: para raw en respuesta["response"].iter_lines(): si no raw: continúe line = raw.decode("utf-8") if isinstance(raw, bytes) else raw si no line.startswith("data: "): continúe intente: chunk = json.loads(line[6:]) if isinstance(chunk, str) y chunk.startswith("data: "): chunk = json.loads(chunk[6:]) excepto json.JSONDecodeError: continuar # fragmento con formato incorrecto — omitir, no fallar si isinstance(chunk, dict) y chunk.get("type") == "text": print(chunk["content"], end="")

Integración de memoria: LangGraph se integra con la memoria AgentCore, una capacidad de Amazon Bedrock AgentCore, a través del paquete langgraph-checkpoint-aws con unas pocas líneas de código, lo que proporciona persistencia de puntos de control a corto plazo y recuperación inteligente de memoria a largo plazo.

La clase AgentCoreMemorySaver, utilizada en este ejemplo, maneja objetos de punto de control que contienen mensajes de usuario, respuestas de IA, estado de ejecución de gráficos y metadatos. Después de completar cada nodo, LangGraph guarda automáticamente los puntos de control en la memoria de AgentCore. Obtiene conversaciones con estado y recuperación del flujo de trabajo sin administrar tablas de Amazon DynamoDB ni implementar una lógica de serialización personalizada.

La clase AgentCoreMemoryStore proporciona capacidades de memoria inteligentes donde AgentCore extrae automáticamente información, resúmenes y preferencias de usuario de las conversaciones. Los agentes pueden buscar estos recuerdos en interacciones futuras, para que usted pueda ofrecer experiencias personalizadas que mejoren con el tiempo. Esto aborda el desafío fundamental de la apatridia de los agentes: cada interacción se basa en conocimientos previos en lugar de comenzar de nuevo.

importar boto3, tiempo REGION = "us-east-1" control_client = boto3.client("bedrock-agentcore-control", region_name=REGION) respuesta = control_client.create_memory( name="MarketSurveillanceMemory", descripción="Memoria para el flujo de trabajo de múltiples agentes de vigilancia de mercado.", eventExpiryDuration=90, # días ) MEMORY_ID = respuesta["memoria"]["id"] print(f"ID de memoria: {MEMORY_ID}") # Espere ACTIVO, con un plazo de 10 minutos. La creación normalmente tarda entre 1 y 3 minutos. fecha límite = time.time() + 600 mientras es Verdadero: estado = control_client.get_memory(memoryId=MEMORY_ID)["memoria"]["status"] if status == "ACTIVE": break if status == "FAILED" o time.time() >= fecha límite: elevar RuntimeError(f"La memoria {MEMORY_ID} es {status!r} (esperado ACTIVO)") tiempo.dormir(10)

En la construcción del gráfico:

puntero de control = AgentCoreMemorySaver (MEMORY_ID, nombre_región = REGION) gráfico = flujo de trabajo.compile (punto de control = puntero de control)

Al invocarlo, pase thread_id y actor_id al agente. Estos son identificadores únicos para el usuario y la sesión:

config = { "configurable": { "thread_id": "surveillance-session-001", "actor_id": "analyst-jane", } } respuesta = await graph.ainvoke( {"query_text": "¿Qué corredores estuvieron más activos?"}, config=config,)

Observabilidad y operaciones: AgentCore incluye observabilidad integrada a través de la integración con Amazon CloudWatch y AWS X-Ray, capturando seguimientos de ejecución de agentes, invocaciones de herramientas y métricas de rendimiento. El servicio proporciona paneles para monitorear el comportamiento de los agentes, identificar cuellos de botella y optimizar costos. Combinado con los eventos OpenTelemetry de LangGraph, obtiene una visibilidad integral desde la orquestación del flujo de trabajo de alto nivel hasta llamadas LLM individuales y pasos de razonamiento.

Conclusión

En esta publicación, mostramos cómo construir sistemas de IA multiagente listos para producción combinando la sólida orquestación del flujo de trabajo de LangGraph con las capacidades de razonamiento inteligente de agentes de Strands, implementadas en la infraestructura de AWS.

La arquitectura híbrida que exploramos demuestra cómo LangGraph sobresale en la orquestación a nivel macro (administración de la coordinación de agentes, persistencia del estado y recuperación del flujo de trabajo), mientras que Strands proporciona el motor de razonamiento detallado dentro de nodos individuales. Con esta separación de preocupaciones, puede crear sistemas sofisticados que manejen procesos comerciales complejos y al mismo tiempo mantengan la confiabilidad de la producción a través de una recuperación basada en puntos de control y una observabilidad integral.

Considere ampliar esta arquitectura a otros escenarios de orquestación complejos, como canales de procesamiento de documentos, automatización del servicio al cliente o sistemas de monitoreo de cumplimiento. La naturaleza independiente del modelo de Strands combinada con la gestión de estado de LangGraph hace que este patrón sea particularmente valioso para aplicaciones empresariales que requieren flexibilidad y confiabilidad.

Para profundizar en los detalles técnicos de la implementación y crear usted mismo un agente de análisis de mercado, consulte el repositorio de GitHub.

Sobre los autores

Gleb Geinke

Gleb es arquitecto de aprendizaje profundo en el Centro de innovación de IA generativa de AWS. Gleb colabora directamente con clientes empresariales para diseñar y escalar soluciones de IA generativa transformacionales para desafíos comerciales complejos.

Siddhesh Tiwari

Siddhesh es científico de datos en AWS Professional Services. Trabaja con clientes empresariales para diseñar y ofrecer soluciones de inteligencia artificial generativa, inteligencia artificial y aprendizaje automático en AWS.

Wang Teng Lee

Wang Teng es consultor de entrega de AWS Professional Services y se especializa en inteligencia artificial y soluciones agentes. Trabaja de manera práctica con los clientes durante todo el ciclo de vida de entrega (arquitectura, desarrollo e implementación) y se mantiene a la vanguardia de la IA, explorando cómo las tecnologías emergentes pueden simplificar la forma en que vivimos y trabajamos.

Efrén Faderanga

Efren Faderanga es consultor de entrega en AWS Professional Services y construye la infraestructura que hace que los sistemas de IA agentes estén listos para la producción. Se especializa en orquestación de múltiples agentes, canalizaciones de CI/CD y las bases sin servidor impulsadas por eventos sobre las que se ejecutan los sistemas autónomos. Su trabajo se centra en cerrar la brecha entre la ingeniería de IA y la infraestructura de producción.

Agente de vigilancia de mercado con LangGraph y Strands en AgentCore

ByEquipo de 7 minutos

Hilos: Razonamiento de agentes inteligentes

LangGraph: orquestación de flujo de trabajo macro

Por qué LangGraph y Strands juntos

Infraestructura e implementación de AWS con AgentCore

Conclusión

Sobre los autores

Gleb Geinke

Siddhesh Tiwari

Wang Teng Lee

Efrén Faderanga

By Equipo de 7 minutos

Related Post

CopilotKit lanza OpenBot 1.0: un compañero de trabajo de IA autohospedable y con licencia del MIT con gobernanza integrada

Tres tecnologías del Laboratorio Lincoln reciben los premios R&D 100 2026 | Noticias del MIT

Implemente modelos de IA de vanguardia en AWS Trainium con SGLang-Neuron

You missed

¿Puede la IA predecir qué medicamentos contra el cáncer funcionarán realmente? Orakl Oncology recauda 30 millones de euros para descubrirlo

¿Estás harto de los estrechos asientos económicos? Estas cinco aerolíneas ofrecen una mejor forma de volar « Euro Weekly News

Julianne Hough de DWTS reacciona al comentario de hermana a amante de Derek Hough

Es más probable que tu gato maúlle cuando estás en la habitación, especialmente para comer