Market surveillance agent with LangGraph and Strands on AgentCore

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:
    """加载特定代理可用报告的列表。

    参数:
        agent_name: 代理名称（例如 'security_monitor'）。

    返回:
        str: 包含报告名称和描述的报告对象 JSON 数组。
    """
    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:
    """加载报告的列定义，以便构建查询。

    参数:
        report_name: 报告名称（例如 'TradeActivity'）。
        query_intent: 要提取数据的描述。

    返回:
        str: 包含参数和列定义的 JSON 对象。
    """
    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]:
    """运行预定义报告。该工具会验证每个过滤条件与报告模式的匹配情况，并构建参数化 SQL 查询。LLM 从不编写原始 SQL，因此过滤值无法注入查询。

    参数:
        report_name: 来自 `get_report_list` 的报告（例如 'TradeActivity'）。
        filters: 以列名为键的等值过滤器，例如
            {"symbol": "AAPL", "date": "2024-03-15"}。
        limit: 可选的行数上限（1..10000）。

    返回:
        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"]}

    # 拒绝任何不在报告允许列列表中的过滤字段。
    unknown = set(filters) - allowed_columns
    if unknown:
        raise ValueError(
            f"未知的过滤字段 {sorted(unknown)} for {report_name}。 "
            f"允许的字段: {sorted(allowed_columns)}"
        )

    # 使用命名绑定参数构建 SQL。
    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 必须是 [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",
)

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]
    # 每个专业代理在此处写入其见解
    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:
    """动态路由 --- 按索引遍历 required_agents 列表."""
    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"

# 构建图
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")

市场分析代理的 LangGraph 工作流编排

## 为什么选择 LangGraph 与 Strands 结合

许多企业用例必须基于严格预定义的工作流。单纯依赖大语言模型的非确定性来执行正确步骤存在风险。然而，这些工作流中的特定步骤确实需要大语言模型的敏捷推理能力。

LangGraph 与 Strands 的结合弥合了这一差距。您可以使用它构建系统，其中确定性编排包含本地化的动态智能。

这种架构组合如何解决复杂的工作流挑战：

**节点级智能**：LangGraph 定义工作流的高层编排。Strands 代理被放置在特定节点中，复杂工作流可能需要大语言模型分析或处理模糊性。它们仅在严格需要灵活性的地方应用自主推理和工具使用。

**节点代理的上下文隔离**：单体代理很容易丢失指令。通过将独立的 Strands 代理放置在 LangGraph 的不同节点中，您可以隔离内存。每个代理管理自己的高度专注的上下文和工具历史，而 LangGraph 保持结构化的总体会话状态，这些不同代理可以独立更新和引用。

**增强编排能力**：本质上，LangGraph 是一个底层路由和编排工具。通过嵌入 Strands，您可以立即将全面的企业代理框架插入到您的图中。您将获得 LangGraph 的强大路由能力，以及 Strands 本地的模型上下文协议（MCP）集成、控制转向、安全防护和评估功能。

具体来说，每个专家都是一个 LangGraph 节点。该节点启动一个带有独立系统提示、工具和隔离上下文的新 Strands 代理。它执行协调器分配的任务并返回状态更新。LangGraph 将这些部分更新合并到其他节点读取的共享状态中。以下是安全监控节点的示例：

## 使用 AgentCore 的 AWS 基础设施和部署

Amazon Bedrock AgentCore 提供了全托管服务，用于大规模部署和运行代理，既能减轻基础设施管理负担，又能提供生产级功能。

运行时部署：作为 Amazon Bedrock AgentCore 的一项功能，AgentCore 运行时通过最小的配置即可将本地代理代码转换为云原生部署。该服务与框架无关，开箱即用支持 LangGraph 和 Strands。它为动态代理工作负载提供专用基础设施，包括用于长期调查的扩展运行时、用于交互式工作流的低延迟执行，以及基于需求的自动扩展。

使用 AgentCore Python SDK（Amazon Bedrock AgentCore 的一项功能）和入门工具包部署您的 LangGraph 编排器和 Strands 代理。运行时会自动处理容器化、网络和计算资源分配。AgentCore 负责容器编排、扩展和会话管理等基础性繁重工作。

# AgentCore 返回一个服务器发送事件流。解析方式如下：
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  # 数据块格式错误 --- 跳过，不中断
    if isinstance(chunk, dict) and chunk.get("type") == "text":
        print(chunk["content"], end="")

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="用于市场监控多代理工作流的记忆。",
    eventExpiryDuration=90,  # 天
)
MEMORY_ID = response["memory"]["id"]
print(f"记忆 ID: {MEMORY_ID}")

# 等待状态变为 ACTIVE，设定 10 分钟截止时间。创建通常需要 1-3 分钟。
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_ID} 状态为 {status!r}（预期为 ACTIVE）")
    time.sleep(10)

checkpointer = AgentCoreMemorySaver(MEMORY_ID, region_name=REGION)
graph = workflow.compile(checkpointer=checkpointer)

config = {
    "configurable": {
        "thread_id": "surveillance-session-001",
        "actor_id": "analyst-jane",
    }
}

response = await graph.ainvoke(
    {"query_text": "哪些经纪人最活跃？"},
    config=config,
)