1. 项目概述：这不是一个“代理”概念，而是一套可协作的智能体工作流设计范式
“agency-agents”这个词组最近在技术社区里频繁出现，但很多人第一反应是把它当成“代理（agent）”的同义反复，甚至下意识联想到传统IT运维里的“代理服务器”或软件安装时弹出的“Agent Installer”。其实完全不是一回事。它指的是一种以目标驱动、具备自主决策能力、能调用工具并与其他同类智能体协同完成复杂任务的软件实体集合。核心关键词就三个：Agency（能动性）、Agents（多角色）、Collaboration（协作）。它解决的不是“怎么连上某个服务”，而是“怎么让一堆各司其职的程序，像一支小型特种作战小队那样，自己商量、分工、试错、回滚、最终拿下目标”。
适合谁看？如果你正在做自动化流程编排、想把RPA升级成真正有判断力的工作流、或是开发需要多步骤推理的AI应用（比如自动写周报+查数据+生成图表+邮件发送），那这个概念就是你当前技术栈的“临门一脚”。它不依赖某个特定大模型API，也不绑定某家云厂商，而是一种架构思想——就像当年大家从写单个函数，进化到理解“微服务”一样，“agency-agents”是你下一步该建立的认知坐标。我去年在一个跨部门数据同步项目里，用这套思路把原本需要3人盯5小时的流程，压缩成一个无人值守的夜间任务，整个过程没有一行硬编码的调度逻辑，全是靠几个轻量级Agent之间“对话”完成的。
2. 核心设计思路拆解：为什么必须是“Agency”而不是“Agent”？
2.1 单点智能体的天然缺陷：它再聪明，也只是一个“执行员”
我们先说清楚一个常见误区：很多人以为给一个LLM加个function calling接口，再喂点提示词，就是一个“Agent”了。这没错，但它只是个“单兵”。就像给你一把瑞士军刀，你能切苹果、开瓶盖、拧螺丝，但你没法独自完成盖一栋楼的任务。单个Agent的瓶颈非常明确：
- 上下文窗口限制：所有思考都挤在一次prompt里，复杂任务一拆就散；
- 状态不可持续：上一步的中间结果，很难自然传递给下一步，容易“断片”；
- 错误无容错机制：一旦某步失败（比如API超时、数据格式异常），整个流程就卡死，不会自动降级或换方案；
- 角色单一：它只能干自己被设定好的事，无法根据现场情况临时切换身份。
我试过让一个单Agent去处理客户投诉工单：它要先读邮件，再查订单系统，再调客服知识库，最后生成回复。结果90%的失败都发生在第三步——知识库接口偶尔返回空结果，Agent就僵在那里，既不会重试，也不会查历史相似案例，更不会喊同事帮忙。它不是懒，是根本没被设计成“会求助”的角色。
2.2 “Agency”范式的破局点：把任务分解成“角色-动作-反馈”闭环
“agency-agents”的核心突破，在于把一个大目标，拆解成一组有明确职责、有通信协议、有失败兜底的微型Agent集群。它们不共享内存，不共用模型，甚至可以跑在不同机器上，只靠结构化消息（比如JSON-RPC风格的请求/响应）来协作。这背后有三根支柱：
第一根支柱：角色专业化（Role Specialization）
每个Agent只负责一个原子能力。比如：
Researcher：只管查资料、调API、抓网页，不写结论；Critic：只管审逻辑、找矛盾、标风险点，不生成新内容；Planner：只管拆任务、排优先级、设检查点，不碰具体数据；Executor：只管调工具、发请求、存结果，不参与任何判断。
这种分工不是为了炫技，而是为了可测试、可替换、可审计。当Researcher查错数据时，你只需重训它，不影响Critic的判断逻辑；当Planner策略过时，你换一个新规划器，其他Agent完全无感。
第二根支柱：消息即契约（Message as Contract）
Agent之间不传“想法”，只传带Schema的结构化消息。比如Planner给Researcher发的不是“你去查一下用户投诉原因”，而是：
{ "task_id": "T-2024-0876", "action": "search_customer_complaint_root_cause", "params": { "order_id": "ORD-987654", "time_window_hours": 72, "data_sources": ["crm", "support_tickets", "payment_logs"] }, "deadline": "2024-05-22T03:15:00Z" }这个JSON就是一份数字契约：明确了要做什么、输入是什么、时限在哪。Researcher收到后，要么返回成功结果，要么返回标准错误码（如ERR_SOURCE_UNAVAILABLE），绝不含糊。这就把模糊的“协作”变成了可编程的“接口调用”。
第三根支柱：反馈驱动演进（Feedback-Driven Iteration）
整个流程不是线性的“计划→执行→结束”，而是环形的“提议→评审→修正→再执行”。典型工作流是：
Planner发初始任务给Executor；Executor执行后，把原始结果发给Critic；Critic分析后，可能提出3种反馈：✅通过、⚠️需补充数据、❌逻辑矛盾；- 如果是⚠️或❌，
Critic会直接把修正建议（比如“请再查退款流水号”）发回Planner，触发新一轮小循环。
这个机制让系统具备了“类人”的试错能力。我实测过一个财报摘要生成任务：第一次Researcher漏掉了子公司数据，Critic立刻标出“母公司营收占比异常高”，Planner马上补发子公司的查询指令，整个过程无人干预，2分17秒就产出终稿。
2.3 与常见架构的对比：它不是微服务，也不是工作流引擎
很多人会拿它和微服务、Airflow、Camunda对比，但关键差异在于决策权归属：
| 维度 | 微服务架构 | Airflow工作流 | agency-agents范式 |
|---|---|---|---|
| 控制流 | 由外部API网关或客户端驱动 | 由DAG定义的静态依赖关系驱动 | 由Agent间实时协商动态生成 |
| 失败处理 | 重试/熔断/降级，需人工配置策略 | 任务失败则整个DAG中断或跳过 | Critic可发起局部重试、换数据源、改方案 |
| 状态管理 | 状态存在数据库或Redis中，服务无状态 | 状态由Scheduler维护，Task Instance记录进度 | 状态隐含在消息链中（task_id贯穿全程），Agent本身无持久状态 |
| 扩展性 | 水平扩容服务实例 | 增加Worker节点 | 新增Agent类型即可支持新能力（如加LegalReviewer审核合规） |
你看，agency-agents不是在造一个新的调度器，而是在构建一套分布式认知系统。它的扩展不是加机器，而是加“脑子”。
3. 核心实现细节与实操要点：从零搭建一个最小可行Agency
3.1 最简技术栈选型：为什么选Python + LangChain + LiteLLM？
很多团队一上来就想用Kubernetes部署Agent集群，结果三个月还在搞服务发现。其实最小可行Agency（MVA）只需要三样东西：一个轻量模型、一个消息总线、一套角色模板。我推荐的组合是：
模型层：LiteLLM统一网关
不用纠结用哪个大模型。LiteLLM就像一个“模型路由器”，你写代码时只认litellm.completion()这个接口，背后可以随时切到OpenAI、Ollama本地模型、甚至阿里云百炼。这样做的好处是：当某天公司禁用OpenAI，你只需改一行配置，所有Agent立刻切换到私有模型，业务零感知。我线上环境就用它做了AB测试：白天走云端模型保质量，凌晨用本地Qwen2-7B省成本，切换完全透明。框架层：LangChain的AgentExecutor + Custom Tools
LangChain的AgentExecutor不是万能胶，但它的tool抽象极其干净。每个Agent本质就是一个Tool集合+一个决策Prompt。比如Researcher的Tool列表可能是：
tools = [ TavilySearchResults(max_results=3), # 公网搜索 SQLDatabaseToolkit(db=customer_db), # 查客户库 RequestsGetTool(), # 调内部API ]关键技巧：不要让Agent自己决定用哪个Tool。我在Planner里预设规则：“订单问题必查CRM，支付问题必查账单系统”，然后把具体Tool名作为参数传给Executor。这样既保证专业性，又避免模型胡乱调用。
- 通信层：内存内消息队列（初期）→ Redis Stream（上线）
起步阶段，用Python的queue.Queue完全够用。所有Agent启动时注册一个Inbox和Outbox，Planner往Outbox塞消息，Executor轮询Inbox收消息。等规模上来，换成Redis Stream，每类Agent订阅自己的Stream（如stream:researcher），天然支持水平扩展和消息持久化。千万别一上来就上Kafka——90%的团队根本用不到它的吞吐量，反而被Topic管理搞崩溃。
提示：消息体必须包含
correlation_id字段。这是整个Agency的“DNA”，从第一个Planner生成，贯穿所有后续消息，最后汇总报告时全靠它串起完整链路。没有它，你就失去了追溯能力。
3.2 四大核心Agent角色模板详解（附可运行代码片段）
下面是我压箱底的四个最常用Agent模板，已脱敏处理，可直接复制使用：
3.2.1 Planner：任务拆解与资源调度中枢
它的核心不是“想点子”，而是“做减法”。Prompt里最关键的约束是：
“你只能输出JSON，字段仅限：
sub_tasks（数组，每个元素含id、description、required_tools、estimated_duration_minutes）、critical_path（字符串数组，列出必须顺序执行的task_id）、fallback_options（字典，key为失败场景，value为备用task_id）”
# planner.py from langchain.agents import AgentExecutor, create_tool_calling_agent from langchain_core.prompts import ChatPromptTemplate planner_prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个严谨的项目规划师。收到目标后，必须按以下规则拆解：1. 每个子任务必须能被单一Tool完成；2. 标出哪些任务必须按序执行；3. 为每个高风险任务预设1个备用方案。只输出JSON，不加任何解释。"), ("human", "{input}") ]) planner_agent = create_tool_calling_agent( llm=litellm_llm, tools=[get_crm_schema, list_api_endpoints], # 只给它查元数据的Tool prompt=planner_prompt ) planner_executor = AgentExecutor(agent=planner_agent, tools=[get_crm_schema, list_api_endpoints])实操心得：Planer的Tool一定要克制。我见过太多团队给它塞了10个Tool，结果它天天在“该查CRM还是该调API”上纠结。记住：Planner只配知道“有什么可用”，不配知道“怎么用”。
3.2.2 Researcher：信息获取的“侦察兵”
它不归纳、不总结，只做三件事：精准调用、清洗原始数据、标注置信度。关键设计是强制返回结构化结果：
# researcher.py class ResearcherTool(BaseTool): name = "research_data" description = "从指定数据源获取原始信息，返回结构化JSON" def _run(self, source: str, query: str) -> dict: if source == "crm": return {"data": crm_client.search(query), "confidence": 0.92, "source_timestamp": "2024-05-21T14:30:00Z"} elif source == "support_tickets": return {"data": ticket_api.search(query), "confidence": 0.78, "source_timestamp": "2024-05-21T10:15:00Z"} # ... 其他数据源注意：
confidence字段不是随便填的。我的规则是：数据库直查=0.9+，API调用=0.7~0.85，网页爬取=0.5~0.7。Critic会重点审查低置信度数据。
3.2.3 Critic：逻辑守门员
这是最容易被低估的角色。它的Prompt必须带可验证的检查清单：
“你收到一份研究结果，请逐项检查：1. 所有数据时间戳是否在目标时间窗内？2. 关键数值是否有单位缺失？3. 是否存在自相矛盾的陈述（如‘已退款’与‘状态为待处理’）？4. 置信度低于0.7的数据是否被用于核心结论？输出JSON：{‘verdict’: ‘PASS’/‘FAIL’/‘NEED_MORE’, ‘issues’: [‘问题1’, ‘问题2’], ‘suggestions’: [‘建议1’]}”
避坑经验：Critic绝不能自己生成新数据！它只能指出问题，不能代替Researcher去重查。否则就乱了职责边界。
3.2.4 Executor：行动派终结者
它最简单，也最危险。我的Executor只做一件事：把Planner指定的Tool和参数，原封不动执行，并把原始返回包一层{"status": "success"/"error", "raw_output": ...}再发出去。绝不做任何格式转换、字段提取、错误美化。因为所有“理解”工作都该由Critic和Planner完成。
# executor.py def execute_tool(tool_name: str, params: dict) -> dict: try: tool = TOOLS_REGISTRY[tool_name] result = tool.invoke(params) return {"status": "success", "raw_output": result} except Exception as e: return {"status": "error", "error_type": type(e).__name__, "error_message": str(e)}关键技巧：TOOLS_REGISTRY必须是全局单例，且每个Tool的invoke()方法要有超时控制（我设为15秒）。曾经有个HTTP Tool没设timeout，卡住整个Agency，排查了两天才发现是某个内部API挂了。
3.3 消息路由与状态追踪：如何让10个Agent不乱成一锅粥？
Agency的“大脑”不在某个中心节点，而在消息协议里。我设计了一套极简但有效的路由规则：
- 消息类型标识：每条消息
type字段只能是PLAN_REQUEST、EXECUTE_COMMAND、CRITIC_FEEDBACK、REPORT_FINAL四种； - 路由表（Routing Table）：一个纯文本文件，内容如下：
PLAN_REQUEST → Planner EXECUTE_COMMAND → Executor CRITIC_FEEDBACK → Planner (if verdict==FAIL) OR Reporter (if verdict==PASS) REPORT_FINAL → UserInterface - 状态快照：每个
task_id对应一个Redis Hash，存{"status": "planning"/"executing"/"critiquing"/"done", "last_updated": "iso_time", "steps": [...]}。前端查进度时，直接HGETALL task:T-2024-0876就行。
实测效果：这套路由在日均5000+任务的测试环境里，平均延迟127ms，99.99%消息零丢失。比用复杂消息中间件还稳——因为够简单，所以够可靠。
4. 完整实操流程演示：用agency-agents自动处理一封客户投诉邮件
4.1 场景还原：一封真实的投诉邮件内容
主题：订单ORD-987654付款失败，但账户被扣款
内容：我在5月20日15:23下单，支付页面显示“支付成功”，但21日查账户发现未到账，订单状态却是“已取消”。客服说要3个工作日处理，我现在很生气！
发件人：customer@xxx.com
时间：2024-05-21T16:02:11Z
这个需求看似简单，但涉及至少4个系统：订单库、支付网关、账户系统、客服工单系统。单Agent会晕，而Agency能拆解。
4.2 全流程消息链路与耗时分析（真实日志节选）
Step 1：Planner接收原始邮件，生成拆解计划（耗时：8.2秒）Planner收到消息后，调用get_crm_schema确认客户邮箱对应ID，再调list_api_endpoints查各系统健康状态，输出：
{ "sub_tasks": [ {"id": "S1", "description": "查订单ORD-987654全生命周期状态", "required_tools": ["query_order_db"], "estimated_duration_minutes": 1}, {"id": "S2", "description": "查支付网关该笔交易原始凭证", "required_tools": ["query_payment_gateway"], "estimated_duration_minutes": 2}, {"id": "S3", "description": "查客户账户余额变更流水", "required_tools": ["query_account_ledger"], "estimated_duration_minutes": 1}, {"id": "S4", "description": "生成客户沟通话术初稿", "required_tools": ["generate_response_draft"], "estimated_duration_minutes": 3} ], "critical_path": ["S1", "S2", "S3"], "fallback_options": {"S2_timeout": "S2_alt"} }Step 2：Executor并发执行S1/S2/S3（耗时：最长任务2.1秒）Executor同时发3个请求。其中S2调支付网关超时（网络抖动），自动触发fallbackS2_alt——改用对账文件解析服务，耗时4.3秒，但拿到关键字段transaction_id: PAY-20240520-889900。
Step 3：Critic校验三份结果（耗时：3.7秒）Critic发现：
- S1数据时间戳是2024-05-20T15:23:44Z，符合要求 ✅
- S2返回的
transaction_id与S3流水中的ref_id匹配 ✅ - S3显示“扣款成功”，但S1订单状态为“已取消”，存在矛盾 ❌
于是发CRITIC_FEEDBACK给Planner，建议：“核查订单取消原因，重点查风控系统日志”。
Step 4：Planner发起第二轮侦查（耗时：1.9秒）Planner新增子任务S5: query_risk_engine_log with transaction_id=PAY-20240520-889900，Executor执行后返回：{"risk_flag": "high", "reason": "IP异常波动"}。
Step 5：Critic最终确认，触发Reporter（耗时：0.8秒）
所有数据闭环，Critic输出PASS，Reporter整合成标准报告：
【投诉处理报告】 - 根本原因：风控系统判定IP异常，自动取消订单（非支付失败） - 客户损失：无，扣款已原路退回（流水号REF-20240521-112233） - 补偿方案：赠送20元无门槛券（券码：COMP-889900） - 沟通话术：已生成，见附件全程总耗时：19.8秒，比人工平均处理时间（22分钟）快66倍。
4.3 配置文件与启动脚本（可直接部署）
所有Agent用同一个config.yaml驱动，确保环境一致：
# config.yaml llm: provider: "litellm" model: "gpt-4-turbo" api_base: "https://api.litellm.ai/v1" timeout: 30 redis: host: "localhost" port: 6379 db: 0 agents: planner: max_retries: 2 tools: ["get_crm_schema", "list_api_endpoints"] researcher: data_sources: crm: {timeout: 5, max_retries: 1} payment_gateway: {timeout: 15, max_retries: 2} critic: confidence_threshold: 0.75 check_list: ["timestamp", "unit", "consistency", "confidence_usage"] logging: level: "INFO" file: "/var/log/agency/agency.log"启动命令极简：
# 启动Planner（监听PLAN_REQUEST） python agent_runner.py --role planner --config config.yaml # 启动Executor（监听EXECUTE_COMMAND） python agent_runner.py --role executor --config config.yaml # 启动Critic（监听CRITIC_FEEDBACK） python agent_runner.py --role critic --config config.yaml关键技巧：所有Agent进程都用supervisord托管，配置autorestart=true。曾有一次Critic因OOM崩溃，3秒内自动拉起，用户完全无感知。
5. 常见问题与实战排障指南：那些文档里不会写的坑
5.1 问题速查表：高频故障现象与根因定位
| 现象 | 可能根因 | 快速验证方法 | 解决方案 |
|---|---|---|---|
| 某个Agent收不到消息 | Redis Stream消费者组未创建，或group名拼错 | redis-cli XINFO GROUPS stream:executor看是否有group | 用XGROUP CREATE手动建组，确保所有Agent用同一group名 |
Critic总是返回NEED_MORE | Researcher返回的confidence字段缺失或为null | 查Redis中该task_id的raw_output字段，看JSON是否完整 | 在ResearcherTool._run()末尾加assert "confidence" in result，强制校验 |
任务卡在planning状态不动 | Planner调用的Tool（如list_api_endpoints）返回空列表 | 直接curl该Tool对应的API端点 | 检查Tool服务是否存活，或API网关路由规则是否失效 |
多个相同task_id消息被重复处理 | Executor未在消费消息后调用XACK | redis-cli XREADGROUP GROUP mygroup myconsumer COUNT 1 STREAMS stream:executor > | 在Executor.run()末尾加redis.xack("stream:executor", "mygroup", message_id) |
Reporter生成的话术过于机械 | generate_response_draftTool的Prompt缺少语气约束 | 查Reporter的输入Prompt，看是否有use_tone: "empathetic"等字段 | 在Tool Prompt里加入：“用口语化中文，带1个emoji，结尾用问句引发客户回应” |
5.2 我踩过的三个深坑及血泪教训
坑一：把“协作”做成“串行调用”，性能崩盘
早期我把Planner→Executor→Critic→Planner做成严格串行，结果一个任务要等4次网络RTT，平均耗时飙升到8秒。后来改成：Planner发完EXECUTE_COMMAND后，立刻开始监听CRITIC_FEEDBACK，不等Executor返回就进入等待态。Executor执行完直接发消息，Critic收到即处理，全程异步。性能提升4倍。
坑二：忽略消息幂等性，导致重复扣款
有一次Executor调支付网关成功，但发EXECUTE_COMMAND结果消息时网络闪断，Planner以为失败，重发了EXECUTE_COMMAND，结果真扣了两次款。解决方案：所有对外操作的Tool，必须带idempotency_key参数（用task_id + step_id哈希生成），网关层强制幂等。
坑三：Critic的“逻辑矛盾”检查太宽松，放过致命错误
最初Critic只检查字段是否存在，结果放过一个bug：Researcher把“退款金额”和“扣款金额”两个字段值填反了，但数值本身合法。后来加了交叉验证规则：“若订单状态为‘已取消’，则退款金额必须大于0且等于扣款金额”，才堵住这个漏洞。
5.3 性能调优四板斧：从100QPS到5000QPS的实录
第一斧：消息批处理Executor不单条处理，而是每200ms从Stream里XREADGROUP最多10条消息，批量执行。实测QPS从120升到310，CPU占用降35%。
第二斧：Critic缓存热规则
把高频检查项（如“时间戳校验”）编译成正则表达式，存入lru_cache(maxsize=1000)。Critic启动时预热，热点规则命中率92%，单次校验耗时从120ms降到8ms。
第三斧：Planner冷启动优化Planner首次启动时，预加载所有Tool的Schema到内存，避免每次都要调get_crm_schema。启动时间从4.2秒降到0.7秒。
第四斧：Redis连接池复用
所有Agent共享一个redis.ConnectionPool，max_connections=50。连接复用后，Redis连接数稳定在12个，不再出现ConnectionResetError。
6. 进阶应用场景与落地建议：别只盯着“自动化”，要看清它的认知革命本质
6.1 超越RPA的三大高价值场景
场景一：合规审计的“数字稽查组”
某金融客户用Agency组建了Regulator（模拟监管规则）、Auditor（查交易流水）、AnomalyDetector（识别可疑模式）三个Agent。当Regulator发现“单日大额转账超限额”，自动触发Auditor查该客户近30天所有流水，AnomalyDetector用孤立森林算法打分，最终生成符合银保监格式的《异常交易核查报告》。整个过程符合《金融机构反洗钱规定》第23条，审计留痕完整。
场景二：研发效能的“虚拟Tech Lead”
在CI/CD流水线里嵌入CodeReviewer（查PR）、TestRunner（跑用例）、DeployPlanner（评估发布风险）。当CodeReviewer标出高危SQL注入漏洞，DeployPlanner自动将本次发布标记为“需人工确认”，并生成修复建议。上线后，严重线上事故下降76%。
场景三：客户服务的“永不疲倦专家团”
把ProductExpert（懂参数）、Troubleshooter（会诊断）、EscalationManager（知流程）组成小组。客户问“打印机卡纸怎么办”，ProductExpert先确认型号，Troubleshooter推送分步视频，若3次操作失败，EscalationManager自动创建高级支持工单，并把前两步操作日志打包附上。首次解决率从68%升至91%。
6.2 给不同角色的落地建议
给技术负责人：别追求“全量迁移”。先选一个痛点最明确、数据最规范的流程（比如“新员工入职IT权限开通”），用Agency重构。2周内做出MVP，用实际节省的工时说服管理层。我见过最成功的案例，是从一个每天要手动处理47次的“发票验真”流程切入，3天上线，首月就省下127小时人力。
给一线开发者：忘掉“造轮子”。直接用LangChain的create_tool_calling_agent，把精力花在设计Tool上。一个好Tool的标准就三条：输入明确、输出结构化、失败可预期。我写的query_payment_gatewayTool，只有120行代码，但支撑了全公司支付类任务。
给业务方：你们要的不是“AI”，而是“确定性”。Agency的价值在于：每一步谁干的、用了什么数据、为什么这么判，全部可追溯。下次审计时，你不用求工程师翻日志，直接导出task_id的完整消息链，就是一份天然的审计证据链。
6.3 我的个人体会：Agency不是终点，而是人机协作的新起点
做了两年Agency项目，最大的感悟是：我们不是在教机器思考，而是在设计一种人类智慧的外延结构。当Planner拆解任务时，它在模仿项目经理的思维框架；当Critic挑刺时，它在复刻资深QA的质疑精神；当Executor沉默执行时，它在承担一线工程师的务实担当。这些Agent本身没有意识，但它们组成的系统，却呈现出一种接近人类团队的协作质感。
上周我看到一个有趣现象：Critic连续三次否决了Planner的方案，Planner在第四次尝试时，主动在fallback_options里加了一条：“若Critic再次拒绝，启用人工审核通道”。那一刻，我意识到——这个系统已经开始学习“何时该谦卑”。它没变成人，但它学会了尊重人的判断边界。这或许就是Agency最迷人的地方：它不取代人，而是让人，更像人。
