Agentes de vigilância de mercado com LangGraph e Strands no AgentCore

Por que sistemas com um único agente não são suficientes

À medida que aplicações de IA avançam além dos chatbots tradicionais, as equipes de engenharia se deparam com um problema concreto: fluxos de trabalho corporativos complexos exigem expertise especializada, decisões dinâmicas e mecanismos robustos de recuperação de falhas — e um único agente raramente dá conta de tudo isso.

O setor financeiro é um exemplo claro. Sistemas de vigilância de mercado precisam coordenar múltiplos agentes especializados para analisar padrões de negociação, investigar atividades suspeitas e produzir relatórios completos, tudo dentro de padrões rígidos de conformidade. Para esse cenário, a AWS publicou um guia técnico detalhando como arquitetar e implantar um sistema multi-agente usando LangGraph e Strands sobre a infraestrutura AWS. O código completo está disponível no GitHub.

A divisão de responsabilidades entre LangGraph e Strands

A arquitetura proposta separa claramente as responsabilidades entre os dois frameworks. O LangGraph cuida da orquestração macro: gerencia o estado compartilhado, define o grafo direcionado de execução e garante a confiabilidade do fluxo. O Strands atua como motor de raciocínio dentro de cada nó individual, executando o loop de análise com o LLM e chamando ferramentas externas conforme necessário.

O LangGraph se destaca no controle granular de execução e no gerenciamento de estado entre agentes. Sua camada de persistência via checkpoints viabiliza recursos críticos para produção, como recuperação de falhas e interações com humano no loop. Já o Strands é agnóstico de modelo — integra-se com diferentes provedores de LLMs — e oferece observabilidade abrangente sobre cada etapa de raciocínio.

Com o lançamento do Amazon Bedrock AgentCore, colocar essa solução em produção ficou consideravelmente mais simples. A combinação dos três forma uma base sólida para sistemas agênticos prontos para ambientes corporativos.

Como o Strands executa o raciocínio nos nós

O Strands Agent opera em um loop de raciocínio que avalia continuamente as saídas das ferramentas e toma decisões com base nos resultados intermediários. Ele inclui gerenciamento de sessão e estado, além de mecanismos para evitar que a janela de contexto estoure.

No agente de vigilância de mercado, a descoberta de dados é separada da recuperação para evitar alucinações e fortalecer a solução contra ataques de injeção de SQL. Ferramentas como get_report_list e get_report_schema localizam os relatórios disponíveis, enquanto run_report constrói a consulta SQL com parâmetros validados e a executa — o LLM nunca escreve SQL diretamente. Veja a definição do agente security_monitor:

from strands import Agent, tool
from strands.models.bedrock import BedrockModel

model = BedrockModel(
    model_id="us.anthropic.claude-sonnet-4-6",
    region_name="us-east-1",
    max_tokens=16000,
    additional_request_fields={
        "thinking": {"type": "adaptive", "budget_tokens": 8000},
    },
    cache_prompt="default",
)

@tool
def get_report_list(agent_name: str) -> str:
    """Load the list of available reports for a specific agent.

    Args:
        agent_name: Name of the agent (e.g. 'security_monitor').

    Returns:
        str: JSON array of report objects with name and description.
    """
    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:
    """Load column definitions for a report so you can build queries.

    Args:
        report_name: Name of the report (e.g. 'TradeActivity').
        query_intent: Description of what data to extract.

    Returns:
        str: JSON object with parameters and column definitions.
    """
    return json.dumps(load_json_report_definition(report_name), indent=2)

@tool
def run_report(
    report_name: str,
    filters: Dict[str, Any],
    limit: Optional[int] = None,
) -> Dict[str, Any]:
    """Run a predefined report. The tool validates every filter against the report's
    schema and builds a parameterised SQL query. The LLM never writes raw SQL,
    so filter values cannot be injected into the query.

    Args:
        report_name: A report from `get_report_list` (e.g. 'TradeActivity').
        filters: Equality filters keyed by column name,
                 e.g. {"symbol": "AAPL", "date": "2024-03-15"}.
        limit: Optional row cap (1..10000).

    Returns:
        dict: {'success': bool, 'data': str (CSV), 'error': str or None}
    """
    schema = load_json_report_definition(report_name)
    allowed_columns = {c["name"] for c in schema["columns"]}

    # Reject any filter field that is not in the report's allowed list of columns.
    unknown = set(filters) - allowed_columns
    if unknown:
        raise ValueError(
            f"Unknown filter field(s) {sorted(unknown)} for {report_name}. "
            f"Allowed: {sorted(allowed_columns)}"
        )

    # Build SQL with named bind parameters.
    where = " AND ".join(f"{field} = :{field}" for field in filters)
    sql = f"SELECT * FROM {schema['reportName']}"
    if where:
        sql += f" WHERE {where}"
    if limit is not None:
        if not isinstance(limit, int) or not 1 <= limit <= 10_000:
            raise ValueError("limit must be an integer in [1, 10000]")
        sql += f" LIMIT {limit}"

    return query_market_data(report_name=report_name, sql=sql, bind=filters)

security_monitor = Agent(
    model=model,
    system_prompt=SECURITY_MONITOR_PROMPT,
    tools=[get_report_list, get_report_schema, run_report],
    name="security_monitor",
)

Como o LangGraph orquestra o fluxo completo

O LangGraph sustenta a orquestração de nível de produção por meio de três pilares:

• Máquinas de estado baseadas em grafos: fluxos de trabalho são modelados como grafos direcionados — nós representam a lógica de cada agente e arestas determinam o caminho de execução, com suporte a ramificação condicional, execução paralela e roteamento dinâmico.
• Persistência de estado via checkpoints: após cada nó executado, o estado completo do fluxo é salvo automaticamente. Isso viabiliza recuperação de falhas, conversas de múltiplos turnos e investigações de longa duração.
• Confiabilidade em produção: o framework inclui estratégias de retry com backoff exponencial e observabilidade via OpenTelemetry, compatível com a maioria das ferramentas de monitoramento.

O trecho abaixo define o grafo de orquestração completo: estado compartilhado, orquestrador que seleciona os agentes especialistas, roteamento condicional entre eles e checkpoints para recuperação:

from typing import TypedDict, Optional, List, Dict, Any
from langgraph.graph import END, StateGraph
from langgraph_checkpoint_aws import AgentCoreMemorySaver

class AgentState(TypedDict):
    query_text: str
    session_id: Optional[str]
    agent_task_map: Optional[Dict[str, str]]
    required_agents: Optional[List[str]]
    current_agent_index: Optional[int]
    # Each specialist writes its insights here
    security_monitor_insights: Optional[Dict[str, Any]]
    broker_monitor_insights: Optional[Dict[str, Any]]
    risk_monitor_insights: Optional[Dict[str, Any]]
    intel_analyst_insights: Optional[Dict[str, Any]]
    synthesizer_insights: Optional[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:
    """Dynamic routing --- walk the required_agents list by index."""
    required = state.get("required_agents", [])
    index = state.get("current_agent_index", 0)
    if not required:
        return END
    if index < len(required):
        return required[index]
    return "synthesizer"

# Build the graph
workflow = StateGraph(AgentState)
workflow.add_node("orchestrator", orchestrator_node)
for name, node_fn in SPECIALIST_NODES.items():
    workflow.add_node(name, node_fn)
workflow.add_node("synthesizer", synthesizer_node)
workflow.set_entry_point("orchestrator")

# Conditional edges --- the orchestrator and every specialist route through
# route_analysts, which can hand off to any specialist or the synthesizer.
ALL_TARGETS = {name: name for name in SPECIALIST_NODES} | {
    "synthesizer": "synthesizer",
    END: END,
}
workflow.add_conditional_edges("orchestrator", route_analysts, ALL_TARGETS)
for name in SPECIALIST_NODES:
    workflow.add_conditional_edges(name, route_analysts, ALL_TARGETS)
workflow.add_edge("synthesizer", END)

# AgentCoreMemorySaver checkpoints the state after every node
checkpointer = AgentCoreMemorySaver(MEMORY_ID, region_name=REGION)
graph = workflow.compile(checkpointer=checkpointer)

Por que usar os dois frameworks juntos

Muitos processos corporativos precisam de fluxos rígidos e previsíveis. Depender exclusivamente do raciocínio não determinístico de um LLM para executar cada passo é um risco real. Ao mesmo tempo, etapas específicas dentro desses fluxos exigem a inteligência e a flexibilidade que só um LLM oferece. A combinação de LangGraph e Strands resolve exatamente essa tensão.

• Raciocínio localizado: o LangGraph define a orquestração de alto nível. Os agentes Strands ficam posicionados nos nós onde o fluxo realmente precisa de análise via LLM — o raciocínio autônomo é aplicado apenas onde é estritamente necessário.
• Contexto isolado por agente: agentes monolíticos perdem facilmente o fio das instruções. Ao posicionar um agente Strands em cada nó separado, a memória fica compartimentalizada. Cada agente gerencia seu próprio contexto, enquanto o LangGraph mantém o estado de sessão estruturado acessível a todos.
• Ecossistema completo: embutir o Strands no LangGraph conecta ao grafo as integrações nativas com o Protocolo de Contexto de Modelo (MCP), controles de direção, guardrails de segurança e avaliação — tudo junto com o roteamento robusto do LangGraph.

Na prática, cada especialista é um nó do LangGraph. O nó instancia um agente Strands com seu próprio prompt de sistema, ferramentas e contexto isolado, executa a tarefa atribuída pelo orquestrador e retorna uma atualização parcial de estado. O LangGraph mescla essa atualização no estado compartilhado que os outros nós leem. Veja o nó do monitor de segurança:

async def security_monitor_node(state: AgentState) -> AgentState:
    """
    Single-day activity analyst agent that assesses price, volume and tick-level trades
    """
    agent = Agent(
        name="security_monitor",
        model=analyst_model,
        system_prompt=SECURITY_MONITOR_PROMPT,
        tools=[get_report_list, get_report_schema, run_report],
        callback_handler=None,
    )

    # Pull this node's task from the shared state the orchestrator populated.
    task = state.get("agent_task_map", {}).get("security_monitor", state["query_text"])

    # Strands runs its own reasoning + tool loop; collect the agent's final text.
    chunks = []
    async for event in agent.stream_async(task):
        if "data" in event:
            chunks.append(event["data"])
    result = "".join(chunks)

    # Return shared state updates
    return {
        "security_monitor_insights": {"task": task, "business_insights": result},
        "current_agent_index": state.get("current_agent_index", 0) + 1,
    }

Implantação e infraestrutura com o AgentCore

O Amazon Bedrock AgentCore oferece um serviço totalmente gerenciado para implantar e operar agentes em escala, eliminando a complexidade de gerenciar infraestrutura enquanto entrega capacidades de nível de produção.

Runtime gerenciado

O AgentCore Runtime transforma código local de agentes em implantações nativas de nuvem com configuração mínima. O serviço é agnóstico de framework, funciona nativamente com LangGraph e Strands, e cuida do trabalho pesado de orquestração de contêineres, escalabilidade automática e gerenciamento de sessões. Veja como implantar o orquestrador usando o SDK Python do AgentCore:

# api.py --- AgentCore runtime entry point
from bedrock_agentcore.runtime import BedrockAgentCoreApp
from src.agents import Workflow

app = BedrockAgentCoreApp()
workflow = Workflow()

@app.entrypoint
async def market_surveillance_workflow(payload):
    """Invoked by AgentCore for each request. Yields streaming chunks."""
    prompt = payload.get("prompt")
    session_id = payload.get("session_id", "default-session")
    actor_id = payload.get("actor_id", "default-actor")

    async for chunk in workflow.stream_query(
        session_id=session_id,
        prompt=prompt,
        actor_id=actor_id
    ):
        yield chunk

if __name__ == "__main__":
    app.run()

# Deploy with the AgentCore starter toolkit
from bedrock_agentcore_starter_toolkit import Runtime

runtime = Runtime()
runtime.configure(
    entrypoint="api.py",
    auto_create_execution_role=True,
    auto_create_ecr=True,
    requirements_file="requirements.txt",
    region="us-east-1",
    agent_name="market_surveillance_workflow",
)
result = runtime.launch()
print(f"Agent ARN: {result.agent_arn}")

# Invoke the deployed agent
import boto3, json

client = boto3.client("bedrock-agentcore", region_name="us-east-1")
response = client.invoke_agent_runtime(
    agentRuntimeArn=result.agent_arn,
    qualifier="DEFAULT",
    payload=json.dumps({
        "prompt": "What caused the AAPL price spike at 11:00 AM on March 15, 2024?",
        "session_id": "session-001",
        "actor_id": "analyst-jane",
    }),
)

# AgentCore returns a server-sent-events stream. Parse it:
for raw in response["response"].iter_lines():
    if not raw:
        continue
    line = raw.decode("utf-8") if isinstance(raw, bytes) else raw
    if not line.startswith("data: "):
        continue
    try:
        chunk = json.loads(line[6:])
        if isinstance(chunk, str) and chunk.startswith("data: "):
            chunk = json.loads(chunk[6:])
    except json.JSONDecodeError:
        continue  # malformed chunk --- skip, don't crash
    if isinstance(chunk, dict) and chunk.get("type") == "text":
        print(chunk["content"], end="")

Memória de curto e longo prazo

O LangGraph se integra com a memória do AgentCore via pacote langgraph-checkpoint-aws. A classe AgentCoreMemorySaver gerencia checkpoints contendo mensagens do usuário, respostas da IA, estado de execução do grafo e metadados — tudo salvo automaticamente após cada nó, sem necessidade de gerenciar tabelas do Amazon DynamoDB ou lógica de serialização customizada.

Já a classe AgentCoreMemoryStore oferece memória inteligente de longo prazo: o AgentCore extrai automaticamente insights, resumos e preferências das conversas, e os agentes podem consultar essas memórias em interações futuras. Isso resolve o problema fundamental da ausência de estado — cada nova interação se baseia no conhecimento acumulado, em vez de começar do zero.

import boto3, time

REGION = "us-east-1"
control_client = boto3.client("bedrock-agentcore-control", region_name=REGION)

response = control_client.create_memory(
    name="MarketSurveillanceMemory",
    description="Memory for market surveillance multi-agent workflow.",
    eventExpiryDuration=90,  # days
)
MEMORY_ID = response["memory"]["id"]
print(f"Memory ID: {MEMORY_ID}")

# Wait for ACTIVE, with a 10-minute deadline. Creation normally takes 1-3 min.
deadline = time.time() + 600
while True:
    status = control_client.get_memory(memoryId=MEMORY_ID)["memory"]["status"]
    if status == "ACTIVE":
        break
    if status == "FAILED" or time.time() >= deadline:
        raise RuntimeError(f"Memory {MEMORY_ID} is {status!r} (expected ACTIVE)")
    time.sleep(10)

# On graph build:
checkpointer = AgentCoreMemorySaver(MEMORY_ID, region_name=REGION)
graph = workflow.compile(checkpointer=checkpointer)

# On invocation, pass thread_id and actor_id to the agent.
# These are unique identifiers for the user and session:
config = {
    "configurable": {
        "thread_id": "surveillance-session-001",
        "actor_id": "analyst-jane",
    }
}
response = await graph.ainvoke(
    {"query_text": "Which brokers were most active?"},
    config=config,
)

Observabilidade integrada

O AgentCore inclui observabilidade nativa via Amazon CloudWatch e AWS X-Ray, capturando traces de execução dos agentes, invocações de ferramentas e métricas de desempenho. Combinado com os eventos OpenTelemetry do LangGraph, é possível ter visibilidade completa — desde a orquestração macro do fluxo até chamadas individuais de LLM e etapas de raciocínio.

Conclusão

A arquitetura apresentada pela AWS combina a orquestração determinística do LangGraph com o raciocínio inteligente do Strands, implantados no AgentCore. O LangGraph cuida da coordenação de agentes, persistência de estado e recuperação de fluxo. O Strands fornece o motor de análise dentro de cada nó. Essa separação de responsabilidades permite construir sistemas sofisticados que lidam com processos corporativos complexos mantendo confiabilidade em produção.

O padrão pode ser estendido para outros cenários — pipelines de processamento de documentos, automação de atendimento ao cliente ou monitoramento de conformidade. A natureza agnóstica de modelo do Strands combinada com o gerenciamento de estado do LangGraph torna essa arquitetura especialmente valiosa para aplicações que exigem tanto flexibilidade quanto previsibilidade. Para explorar os detalhes técnicos e construir um agente de análise de mercado, acesse o repositório no GitHub.

Fonte

Market surveillance agent with LangGraph and Strands on AgentCore (https://aws.amazon.com/blogs/machine-learning/market-surveillance-agent-with-langgraph-and-strands-on-agentcore/)