agency-agents：将能动性作为一级设计要素的智能体架构-平芜编程栈

1. 项目概述：从“agency-agents”这个词组看智能体系统的设计本质

“agency-agents”不是某个具体软件、平台或开源库的官方名称，而是一个高度凝练的复合概念词——它把agency（能动性）和agents（智能体）两个核心术语强行粘合在一起，形成一种带有强烈设计意图的命名方式。我在过去三年里参与过多个跨团队协作的智能体系统落地项目，从某高校实验室的教育辅助Demo，到某公司内部的知识协同平台原型，再到模拟项目X中的多角色任务编排系统，凡是用到这个词组的地方，背后几乎都指向同一个问题：我们不再满足于让AI“执行指令”，而是要让它“理解目标、权衡路径、主动协调、承担后果”。这四个关键词，就是“agency-agents”的真实落点。

这个词组在近期技术社区讨论中高频出现，并非偶然。它本质上是对当前LLM应用层瓶颈的一次集体反思：当提示工程走到极限、RAG召回率卡在82%、函数调用链超过7层就开始不可控时，开发者开始本能地回溯更底层的架构逻辑——智能体是否真的具备“代理资格”？它有没有自己的目标函数？能不能在无人干预下判断“现在该不该调用工具”？会不会因为上游数据延迟而主动降级策略？这些不是功能清单里的勾选项，而是系统级的契约责任。所以，“agency-agents”不是语法错误，而是一种刻意为之的语义挤压：把本该属于人类委托关系（agency）的权责，明确压给技术实体（agents）。它适合两类人深度参考：一类是正在设计复杂工作流的后端/架构师，需要厘清智能体之间的契约边界；另一类是刚接触Agent开发的新手，容易把“能调API”误认为“有能动性”，这个词组就是一面照妖镜。

我试过用纯LangChain搭一个带记忆的客服bot，表面看它能记住用户上句话提过“快递单号”，但一旦用户突然问“那我昨天投诉的赔偿进度呢”，它就卡住——因为它没有内置的“问题归属判断模块”，也不具备“主动触发跨系统查询”的权限意识。而真正按agency-agents思路设计的系统，会在初始化阶段就为每个agent分配三样东西：目标权重表（goal weight matrix）、资源访问契约（resource access covenant）、失败回退协议（fallback protocol）。这不是炫技，而是把模糊的“智能”拆解成可配置、可审计、可熔断的工程要素。接下来我会从设计逻辑、核心细节、实操实现和典型故障四个维度，把这套思路掰开揉碎讲清楚。你不需要会写Python装饰器，但得明白为什么agent的retry机制不能只靠time.sleep(2)。

2. 内容整体设计与思路拆解：为什么必须把“能动性”作为一级设计要素

2.1 传统Agent架构的隐性缺陷：工具链思维 vs 目标链思维

市面上90%的Agent教程，教的其实是“高级版脚本编写”：定义几个function，写好tool description，丢进LLM的function calling接口，再套个while循环做迭代。这种模式在演示场景很惊艳，但一进真实业务就露馅。我参与过某跨平台系统的POC验证，客户要求agent完成“根据销售报表异常值，自动定位数据源、联系责任人、生成整改建议并同步至OA”。我们按标准流程搭完，结果在第三步就崩了——agent查到数据库字段名是sales_amt_q3，但OA系统要求的字段ID却是Q3_SALES_VALUE，它不会主动做映射，更不会因为映射失败就降级为发送原始SQL截图。问题出在哪？在于整个设计默认agent是“被动工具调用者”，而非“主动目标分解者”。

提示：传统Agent框架（如早期LangChain AgentExecutor）的致命假设是“所有工具都已完备且语义对齐”。但现实里，工具是分阶段上线的，A团队的API文档还没更新，B团队的权限模型刚重构，C团队的错误码含义和上周都不一样。这时候还让agent硬着头皮调用，等于逼一个没驾照的人开改装车。

agency-agents的设计起点，恰恰是承认这种不完备性。它的核心思路是：把目标（goal）作为第一公民，把工具（tool）降级为可选资源。这意味着系统启动时，首先加载的不是tools列表，而是goal schema——一个JSON Schema定义的目标结构，包含：

primary_objective（主目标，不可拆分）
success_criteria（成功判定规则，支持布尔表达式）
resource_constraints（资源限制，如“最多调用3次外部API”）
fallback_strategies（降级策略，如“若API超时，则改用本地缓存+人工审核标记”）

这个schema不是写死的，而是由上游业务系统动态注入。比如财务系统传来的目标可能是{"primary_objective": "verify_vat_invoice", "resource_constraints": {"max_api_calls": 2, "timeout_ms": 5000}}，而HR系统传来的可能是{"primary_objective": "onboard_new_hire", "fallback_strategies": ["send_welcome_email_only", "skip_background_check"]}。agent拿到这个goal schema后，才开始规划tool调用序列。这就从根本上避免了“为了调用而调用”的陷阱。

2.2 “能动性”的四层实现阶梯：从响应式到责任式

很多开发者以为加个“请自主决策”提示词就能赋予agency，这是最大的误区。真正的能动性必须分层构建，每一层都对应具体的工程实现：

层级	名称	关键能力	典型实现方式	我踩过的坑
L1	响应式能动	对输入做出合理响应	Prompt中嵌入role definition + few-shot examples	例子太少导致泛化差，加5个例子后准确率从63%升到89%
L2	规划式能动	将目标分解为可执行步骤	使用ReAct模式，但强制要求每步输出`<plan>`标签包裹的原子动作	曾忽略标签闭合，导致解析器直接崩溃，后来加了正则校验
L3	协作式能动	在多agent环境中协商资源与优先级	引入轻量级消息总线（如Redis Pub/Sub），每个agent发布`intent`和`capacity`	初始用HTTP轮询，10个agent并发时延迟飙升至8s，换消息总线后稳定在120ms内
L4	责任性能动	对结果负责，主动触发补救或上报	每个agent维护本地`outcome_log`，失败时自动触发`report_outcome({"status": "failed", "reason": "...", "suggestion": "..."})`	最初只记录成功日志，线上故障排查耗时增加4倍，补全失败日志后MTTR缩短70%

注意：这四层不是线性升级关系，而是叠加关系。一个L4级agent必然同时具备L1-L3能力。但实践中，我们往往先实现L2（规划），再补L3（协作），最后攻坚L4（责任）。因为L4涉及系统可观测性改造，需要和监控平台打通。某导师曾提醒我：“别急着造火箭，先确保你的轮子不会在平地上散架。”这句话让我少走了半年弯路。

2.3 架构选型背后的残酷权衡：为什么放弃“全能Agent”幻想

看到这里，你可能会想：既然这么复杂，为什么不直接用AutoGen或CrewAI？它们不是号称“开箱即用的多agent框架”吗？我实测过三个主流方案，结论很明确：它们解决的是“如何快速启动”，而不是“如何安全运行”。AutoGen的GroupChatManager在10个agent以上就会出现消息乱序；CrewAI的Task依赖链在异步环境下经常丢失上下文；而LangGraph虽然灵活，但要求开发者手动管理State Schema，一个字段名拼错，整个流程就静默失败。

agency-agents的架构选择，本质上是一场关于“可控性”和“便利性”的博弈。我们最终采用的方案是：自研轻量级Agent Runtime（约1200行Python），配合标准化Goal Schema + 外部工具注册中心。这个Runtime只做四件事：

加载goal schema并校验合法性
执行规划器（Plan Engine），输出带优先级的动作队列
管理工具调用生命周期（超时、重试、熔断）
记录outcome log并触发hook（如告警、审计）

所有“智能”相关逻辑（比如用LLM做规划）都作为插件注入，Runtime本身不碰任何大模型。这样做的好处是：当某天需要把规划器从GPT-4换成本地Qwen2-7B时，只需替换一个插件，Runtime完全不用动。而AutoGen这类框架，模型切换意味着重写整个orchestration逻辑。某公司曾因合规要求必须禁用所有云API，在旧架构上花了6周迁移，而我们的新架构3天就切完了——因为所有外部依赖都在插件层，主干代码零修改。

3. 核心细节解析与实操要点：Goal Schema的工程化落地

3.1 Goal Schema不是文档，而是可执行契约

很多人把goal schema当成需求文档来写，这是根本性错误。它必须是能被程序直接解析、校验、执行的契约。我们定义的schema v1.2版本如下（精简关键字段）：

{ "version": "1.2", "primary_objective": "validate_payment_receipt", "success_criteria": { "type": "boolean_expression", "expression": "(receipt_status == 'verified') && (amount_matched == true)" }, "resource_constraints": { "max_api_calls": 3, "timeout_ms": 8000, "memory_mb": 256 }, "fallback_strategies": [ { "name": "use_cached_verification", "condition": "api_latency > 5000", "action": "load_from_cache('payment_receipts')" }, { "name": "escalate_to_human", "condition": "verification_attempts >= 3", "action": "send_slack_alert('@finance-team', 'URGENT: receipt validation failed')" } ], "required_tools": ["payment_gateway_api", "document_ocr_service"], "output_format": { "schema": { "receipt_id": "string", "verified_at": "iso_datetime", "confidence_score": "float_range(0.0, 1.0)" } } }

这个schema的关键在于success_criteria.expression和fallback_strategies[].condition字段。它们不是自然语言描述，而是可求值的表达式。我们用simpleeval库（轻量、安全、无代码执行风险）来解析执行。比如api_latency > 5000会被转换为Python字节码，在沙箱中运行，返回True/False。这比用LLM判断“是不是超时了”可靠100倍——后者可能把“5.2秒”识别成“52秒”。

注意：simpleeval不支持import和复杂函数，但我们扩展了它的命名空间，加入了datetime.now()、math.isclose()等安全函数。所有扩展函数都经过白盒审计，确保无副作用。千万别用eval()直接执行用户输入，这是生产环境红线。

3.2 工具注册中心：让agent“认识”工具，而不是“背诵”工具

传统做法是把tool description硬编码在prompt里，这导致两个问题：一是LLM token数爆炸，二是工具变更时要重训整个系统。agency-agents的解法是：建立中心化工具注册中心（Tool Registry），agent通过ID动态获取工具元数据。

注册中心本质是个带版本控制的JSON存储（我们用PostgreSQL的JSONB字段）。每个tool entry长这样：

{ "tool_id": "payment_gateway_api", "version": "v2.1", "description": "Verify payment receipt status and amount against gateway records", "input_schema": { "type": "object", "properties": { "receipt_id": {"type": "string"}, "expected_amount": {"type": "number"} } }, "output_schema": { "type": "object", "properties": { "status": {"enum": ["verified", "rejected", "pending"]}, "amount_matched": {"type": "boolean"}, "gateway_response_code": {"type": "string"} } }, "availability": "high", // high/medium/low，影响agent调度优先级 "sla_ms": 3000 }

agent在规划阶段，会向注册中心发起查询：GET /tools?ids=payment_gateway_api,document_ocr_service&version=v2.1。注册中心返回结构化元数据，agent据此生成调用参数。如果某天payment_gateway_api升级到v2.2，新增了currency字段，注册中心只需更新这一条记录，所有agent下次规划时自动感知变更——无需重启服务，更不用改一行代码。

实操心得：我们给注册中心加了“影子模式”（shadow mode）。当agent调用v2.2工具时，系统会并行调用v2.1做对比，记录差异并告警。上线首周就发现v2.2在特定金额区间返回了错误的状态码，及时回滚，避免了资损。

3.3 Outcome Log：能动性的证据链，也是故障的破案线索

L4级能动性的基石是outcome log。它不是简单的日志打印，而是结构化的证据链。每条log必须包含：

event_id: UUIDv4，全局唯一
agent_id: 执行agent的标识
goal_id: 关联的目标ID（来自goal schema）
step_id: 当前执行步骤序号（如"plan_step_2"）
action_type: "plan"/"execute"/"fallback"/"report"
timestamp: ISO8601格式，精确到毫秒
input_data: 输入参数的哈希摘要（避免日志过大，保留可追溯性）
output_data: 输出结果的哈希摘要
status: "success"/"failed"/"partial"/"skipped"
error_details: 仅在failed时存在，包含错误类型、堆栈摘要、建议修复项

这个schema看似繁琐，但带来的价值巨大。某次线上故障，用户反馈“报销单一直卡在验证环节”。运维同事查常规日志只看到“API timeout”，但通过outcome log关联goal_id，发现所有失败事件都集中在document_ocr_service，进一步查error_details显示“rate limit exceeded”。原来OCR服务商悄悄把免费额度从1000次/天降到100次/天。没有outcome log，这个问题会归因为“网络不稳定”，彻底跑偏。

提示：我们用Elasticsearch存储outcome log，建立专门的Kibana仪表盘。最常用的查询是：status: "failed" AND action_type: "execute" AND input_data.hash: "abc123"，能5秒内定位到同一批次的所有失败实例。

4. 实操过程与核心环节实现：从零搭建一个可验证的agency-agents系统

4.1 环境准备与最小可行Runtime（15分钟搞定）

我们用Python 3.11+构建Runtime，依赖极简：fastapi==0.111.0,pydantic==2.7.1,simpleeval==0.9.12,redis==4.6.0。不装任何LLM SDK，保持框架中立。

第一步：创建核心数据模型（models.py）

from pydantic import BaseModel, Field, validator from typing import List, Dict, Optional, Any from enum import Enum class ToolAvailability(str, Enum): HIGH = "high" MEDIUM = "medium" LOW = "low" class ToolMetadata(BaseModel): tool_id: str version: str description: str input_schema: Dict[str, Any] output_schema: Dict[str, Any] availability: ToolAvailability sla_ms: int class FallbackStrategy(BaseModel): name: str condition: str # boolean expression action: str # e.g., "load_from_cache('receipts')" class GoalSchema(BaseModel): version: str = "1.2" primary_objective: str success_criteria: Dict[str, Any] # { "type": "boolean_expression", "expression": "..." } resource_constraints: Dict[str, Any] fallback_strategies: List[FallbackStrategy] required_tools: List[str] output_format: Dict[str, Any] class OutcomeLog(BaseModel): event_id: str agent_id: str goal_id: str step_id: str action_type: str # plan/execute/fallback/report timestamp: str input_data: Dict[str, str] # { "hash": "sha256...", "size_bytes": 1234 } output_data: Dict[str, str] status: str # success/failed/partial/skipped error_details: Optional[Dict[str, str]] = None

第二步：实现Goal Schema校验器（validator.py）

from simpleeval import SimpleEval from models import GoalSchema, FallbackStrategy class GoalValidator: def __init__(self): self.evaluator = SimpleEval() # 安全函数注入 self.evaluator.names.update({ 'datetime': __import__('datetime'), 'math': __import__('math'), 'len': len, 'str': str, 'float': float, 'int': int }) def validate_expression(self, expr: str, context: Dict[str, Any]) -> bool: """校验布尔表达式是否合法且可求值""" try: # 防注入：只允许简单表达式 if any(c in expr for c in ['import', '__', 'exec', 'eval']): raise ValueError("Unsafe expression detected") return bool(self.evaluator.eval(expr, names=context)) except Exception as e: raise ValueError(f"Invalid expression '{expr}': {e}") def validate_goal(self, goal_dict: Dict[str, Any]) -> GoalSchema: """完整校验goal dict并返回pydantic模型""" try: goal = GoalSchema(**goal_dict) except Exception as e: raise ValueError(f"Schema validation failed: {e}") # 校验success_criteria表达式 if goal.success_criteria.get("type") == "boolean_expression": test_context = {"receipt_status": "verified", "amount_matched": True} self.validate_expression(goal.success_criteria["expression"], test_context) # 校验fallback条件 for fb in goal.fallback_strategies: test_context = {"api_latency": 6000, "verification_attempts": 4} self.validate_expression(fb.condition, test_context) return goal

第三步：启动FastAPI服务（main.py）

from fastapi import FastAPI, HTTPException, Depends from models import GoalSchema, OutcomeLog, ToolMetadata from validator import GoalValidator import redis import json import uuid from datetime import datetime app = FastAPI(title="Agency-Agents Runtime") validator = GoalValidator() r = redis.Redis(host='localhost', port=6379, db=0) @app.post("/validate-goal") def validate_goal(goal: GoalSchema): try: validated = validator.validate_goal(goal.dict()) return {"status": "valid", "goal_id": str(uuid.uuid4())} except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) @app.post("/log-outcome") def log_outcome(log: OutcomeLog): log.timestamp = datetime.utcnow().isoformat() r.lpush("outcome_logs", log.json()) return {"status": "logged"} @app.get("/tools") def get_tools(tool_ids: str, version: str = "latest"): # 实际项目中这里对接数据库 # 示例返回静态数据 tools = [] for tid in tool_ids.split(","): if tid == "payment_gateway_api": tools.append(ToolMetadata( tool_id="payment_gateway_api", version="v2.1", description="Verify payment receipt...", input_schema={"type": "object", "properties": {"receipt_id": {"type": "string"}}}, output_schema={"type": "object", "properties": {"status": {"enum": ["verified"]}}}, availability=ToolAvailability.HIGH, sla_ms=3000 )) return tools

运行命令：uvicorn main:app --reload。访问http://localhost:8000/docs即可测试API。这个Runtime只有3个端点，但已具备agency-agents的核心骨架：校验目标、记录结果、发现工具。后续所有“智能”都作为插件注入，Runtime永远保持稳定。

4.2 规划器（Plan Engine）实战：用LLM生成可执行动作序列

规划器是agent的“大脑”，但它必须受Runtime约束。我们不追求“最聪明的规划”，而追求“最可控的规划”。规划器输入是validated goal，输出是带优先级的动作列表。

核心设计原则：

输出必须结构化：用JSON Schema强制约束，避免LLM自由发挥
动作必须原子化：每个动作只能调用一个tool，或执行一个本地操作（如sleep(1)）
必须包含失败处理声明：每个动作指定on_failure策略

规划器Prompt模板（精简版）：

你是一个严格遵守契约的规划引擎。请根据以下Goal Schema，生成一个可执行的动作序列。 要求： 1. 动作必须按执行顺序编号（1,2,3...） 2. 每个动作必须包含：action_type（"tool_call" or "local_op"）, tool_id（仅tool_call时存在）, parameters（JSON对象）, on_failure（"retry"/"skip"/"fallback"） 3. 输出严格为JSON，格式：{"actions": [{"id": 1, "action_type": "...", ...}, ...]} 4. 不要任何解释、不要markdown、不要额外字符。 Goal Schema: {goal_json}

调用示例（用OpenAI API）：

import openai def generate_plan(goal: GoalSchema) -> List[Dict]: prompt = build_prompt(goal) response = openai.ChatCompletion.create( model="gpt-4-turbo", messages=[{"role": "user", "content": prompt}], temperature=0.1, max_tokens=1000 ) try: plan = json.loads(response.choices[0].message.content) return plan["actions"] except json.JSONDecodeError: # 降级：用规则引擎生成基础plan return rule_based_plan(goal)

实测下来，GPT-4-Turbo在结构化输出上准确率92%，但仍有8%概率输出非法JSON。因此我们加了双重保障：

JSON Schema校验（用jsonschema库）
备用规则引擎（rule_based_plan）：对常见goal objective（如validate_payment_receipt）预设动作序列

实操心得：别迷信LLM。我们给规划器加了“确定性开关”——当goal的primary_objective在白名单中（如["validate_payment_receipt", "onboard_new_hire"]），强制走规则引擎，速度更快、100%可靠；只有遇到新objective时，才启用LLM。上线后规划平均耗时从1.2s降到0.3s，错误率归零。

4.3 工具调用生命周期管理：超时、重试、熔断的三位一体

工具调用不是发个HTTP请求就完事。agency-agents要求每个调用都携带“契约意识”。我们在Runtime中实现了统一的Tool Executor：

import asyncio import time from typing import Dict, Any class ToolExecutor: def __init__(self, timeout_ms: int = 5000, max_retries: int = 2): self.timeout_ms = timeout_ms self.max_retries = max_retries self.circuit_breaker = CircuitBreaker() async def execute(self, tool_id: str, params: Dict[str, Any], goal_id: str, step_id: str) -> Dict[str, Any]: # 1. 熔断检查 if not self.circuit_breaker.allow_request(tool_id): raise CircuitBreakerOpenError(f"Circuit breaker open for {tool_id}") # 2. 执行带超时和重试 for attempt in range(self.max_retries + 1): try: start_time = time.time() # 这里调用实际工具，如requests.post(...) result = await self._call_tool(tool_id, params) elapsed_ms = int((time.time() - start_time) * 1000) # 3. 成功：记录日志，重置熔断器 self.circuit_breaker.record_success(tool_id) return { "status": "success", "data": result, "latency_ms": elapsed_ms } except Exception as e: elapsed_ms = int((time.time() - start_time) * 1000) self.circuit_breaker.record_failure(tool_id) if attempt == self.max_retries: raise e await asyncio.sleep(min(1 * (2 ** attempt), 10)) # 指数退避 return {"status": "failed", "error": "Unknown error"}

CircuitBreaker实现基于滑动窗口统计（最近60秒内失败率>50%则熔断）。这个设计让系统在依赖服务抖动时，能自动降级，而不是雪崩。某次支付网关故障，我们的agent在3秒内检测到连续失败，自动熔断，转而执行fallback策略（用本地缓存+人工审核），保障了核心业务可用性。

5. 常见问题与排查技巧实录：那些文档里不会写的血泪教训

5.1 典型问题速查表

问题现象	可能原因	排查步骤	解决方案	我的实操记录
Goal校验总是失败，报"Invalid expression"	表达式含非法字符（如中文括号、全角符号）	1. 复制expression到在线正则校验器 2. 检查`simpleeval`命名空间是否注入了所需函数	用VS Code的"显示不可见字符"功能，发现用户粘贴时带了全角空格；在validator中加预处理：`expr.replace(' ', ' ').replace('（', '(').replace('）', ')')`	发生3次，每次平均耗时2小时定位，现在加了预处理后0发生
Agent规划出的动作序列里，tool_id拼写错误（如`paymnt_gateway_api`）	工具注册中心未校验tool_id格式，LLM自由发挥	1. 查`/tools`接口返回的tool_id列表 2. 检查规划prompt中是否强调"必须使用注册中心返回的tool_id"	在GoalValidator中增加`required_tools`校验：遍历`required_tools`，确认每个tool_id都在注册中心存在	上线首周发现，立即加校验，后续0发生
Outcome log里`input_data.hash`和`output_data.hash`总是空	日志记录逻辑未处理大对象，直接跳过哈希计算	1. 查`log_outcome`端点代码 2. 检查`input_data`是否为None或空dict	改为：`if input_data: hash_val = hashlib.sha256(json.dumps(input_data, sort_keys=True).encode()).hexdigest()[:16]`	因为日志量太大，初期故意留空，后来发现无法追溯，补全后排查效率提升3倍
多个agent同时处理同一goal，结果冲突	Goal ID未全局唯一，或注册中心未加锁	1. 查goal_id生成逻辑 2. 检查注册中心读写是否线程安全	Goal ID强制用UUIDv4；注册中心读操作无锁，写操作用Redis SETNX加分布式锁	某次批量导入引发数据覆盖，损失2小时数据，从此加锁

5.2 独家避坑技巧：让agency-agents真正“稳”下来

技巧1：给LLM规划器加“温度钳制”LLM的temperature参数直接影响输出稳定性。我们发现，temperature=0.1时，相同goal的规划序列95%一致；但temperature=0.7时，只有60%一致。这会导致同样的目标，今天生成动作A-B-C，明天变成A-C-B，而C依赖B的输出。解决方案：在规划器调用时，强制temperature=0.0，并用top_p=0.95保底。实测下来，规划一致性达99.8%，且不牺牲合理性。

技巧2：Fallback不是“兜底”，而是“分级弹射”很多团队把fallback写成“实在不行就发邮件”。这太粗暴。我们设计了三级弹射：

Level 1（自动）：用缓存/默认值/简化逻辑（如“跳过金额校验，只验单号格式”）
Level 2（半自动）：生成带上下文的Slack消息，@指定角色，附带一键执行按钮（点击即触发Level 1）
Level 3（人工）：创建Jira ticket，自动填充goal_id、失败日志URL、建议排查路径这样，90%的问题在Level 1解决，9%在Level 2解决，只有1%需要Level 3。运维压力直线下降。

技巧3：Outcome Log的“黄金15字段”别记录所有字段！我们只保留15个必填字段，其他全删。这15个是：event_id,agent_id,goal_id,step_id,action_type,timestamp,input_hash,output_hash,status,error_type,error_message,latency_ms,tool_id,tool_version,fallback_used。上线后日志体积减少70%，查询速度提升5倍，且关键信息一个不落。

5.3 性能压测实录：从100到10000 QPS的演进路径

我们用Locust对Runtime做了三轮压测：

Round 1（裸奔）：单进程，无连接池，直接requests调用。结果：100 QPS时，P95延迟1200ms，错误率8%。瓶颈在HTTP连接创建。
Round 2（连接池+异步）：用httpx.AsyncClient，连接池大小设为100。结果：500 QPS时，P95延迟320ms，错误率0.2%。但内存占用飙升，GC频繁。
Round 3（连接池+异步+本地缓存）：为/tools接口加Redis缓存（TTL=60s），工具元数据变更时主动失效。结果：10000 QPS时，P95延迟210ms，错误率0.01%，内存稳定。

关键发现：工具元数据查询是最大瓶颈。因为每个规划都要查一次，10000 QPS就是10000次查询。加缓存后，99%的请求走Redis，耗时从50ms降到0.2ms。这个优化比升级服务器配置有效10倍。

6. 后续可扩展方向：让agency-agents从“能用”走向“好用”

做完上述所有，你已经拥有了一个生产级的agency-agents基础。但真正的价值延伸，在于它如何融入更大的技术生态。根据我们多个项目的实践，有三个高价值扩展方向值得立刻规划：

方向一：Goal Schema的可视化编辑器现在schema靠手写JSON，对非技术人员不友好。我们正在开发一个低代码编辑器，左侧拖拽组件（目标、条件、工具、降级策略），右侧实时生成schema。最关键是“表达式校验”——输入api_latency > 5000，编辑器立刻调用Runtime的validate_expression接口，返回✅或❌。这能让业务方直接参与目标定义，而不是等技术团队排期。

方向二：Outcome Log驱动的自动归因系统当前故障排查靠人工。下一步，我们训练一个轻量级分类模型，输入outcome log的error_message和tool_id，输出根因标签（如"network_timeout","auth_expired","schema_mismatch"）。模型用历史log微调，准确率已达89%。未来，当status=="failed"时，系统自动打上根因标签，并推荐修复方案（如“请更新tool_id=payment_gateway_api的token”）。

方向三：跨系统Goal Schema联邦不同团队有自己的goal schema标准。某高校实验室用objective字段，某公司用primary_goal。我们设计了一个Schema Mapping Layer，用YAML定义映射规则：

mapping_rules: - source_field: "objective" target_field: "primary_objective" transform: "upper_case" - source_field: "constraints" target_field: "resource_constraints"

这样，上游系统推送的任意格式goal，都能被自动转换为Runtime认可的schema。这解决了企业级落地的最大障碍——标准不统一。

我个人在实际操作中的体会是：agency-agents不是银弹，它解决不了“需求不清晰”的问题，也治不好“数据质量差”的顽疾。但它像一把手术刀，把模糊的“智能”切成可测量、可调试、可追责的工程模块。当你第一次看到outcome log里清晰标记出“第3次fallback因token过期触发，已自动刷新”，那种掌控感，是任何demo都无法替代的。