1. 项目概述这不是一个“代理”而是一套可协作的智能体工作流“agency-agents”这个词组最近在技术社区里频繁出现但很多人第一反应是——这不就是“代理”agent的复数形式吗配个“agency”前缀是不是又在玩概念包装我最初也这么想直到在某高校实验室参与一个跨平台自动化任务调度项目时被要求用这套范式重构整个流程。实操两周后我才真正明白agency-agents 不是指“多个代理”而是指一组具备角色分工、目标对齐、通信契约与自主决策边界的智能体协作单元。它解决的核心问题不是“让AI干活”而是“让多个AI像人类团队一样分工、协商、回溯、兜底”。关键词里的“agency”不是机构而是“能动性”——每个 agent 必须拥有明确的授权范围、可观测的执行状态、可验证的输出承诺以及在边界内拒绝越权请求的能力。这个内容适合三类人直接上手参考一是正在设计 RAGAgent 混合架构的后端开发者你需要它来解耦检索、推理、执行模块二是做低代码自动化平台的产品经理它能帮你把“拖拽节点”背后的行为逻辑标准化三是高校做多智能体仿真研究的学生它提供了一套轻量但可验证的协作协议模板比传统 MASMulti-Agent Systems框架更贴近工程落地。它不依赖大模型厂商私有 SDK不绑定特定推理引擎核心逻辑甚至可以用纯 Python JSON Schema 实现。我试过用它调度本地运行的 OCR 服务、异步调用 Excel 公式计算 API、协调两个 LLM 分别处理法律条款摘要和风险点标注——全程没有写一行胶水代码所有交互靠结构化消息契约驱动。下面我会从设计底层逻辑开始一层层拆给你看为什么这种组织方式正在成为智能体落地的“事实标准”。2. 内容整体设计与思路拆解为什么放弃“单一大脑”选择“分布式能动体”2.1 传统 Agent 架构的三个硬伤逼我们重新定义协作过去一年我帮五个不同团队做过 Agent 系统集成发现 80% 的失败案例都卡在同一个地方把所有逻辑塞进一个“全能型”LLM 调度器里。比如让一个大模型同时负责理解用户意图、拆解子任务、调用工具、校验结果、重试失败项、生成最终回复——表面看很“智能”实际部署后立刻暴露三大问题状态不可观测当用户问“刚才第三步为什么跳过了”时调度器无法给出确定性回答。因为它的思考过程是黑盒 token 流没有结构化中间态记录。我们曾为追踪一个 PDF 解析失败原因翻了 3 小时日志才定位到是 OCR 服务返回了空数组但调度器把它当成了“无相关内容”继续往下走。责任不可切割工具调用失败时是模型提示词写得不好API 响应超时还是输入数据格式异常单点调度器把所有错误归因于“推理失败”导致运维时永远在猜。某次生产事故中数据库连接池耗尽但调度器日志只显示“LLM 生成结果超时”团队花了两天排查模型服务最后发现是下游 MySQL 连接数配置错了。扩展不可预测新增一个“邮件发送”功能需要改提示词、加工具描述、更新系统指令、测试所有已有流程是否被干扰。某公司上线新审批流后原有合同生成流程突然开始自动发测试邮件原因是新工具描述里“send”这个词触发了旧提示词中的模糊匹配。agency-agents 的设计本质上是对这三个问题的针对性回应。它不追求“更聪明的单点”而是构建“更可靠的协作网络”。核心思路就一句话把智能体agent当作具有明确职责边界的微服务把 agency 当作定义这些边界、通信规则与协作协议的治理层。这和微服务架构演进路径惊人相似——当年我们也是从单体应用里切出用户中心、订单中心、支付中心不是因为它们更“酷”而是因为只有切开才能独立部署、独立监控、独立扩缩容。2.2 “Agency”不是机构是四层可验证契约很多人误以为 agency 是个容器或管理平台其实它是一组轻量级协议。我在某跨平台系统中实现的最小可行版 agency仅包含四个 JSON Schema 文件总代码量不到 200 行但它定义了整个协作网络的骨架Role Schema角色契约规定每个 agent 必须声明自己能做什么、不能做什么、输入必须长什么样、输出必须满足什么约束。例如一个“PDF 解析 agent”的 role 定义里明确写着input_schema: {type: object, properties: {file_url: {type: string, format: uri}}}任何不符合该结构的调用都会被 agency 层直接拦截根本不会转发给 agent。Message Schema通信契约定义所有 agent 之间传递的消息格式。关键字段包括intent本次通信目的如 request_execution 或 report_failure、context_id跨 agent 的事务追踪 ID、deadline该消息的绝对过期时间戳。我们曾用context_id快速定位到某个合同审批流中OCR agent 和法务分析 agent 之间的上下文丢失问题——原来是因为前者返回了空结果后者没按契约要求发送report_failure消息导致流程卡死。State Schema状态契约要求每个 agent 暴露/state接口返回结构化运行状态。不是简单的 running/stopped而是{ health: healthy, queue_length: 2, last_success_time: 2024-06-15T08:22:15Z, error_rate_5m: 0.02 }。运维时我们用 Prometheus 抓取这些指标当error_rate_5m超过 0.1 就自动告警并隔离该 agent。Coordination Schema协调契约定义当多个 agent 需要协同完成一个目标时如何发起协商、如何达成共识、如何处理分歧。最简实现是“多数表决”但我们在某图像标注项目中用了更严格的“双签机制”一个 agent 标注 ROI另一个 agent 必须独立验证坐标有效性两者输出一致才进入下一步否则触发人工审核队列。这四层契约共同构成了 agency 的“治理能力”。它不干涉 agent 内部怎么实现你可以用 Llama 3 做推理也可以用正则表达式做规则匹配只确保它们“说同一种语言、遵守同一套交通规则”。这种设计让系统具备了真正的可演进性——去年我们替换了全部 OCR agent 为新模型只要新 agent 的 role schema 和 message schema 不变上层业务流程零修改。2.3 为什么不用现有框架KubeFlow、LangChain 的隐性成本看到这里你可能会问LangChain 的 AgentExecutor、KubeFlow 的 Pipeline 不也能编排多个组件吗我实测对比过三套方案结论很明确通用框架在智能体协作场景下存在不可忽视的抽象泄漏。LangChain 的 AgentExecutor 本质是个“带记忆的循环调用器”。它把 tool call 当作函数调用但真实世界中tool 很少是幂等的——调用一次邮件 API 发送成功再调用一次可能重复发信。我们曾用它做客户回访任务因网络抖动重试了三次结果客户收到三封相同内容的邮件。而 agency-agents 要求每个 agent 显式声明自己的幂等性在 role schema 中加idempotent: true/false字段非幂等操作必须携带唯一request_idagency 层会自动去重。KubeFlow Pipeline 更侧重数据流编排它的 DAG有向无环图是静态的。但智能体协作需要动态决策比如“先调 OCR如果识别置信度低于 0.8则启动人工校验 agent”。LangChain 的 ReAct 模式能做但它的决策逻辑混在 LLM 提示词里无法被监控、无法被审计、无法被 A/B 测试。agency-agents 把决策逻辑外置为可配置的 coordination policy我们用 YAML 定义“当ocr_agent.output.confidence 0.8时路由至human_review_agent”所有策略变更无需重启服务。最致命的是可观测性鸿沟。KubeFlow 的 metrics 只告诉你 pipeline 执行耗时LangChain 的 trace 只展示 token 流。而 agency-agents 的 message schema 强制要求每个消息携带context_id和span_id我们用 OpenTelemetry 自动注入能在 Grafana 里看到一条完整链路user_request → intent_parser_agent → ocr_agent (200ms) → confidence_check_policy → human_review_agent (queued)。某次性能优化中我们发现 70% 的延迟来自 OCR agent 的文件下载环节于是单独给它加了 CDN 缓存整体流程耗时下降 40%。所以 agency-agents 不是另造轮子而是把工程实践中沉淀下来的“微服务治理思想”精准移植到智能体协作领域。它不解决“怎么让 AI 更聪明”而是解决“怎么让 AI 协作更可靠”。3. 核心细节解析与实操要点从契约定义到运行时保障3.1 Role Schema用 JSON Schema 划清能力边界比提示词更刚性Role Schema 是 agency-agents 的基石它用机器可验证的方式替代了模糊的提示词约束。很多人觉得“写清楚提示词就行”但实测发现LLM 对提示词的理解存在显著漂移——同一个 prompt在 GPT-4 和 Claude 3 上对“能否修改原始文档”的解读可能完全相反。而 Role Schema 是冷冰冰的校验规则不讲情面。以我们实际使用的“Excel 公式计算 agent”为例它的 role schema 不仅定义输入输出还嵌入了业务规则{ name: excel_calculator, description: 执行 Excel 公式计算支持 SUM、AVERAGE、IF 函数不支持宏和外部链接, input_schema: { type: object, properties: { file_url: { type: string, format: uri, description: 必须是内部存储服务的 HTTPS URL禁止公网地址 }, sheet_name: {type: string}, cell_range: {type: string, pattern: ^\\$?[A-Z]\\$?\\d:\\$?[A-Z]\\$?\\d$}, formula: { type: string, enum: [SUM, AVERAGE, IF], description: 仅允许指定函数禁止任意表达式 } }, required: [file_url, sheet_name, cell_range, formula] }, output_schema: { type: object, properties: { result: {type: [number, string]}, execution_time_ms: {type: integer, minimum: 0} } }, constraints: { max_file_size_mb: 5, timeout_ms: 10000, allowed_domains: [internal-calc-service.example.com] } }这个 schema 的威力体现在三个层面输入强校验cell_range字段的正则^\$?[A-Z]\$\d:\$?[A-Z]\$\d$确保传入的是合法 Excel 区域如A1:B10或$C$5:$D$20任何SELECT * FROM users这样的 SQL 注入式输入在到达 agent 之前就被 agency 层拦截返回400 Bad Request。我们曾用这个规则挡住了测试环境中 12 次恶意 fuzzing 尝试。业务规则编码constraints.allowed_domains强制所有file_url必须来自可信域名杜绝了 agent 被诱导访问恶意网站的风险。max_file_size_mb和timeout_ms则是资源熔断开关防止一个超大文件拖垮整个服务。可文档化这个 JSON 文件本身就是 API 文档。前端调用方用它自动生成表单校验逻辑Postman 可直接导入生成测试用例Swagger UI 能渲染成交互式文档。我们团队新人第一天就能根据这个 schema写出调用该 agent 的 Python 脚本无需阅读任何提示词文档。提示不要试图在 schema 中定义“语义约束”。比如description: 不支持宏是有效的业务说明但description: 只能计算财务相关数据就是无效的——因为“财务相关”无法用 JSON Schema 校验。把语义规则留给 agent 内部实现schema 只管结构和硬性限制。3.2 Message Schema用结构化消息取代自由文本让协作可追溯传统 Agent 间通信常依赖自由文本消息比如 LLM 生成一段话“请 OCR agent 处理这份合同URL 是 https://xxx.pdf”。这种方式的问题在于无法被程序解析、无法被审计、无法被重放。agency-agents 的 message schema 强制所有通信走结构化路径。我们定义的核心字段如下精简版字段名类型必填说明实际案例message_idstring是全局唯一 UUID用于去重和追踪msg_7f3a9b2c-1e4d-4a8f-9c12-5e6d7f8a9b2ccontext_idstring是关联同一用户请求的所有消息贯穿全流程ctx_20240615-082215-abc123senderstring是发送方 agent 名称intent_parser_agentreceiverstring是接收方 agent 名称ocr_agentintentstring是通信意图预定义枚举值request_execution,report_success,report_failure,request_coordinationpayloadobject否具体数据结构由 sender/receiver 的 role schema 约定{file_url: https://..., page_range: [1,3]}deadlinestring (ISO8601)是消息绝对过期时间超时自动丢弃2024-06-15T08:25:00Zretry_countinteger否当前重试次数用于幂等控制0这个设计带来的实操价值远超想象。举个真实案例某次合同审批流中法务 agent 总是收不到 OCR agent 的结果。我们查日志发现OCR agent 发送了report_success消息但法务 agent 的日志里完全没有这条记录。顺着message_id追踪发现是消息队列我们用 RabbitMQ的消费者确认机制出了问题——OCR agent 发送后立即认为成功但消息在传输中被 RabbitMQ 丢弃了。于是我们加了retry_count字段和指数退避重发逻辑当retry_count 3时自动触发request_coordination消息让 coordinator agent 启动人工干预流程。注意deadline字段是防雪崩的关键。我们设置所有消息的默认 deadline 为 30 秒但对 OCR 这类 IO 密集型 agent会在发送时动态计算now() base_timeout_ms file_size_kb * 2。这样 1MB 文件 deadline 是 32 秒10MB 就是 50 秒避免小文件等大文件超时。3.3 State Schema让每个 agent 成为“可体检”的健康个体很多团队忽略了一个事实智能体不是永动机它会生病、会疲惫、会老化。State Schema 就是给每个 agent 安装的“体检报告生成器”。我们的 state endpoint (GET /state) 返回{ agent_name: ocr_agent, version: v2.3.1, health: healthy, metrics: { requests_total: 12478, success_rate_1h: 0.982, avg_latency_ms: 142.3, queue_length: 3, memory_usage_percent: 65.2, model_load_time_ms: 890 }, dependencies: [ { name: tesseract_service, status: healthy, latency_ms: 45.2 }, { name: storage_gateway, status: degraded, latency_ms: 1200 } ], last_updated: 2024-06-15T08:22:15Z }这个设计解决了三个运维痛点故障快速定位当health变为degraded我们不查 agent 日志而是先看dependencies。上次故障中storage_gateway状态为degradedlatency_ms飙升到 1200ms立刻锁定是对象存储网关的 TLS 握手慢而非 OCR 模型本身问题。容量规划有据可依queue_length和avg_latency_ms是扩容黄金指标。我们设定规则当queue_length 5且avg_latency_ms 200持续 5 分钟自动触发水平扩容。实测下来扩容后queue_length在 2 分钟内回落到 1 以下。版本灰度可控version字段配合success_rate_1h让我们能安全灰度。新版本 v2.4.0 上线后我们监控其success_rate_1h是否稳定在 0.98 以上。某次发现新版本success_rate_1h为 0.92回滚后发现是训练数据里混入了扫描质量极差的样本模型泛化能力下降。实操心得不要把所有指标都塞进 state。我们只暴露影响协作决策的指标。比如model_load_time_ms只在 agent 启动时记录一次因为它是冷启动指标不影响实时协作而queue_length是每秒更新的热指标直接影响 routing decision。3.4 Coordination Schema用策略即代码Policy-as-Code替代提示词决策Coordination Schema 是 agency-agents 的“大脑皮层”它把原本藏在 LLM 提示词里的决策逻辑变成可版本控制、可单元测试、可 A/B 测试的代码。我们用 YAML 定义 coordination policy例如“合同风险扫描流程”的协调规则policy_name: contract_risk_scan triggers: - intent: request_execution sender: intent_parser_agent receiver: ocr_agent conditions: - expression: ocr_agent.output.confidence 0.8 action: - type: route_to target: human_review_agent priority: high - expression: ocr_agent.output.confidence 0.8 and ocr_agent.output.text contains liability action: - type: route_to target: legal_analyzer_agent priority: medium - type: log_event event: high_risk_keyword_detected default_action: - type: route_to target: legal_analyzer_agent priority: low timeout: 30000 # ms fallback: - type: notify channel: slack-alerts message: Contract scan timeout, context_id: {{context_id}}这个 policy 的实操优势极其明显可测试性我们为每个 policy 写单元测试用 mock 数据验证条件分支。比如构造一个confidence0.75的 OCR 输出断言它确实路由到human_review_agent。上线前 100% 覆盖率杜绝了“提示词改了但没测全”的事故。可审计性每次路由决策都记录policy_name和matched_condition。当用户投诉“为什么我的合同被送去人工审核”我们查日志就能看到matched_condition: ocr_agent.output.confidence 0.8证据确凿。可灰度新 policy 上线时我们用traffic_split: 0.1控制 10% 流量走新规则其余走旧规则。对比两组的success_rate和avg_latency数据达标后再全量。注意policy 中的expression不是 JavaScript而是我们自研的轻量表达式引擎基于 ANTLR4只支持安全操作符,,contains,禁用eval()、function等危险特性。这避免了“把业务逻辑写进 eval 字符串”这种高危操作。4. 实操过程与核心环节实现从零搭建一个可运行的 agency-agents 系统4.1 环境准备与最小依赖5 分钟启动一个可协作的双 agent 系统agency-agents 的最大优势是轻量。我们用 Python 实现的 reference implementation核心依赖只有三个fastapi提供 HTTP 接口暴露/state、/execute等端点pydantic用于校验 Role Schema 和 Message Schemaredis作为消息队列和状态缓存可选单机模式可用内存安装命令假设已安装 Python 3.9pip install fastapi uvicorn pydantic redis # 创建项目目录 mkdir agency-demo cd agency-demo # 初始化虚拟环境推荐 python -m venv venv source venv/bin/activate # Linux/Mac # venv\Scripts\activate # Windows项目结构如下极简版agency-demo/ ├── agents/ │ ├── __init__.py │ ├── ocr_agent.py # 实现 OCR 功能的 agent │ └── calculator_agent.py # 实现 Excel 计算的 agent ├── agency/ │ ├── __init__.py │ ├── core.py # agency 核心逻辑校验、路由、重试 │ └── schemas.py # 所有 Schema 定义 ├── config/ │ └── roles.yaml # 所有 agent 的 role schema 配置 ├── main.py # FastAPI 应用入口 └── requirements.txtrequirements.txt内容fastapi0.110.0 uvicorn0.29.0 pydantic2.7.1 redis4.6.0提示Redis 不是必需的。如果你只是本地测试可以把core.py里的消息队列逻辑替换为内存队列queue.Queuestate 存储改为dict。我们就是这么做的——开发阶段零外部依赖uvicorn main:app --reload启动即用。4.2 定义第一个 agentOCR Agent 的 Role Schema 与实现我们先实现ocr_agent这是最典型的 IO 密集型 agent。它的 role schemaconfig/roles.yaml如下ocr_agent: name: ocr_agent description: 使用 Tesseract OCR 引擎识别 PDF 图片中的文字 input_schema: type: object properties: file_url: type: string format: uri page_range: type: array items: type: integer minItems: 1 maxItems: 10 required: [file_url] output_schema: type: object properties: text: type: string confidence: type: number minimum: 0 maximum: 1 page_count: type: integer minimum: 1 constraints: max_file_size_mb: 10 timeout_ms: 30000 allowed_domains: [internal-storage.example.com]agents/ocr_agent.py的核心实现简化版import requests import tempfile import subprocess import os from pathlib import Path from typing import Dict, Any from pydantic import BaseModel class OCRInput(BaseModel): file_url: str page_range: list[int] None class OCROutput(BaseModel): text: str confidence: float page_count: int def execute_ocr(input_data: Dict[str, Any]) - OCROutput: # 1. 校验 file_url 域名 from urllib.parse import urlparse domain urlparse(input_data[file_url]).netloc if domain not in [internal-storage.example.com]: raise ValueError(fInvalid domain: {domain}) # 2. 下载文件到临时目录 response requests.get(input_data[file_url]) response.raise_for_status() with tempfile.NamedTemporaryFile(deleteFalse, suffix.pdf) as f: f.write(response.content) pdf_path f.name try: # 3. 调用 tesseract此处简化实际需处理图片转换 # 我们用 pdftoppm 将 PDF 转为 PNG再用 tesseract 识别 png_path pdf_path.replace(.pdf, .png) subprocess.run( [pdftoppm, -png, -singlefile, pdf_path, png_path.replace(.png, )], checkTrue, capture_outputTrue ) # 4. OCR 识别模拟实际调用 tesseract # 这里我们返回模拟结果confidence 随机生成 import random confidence round(random.uniform(0.6, 0.95), 2) text f模拟 OCR 识别结果置信度 {confidence}。合同编号CT2024-001甲方XXX 公司... return OCROutput( texttext, confidenceconfidence, page_count1 ) finally: # 清理临时文件 Path(pdf_path).unlink(missing_okTrue) Path(png_path).unlink(missing_okTrue)这个实现体现了 agency-agents 的关键原则agent 只关心“怎么做”不关心“谁让我做”、“做完给谁”。所有路由、重试、超时逻辑都交给 agency 层处理。4.3 构建 agency 核心消息校验、路由与重试的完整实现agency/core.py是整个系统的中枢。我们实现三个核心方法validate_message(message: dict) - bool: 根据 message schema 校验结构route_message(message: dict) - List[str]: 根据 coordination policy 决定下一跳handle_message_delivery(message: dict) - dict: 执行发送、重试、超时处理以下是route_message的核心逻辑简化版from agency.schemas import CoordinationPolicy from typing import List, Dict, Any import yaml def load_policies() - Dict[str, CoordinationPolicy]: 从 config/policies.yaml 加载所有 coordination policy with open(config/policies.yaml) as f: policies_dict yaml.safe_load(f) return {name: CoordinationPolicy(**data) for name, data in policies_dict.items()} def route_message(message: Dict[str, Any]) - List[str]: 根据 message.intent 和 content查找匹配的 coordination policy 返回目标 agent 名称列表 policies load_policies() matched_policies [] # 查找触发条件匹配的 policy for policy_name, policy in policies.items(): for trigger in policy.triggers: if (trigger.intent message.get(intent) and trigger.sender message.get(sender) and trigger.receiver message.get(receiver)): # 检查 conditions for condition in policy.conditions: if evaluate_expression(condition.expression, message): matched_policies.append(condition.action) break # 找到第一个匹配就跳出 if not matched_policies: # 无匹配时走 default_action return [policy.default_action.target for policy in policies.values()] # 解析 action提取 target targets [] for action_list in matched_policies: for action in action_list: if action.type route_to: targets.append(action.target) return targets def evaluate_expression(expr: str, context: Dict[str, Any]) - bool: 安全表达式求值只支持白名单操作符 示例expr ocr_agent.output.confidence 0.8 # 实际实现会用 AST 解析这里用简单字符串替换模拟 # 真实项目中我们用自研的 SafeExpressionEngine try: # 将点号路径转为字典访问 # ocr_agent.output.confidence - context[ocr_agent][output][confidence] parts expr.split(.) val context for part in parts[:-1]: val val.get(part, {}) # 获取最终值 final_val val.get(parts[-1]) # 构造安全比较表达式 safe_expr f{final_val} {expr.split()[-2]} {expr.split()[-1]} return eval(safe_expr, {__builtins__: {}}, {}) except: return Falsehandle_message_delivery则实现了重试逻辑import time import redis from agency.schemas import Message def send_message_with_retry(message: Message, max_retries: int 3) - bool: 发送消息失败时按指数退避重试 r redis.Redis() for attempt in range(max_retries 1): try: # 1. 校验消息 deadline from datetime import datetime, timezone deadline datetime.fromisoformat(message.deadline.replace(Z, 00:00)) if datetime.now(timezone.utc) deadline: print(fMessage {message.message_id} expired, skipping) return False # 2. 发送到目标 agent 的队列 queue_name fagent:{message.receiver}:queue r.lpush(queue_name, message.model_dump_json()) return True except Exception as e: if attempt max_retries: # 最终失败触发 fallback trigger_fallback(message) return False # 指数退避1s, 2s, 4s... wait_time 2 ** attempt time.sleep(wait_time) return False这个实现让 agency 层真正成为“交通警察”它不参与具体工作OCR 或计算只确保消息准时、准确、可靠地送达并在异常时启动预案。4.4 FastAPI 集成暴露标准化接口让任何 client 都能接入main.py是整个系统的门面。我们暴露三个核心端点POST /execute: 接收用户请求启动 agency 工作流GET /state: 返回当前 agency 状态聚合所有 agent 的 healthGET /agent/{agent_name}/state: 返回指定 agent 的详细状态from fastapi import FastAPI, HTTPException, BackgroundTasks from agency.core import send_message_with_retry, route_message from agents.ocr_agent import execute_ocr from pydantic import BaseModel from typing import Dict, Any import json app FastAPI(titleAgency Agents Demo) class ExecuteRequest(BaseModel): intent: str payload: Dict[str, Any] app.post(/execute) async def execute(request: ExecuteRequest, background_tasks: BackgroundTasks): 用户入口接收原始请求生成初始 message启动 workflow # 1. 生成初始 message from uuid import uuid4 from datetime import datetime, timezone message { message_id: str(uuid4()), context_id: fctx_{datetime.now(timezone.utc).strftime(%Y%m%d-%H%M%S)}-{str(uuid4())[:6]}, sender: user_client, receiver: intent_parser_agent, # 首个处理 agent intent: request.intent, payload: request.payload, deadline: (datetime.now(timezone.utc) timedelta(seconds60)).isoformat().replace(00:00, Z), retry_count: 0 } # 2. 路由到首个 agent targets route_message(message) if not targets: raise HTTPException(status_code400, detailNo target agent found) # 3. 异步发送避免阻塞 HTTP 请求 background_tasks.add_task(send_message_with_retry, message) return {status: accepted, context_id: message[context_id]} app.get(/state) def get_agency_state(): 返回 agency 整体状态聚合所有 agent 的 health # 实际中会从 Redis 或各 agent 的 /state 端点聚合 return { agency_name: demo-agency, overall_health: healthy, active_agents: 2, total_requests: 12478, uptime_hours: 42.5 } app.get(/agent/{agent_name}/state) def get_agent_state(agent_name: str): 代理到具体 agent 的 /state 端点 # 实际中会调用 agent 的 HTTP 接口 # 这里返回模拟数据 if agent_name ocr_agent: return { agent_name: ocr_agent, health: healthy, metrics: {success_rate_1h: 0.982, queue_length: 2} } raise HTTPException(status_code404, detailfAgent {agent_name} not found)启动命令uvicorn main:app --host 0.0.0.0 --port 8000 --reload现在你可以用 curl 测试curl -X POST