ARTICLE DETAIL

资讯详情

深耕网站视觉设计与运营推广的一线实战洞察。

多 Agent 协作实战:Python 实现最小版 Bot Mode 编排与上下文隔离

多 Agent 协作实战:Python 实现最小版 Bot Mode 编排与上下文隔离 很多做 AI Agent 的同学都会碰到同一种尴尬单个 Agent 写周报、查资料、改代码都能完成可任务一旦带上“然后”“同时”“分别”这类词比如“先查最近 1 小时的线上错误日志再做根因分析按模块生成日报最后把结论同步给对应负责人”单 Agent 就开始崩盘。不是模型变笨了而是单 Agent 处理长链路任务时有结构性瓶颈上下文越来越长、无关信息越来越多、中间一个步骤出错后面全跟着错而且每一步的结果很难拆出来单独复用。于是“多 Agent 协作”就成了 2026 年 AI Agent 领域最被频繁讨论的解法之一。这篇文章要讲的是 Hermes Bot Mode 这一类执行模式一个主 Bot 负责承接用户请求把复杂任务拆成多个子任务分派给多个分工明确的 Agent 分别执行再汇总结果。本文会从架构原理讲到适用边界最后用 Python 手写一个最小版 Bot Mode演示多 Agent 是如何“坐下来开一场会”的。1. 多个 AI Agent 协同究竟解决了什么问题先回到最朴素的开发场景。假设你手上有一个订单服务线上突然出现一批超时报警。你打开 Hermes 这类 Agent 工具输入“分析订单服务最近 1 小时错误日志定位根因生成日报发给负责人。”如果只用一个 Agent 来做这件事会发生什么Agent 会把“查日志”“分析根因”“写日报”“想发送方式”全部塞进同一个上下文里。每一次工具调用返回的日志、中间分析、改写的文案都会追加到上下文末尾。任务越长上下文中无效信息占比越高Agent 越容易忽略关键线索甚至把前一步的分析结论和下一步任务搞混。这不是模型能力不够而是架构选择出了问题。单 Agent 的本质是“一人分饰多角”而多 Agent 协同的本质是“专人专事各自负责一段”。Bot Mode 的解决思路非常直接保留一个主 Agent 作为协调者它不直接干脏活累活而是负责理解用户意图、拆任务、盯进度、合并结果。真正查日志、写代码、做分析的是由它调度的多个专业 Agent。这些子 Agent 各自拥有独立的上下文互不污染完成后只把结论交回给主 Agent。这个设计带来的直接收益有三个上下文隔离。每个 Agent 只关注自己的任务片段不会因为“写日报”而被迫读取几千条原始日志。出错域缩小。某个子 Agent 执行失败只需要重跑这一步不需要重跑整个任务。能力可复用。日志分析 Agent 可以被任何任务调度而不是每次重新写一遍提示词。也就是说Bot Mode 真正降低的不是 API 调用成本而是“长链路任务失败后的返工成本”。2. Bot Mode 到底是什么从单 Agent 到多 Agent 的演进2.1 三种常见形态把 AI Agent 的执行模型拉出来横向对比会发现多数工具都经历过这三个阶段形态交互方式上下文管理典型问题单 Agent 单轮用户提问Agent 回答一次请求一个上下文无法处理复杂任务单 Agent 多工具一个 Agent 反复调用工具所有工具结果堆积在同一上下文长任务后上下文污染多 Agent 协作主 Agent 拆分子 Agent 并行执行每个子 Agent 独立上下文协议设计和编排较复杂Bot Mode 属于第三种形态。它名字里的“Bot”不是指聊天机器人而是指“常驻的主 Agent”。这个主 Agent 类似项目负责人不亲自写代码但清楚每个成员擅长什么。在 Hermes 这类工具中Bot Mode 往往表现为一种可切换的执行模式你切到 Bot Mode 后可以同时加载多个 Agent 和 Skill用自然语言或命令把一个任务“甩”给主 Bot由它来决定如何分发。具体入口和命令会随版本变化这里不绑定某个版本只讨论它背后的执行模型。2.2 Agent 和 Skill 的区别很多人没搞清楚热门搜索词里经常同时出现 “Agent” 和 “Skill”很多初学者会把它们当成同一件事。这里做一个明确区分Skill 是能力。它对应的是一个可被调用的函数、脚本或提示词模板比如“读取 ES 日志”“生成 PPT”“扫描端口”。它解决的是“能不能做”的问题。Agent 是角色。它包含系统提示词、上下文窗口、可用 Skill 集合和自主决策逻辑。它解决的是“要不要做、怎么做”的问题。更直白一点Agent 手里能用的每一张牌就是 Skill。同一个 Agent 可以拥有多个 Skill同一个 Skill 也可以被多个 Agent 复用。在 Bot Mode 里主 Agent 根据任务类型选择合适的子 Agent子 Agent 再调用自己的 Skill 去完成任务。这种分层让 Skill 的复用率变得很高也让 Agent 的职责边界非常清楚。3. 什么场景适合 Bot Mode什么场景不适合多 Agent 协作不是银弹。用错了场景只会让系统更慢、更贵、更难排查。3.1 适合的场景第一类是跨上下文任务。比如日志分析、数据库查询、多仓库代码检索这类任务天然需要读取大量原始数据如果让一个 Agent 全程握着这些数据上下文很容易爆炸。拆成独立 Agent 后每个 Agent 只保留与自己相关的片段。第二类是需要多角色交叉验证的任务。比如“写一段代码然后让另一个 Agent 做 Code Review”或者“生成一份发布清单再由安全 Agent 检查一遍”。这类任务想要的结果不是一次生成而是多次检查角色隔离很重要。第三类是高频复用的流水线任务。比如每天的日报生成、每周的日志巡检、每次发布前的检查清单。把这些固定动作沉淀成 Agent Skill之后只需要主 Agent 统一调度。3.2 不适合的场景如果是“今天天气怎么样”“这段代码什么意思”这类单点问答不要用多 Agent。你只需要一个普通 Agent 加一个问答模型多 Agent 的优势完全发挥不出来反而白白增加一次编排调用。如果是延迟敏感的实时交互场景比如在线客服、实时语音助手多 Agent 也需要慎重。多 Agent 协作天然意味着多次模型调用即使子 Agent 是并行执行的整体响应延迟也会明显高于单 Agent。判断标准很简单任务是否足够复杂拆开之后的收益是否大于编排开销如果答案不确定先跑通单 Agent 再逐步演进。4. 环境准备与前置条件本文的示例代码不绑定特定框架只依赖 Python 和常见的 HTTP 客户端。无论你最终用的是 Hermes 还是自研的多 Agent 系统先理解这套最小实现再迁移到自己的框架会容易得多。建议环境如下Python 3.10 以上版本以本地环境为准不影响核心逻辑requests库用于调用 LLM API 和 ES REST API一个支持 Chat Completions 格式的 LLM 服务以及对应的 API Key一个可访问的 Elasticsearch 实例用于日志查询演示没有 ES 环境也可以先跳过日志部分安装依赖pip install requests python-dotenv在项目目录下创建.env文件# LLM 服务配置 LLM_API_URLhttps://your-llm-service/v1/chat/completions LLM_API_KEYsk-your-key LLM_MODELyour-model-name # Elasticsearch 配置 ES_URLhttp://localhost:9200 ES_USERelastic ES_PASSWORDchange-me这里有一点要注意LLM_MODEL的填写必须和你使用的模型服务保持一致不同服务支持的模型名不同别照抄网上的配置。5. 核心流程拆解一个复杂任务是如何被拆开的在写代码之前先把 Bot Mode 的完整执行链路过一遍。一个主 Bot 处理复杂任务时大致会经历以下五个环节第一步接收用户请求。主 Agent 拿到一段自然语言描述比如“分析订单服务错误日志并生成日报”。第二步任务规划。主 Agent 调用规划模块把用户请求拆成多个子任务。这一步通常由 LLM 完成输出一个结构化列表例如“日志分析 Agent 负责查询和分析日志”“日报 Agent 负责整理输出”。第三步任务分派。主 Agent 根据子任务类型选择匹配的 Agent把各自的任务描述传递过去。这里可以选择串行执行或并行执行。并行能提速但要注意子 Agent 之间是否有依赖关系。第四步子 Agent 执行。每个子 Agent 在自己独立的上下文里工作按需调用 Skill。比如日志分析 Agent 可能会调用query_es_logs这个 Skill 去 ES 里拉数据。第五步结果汇总与反馈。子 Agent 把结论返回给主 Agent主 Agent 判断是否达到最终目标。如果某些子任务失败主 Agent 可以决定重试或更换策略最后把统一结果反馈给用户。这个流程的本质是把“一次巨大的 Prompt”改写成了“多次专业的 Prompt 明确的输入输出协议”。协议越清晰多 Agent 系统越稳定。6. 完整示例用 Python 实现一个最小版 Bot Mode下面我们实现一个日志分析场景的 Bot Mode。两个子 Agent 分工LogAgent负责查日志、做根因分析SummaryAgent负责把分析结果改写成日报。6.1 工具层LLM 调用与 ES 日志查询创建toolkit.py负责所有外部依赖的最底层操作# toolkit.py 基础工具LLM 调用、ES 日志查询。 import os import requests LLM_API_URL os.getenv(LLM_API_URL, https://your-llm-service/v1/chat/completions) LLM_API_KEY os.getenv(LLM_API_KEY, ) LLM_MODEL os.getenv(LLM_MODEL, ) def chat(system: str, user: str) - str: 调用兼容 Chat Completions 格式的 LLM 服务。 resp requests.post( LLM_API_URL, headers{Authorization: fBearer {LLM_API_KEY}}, json{ model: LLM_MODEL, messages: [ {role: system, content: system}, {role: user, content: user}, ], temperature: 0.2, }, timeout60, ) resp.raise_for_status() return resp.json()[choices][0][message][content] def query_es_logs(index: str, body: dict, size: int 100) - list: 查询 Elasticsearch 日志返回命中列表。 es_url os.getenv(ES_URL, http://localhost:9200) auth (os.getenv(ES_USER, elastic), os.getenv(ES_PASSWORD, )) resp requests.get( f{es_url}/{index}/_search, json{**body, size: size}, authauth, timeout30, ) resp.raise_for_status() return resp.json()[hits][hits]这段代码里的chat函数是后续所有 Agent 的“嘴”负责和模型对话query_es_logs是日志分析 Agent 的一只手负责从 ES 里抓数据。如果只想本地验证流程可以在这两个函数里加一个mock分支当LLM_API_KEY为空时返回固定文本。这会让我们在没有外部依赖时也能跑通整个 Bot Mode 流程。6.2 定义 Agent 与 Bot Mode 编排核心创建bot_mode.py实现 Agent 的基类和主 Bot 的调度逻辑# bot_mode.py 一个最小版 Bot Mode 实现主 Bot 负责拆分任务多个 Agent 负责执行。 import json from dataclasses import dataclass, field from typing import Callable dataclass class Agent: name: str system_prompt: str skills: dict[str, Callable[..., str]] field(default_factorydict) def run(self, task: str) - str: 默认执行逻辑把任务交给 LLM 处理。 from toolkit import chat return chat(self.system_prompt, task) class BotMode: def __init__(self, agents: list[Agent]): self.agents agents def plan(self, user_request: str) - list[dict[str, str]]: 把用户请求拆分成多个子任务每个子任务指定执行者。 from toolkit import chat planner_prompt ( 你是任务编排器。请把用户请求拆分为多个子任务 每个子任务指定一个执行者。 输出 JSON 数组格式为 [{agent: agent名称, task: 子任务描述}]。 不要输出其他内容。 ) raw chat(planner_prompt, user_request) try: return json.loads(raw) except json.JSONDecodeError: # 容错从返回内容中截取 JSON 数组部分 start raw.find([) end raw.rfind(]) 1 return json.loads(raw[start:end]) def dispatch(self, user_request: str) - dict[str, str]: 分派任务给各个 Agent并返回每个 Agent 的结果。 plan_items self.plan(user_request) agent_map {agent.name: agent for agent in self.agents} results {} for item in plan_items: agent_name item[agent] task item[task] if agent_name not in agent_map: results[agent_name] 错误未找到该 Agent continue results[agent_name] agent_map[agent_name].run(task) return resultsplan方法里做了一次 JSON 解析容错这是实际开发中很容易踩的坑LLM 返回的“JSON”经常带着 Markdown 代码块标记或额外解释文字直接json.loads会失败。从第一个[截到最后一个]这个技巧虽然粗暴但在多数场景下都有效。6.3 组装主流程并运行创建main.py注册两个子 Agent一个负责日志分析一个负责日报生成# main.py from bot_mode import Agent, BotMode from toolkit import chat, query_es_logs class LogAgent(Agent): def run(self, task: str) - str: hits query_es_logs( order-service-logs, { query: { range: { timestamp: {gte: now-1h, lte: now} } } }, size20, ) log_text \n.join(str(h[_source]) for h in hits) return chat( self.system_prompt, f{task}\n\n以下是最近一小时订单服务日志\n{log_text}, ) class SummaryAgent(Agent): def run(self, task: str) - str: return chat(self.system_prompt, task) log_agent LogAgent( namelog_agent, system_prompt你是日志分析专家。请根据给出的日志输出分析错误根因和影响范围给出排查线索。, ) summary_agent SummaryAgent( namesummary_agent, system_prompt你是日报整理专家。请把日志分析结论改写成适合发送给团队负责人的日报包含影响范围和下一步建议。, ) bot BotMode(agents[log_agent, summary_agent]) if __name__ __main__: request 分析订单服务最近一小时错误日志定位根因并生成日报简报 results bot.dispatch(request) for agent_name, output in results.items(): print(f----- {agent_name} -----) print(output)这个主流程的价值在于它展示了两件事LogAgent在执行时先通过query_es_logsSkill 拿到实时日志再结合系统提示词做分析而不是让 LLM 凭空猜测。SummaryAgent拿到的输入只有task字符串它不会接触原始日志只需要把log_agent的结论改写成日报。这就是上下文隔离的直观体现。实际生产环境中SummaryAgent应该拿到的是log_agent的输出而不是直接拿到用户原始请求。为了演示主流程结构这里保持简化但你应该在dispatch中把前序 Agent 的结果拼接进后续 Agent 的任务描述。6.4 使用 Skill 注册机制扩展能力上面的示例把日志查询逻辑硬编码在了LogAgent.run里。更合理的做法是把“查询日志”抽象成一个 Skill注册到 Agent 上# skill_example.py 把日志查询封装成 Skill 并注册到 Agent 上。 from bot_mode import Agent from toolkit import query_es_logs def skill_query_order_error_logs(params: str) - str: Skill查询订单服务最近一小时的错误日志。 hits query_es_logs( order-service-logs, { query: { bool: { must: [ {range: {timestamp: {gte: now-1h, lte: now}}}, {match: {level: ERROR}}, ] } } }, size50, ) return \n.join(str(h[_source]) for h in hits) log_agent Agent( namelog_agent_with_skill, system_prompt你是日志分析专家请结合工具返回的日志内容输出分析结论。, skills{ query_order_error_logs: skill_query_order_error_logs, }, ) # 使用示例 # result log_agent.run(请分析最近一小时订单服务的错误日志)这种“Agent Skills”的组合方式就是前面讲的 Agent 和 Skill 分层思想。Agent 负责决定要不要调用、何时调用Skill 只负责执行具体的工具逻辑。后续如果要支持“生成 PPT”“发送钉钉消息”只需要新增一个 Skill然后注册到对应 Agent 上完全不需要改动编排逻辑。7. 运行效果与验证方法先确认.env里的配置都已经生效然后执行python main.py如果 LLM 服务正常、ES 数据存在输出大约是这种形态----- log_agent ----- 订单服务最近 1 小时共出现 23 条 ERROR 日志主要集中在 order-api 实例上。 根因可能是支付回调超时导致订单状态长时间停留在“待支付” 而超时重试机制又会反复触发数据库更新形成性能热点。 建议先查看 payment_service 的响应时间指标。 ----- summary_agent ----- 【订单服务日志日报】 影响范围order-api 实例涉及支付回调链路。 错误特征23 条 ERROR支付回调超时集中在 14:50-15:10。 核心结论支付服务响应变慢是主要诱因。 下一步建议1. 联系支付网关确认回调状态2. 调整超时阈值3. 增加局部熔断。如果输出完全对不上优先检查三件事第一容器日志或终端是否有异常堆栈。如果是requests.exceptions.ConnectionError通常是网络不通或 URL 配置错误。第二ES 查询返回是否为空。如果hits为空说明索引名写错、时间范围没有数据或者认证失败。可以先在浏览器或 Postman 里直接用同一个查询条件试一次。第三任务规划的结果是否符合预期。如果plan_items里只有一条任务说明规划 Prompt 没有把任务拆开可以增加示例输出让 LLM 照着格式生成。8. 常见问题与排查思路我把实际开发中高频出现的问题整理成一张排查表问题现象可能原因排查方式解决方案json.loads解析失败LLM 返回了 Markdown 代码块或多余文字打印raw原始内容观察输出格式从第一个[截取到最后一个]同时要求 LLM 只输出 JSON所有子任务都分配给同一个 Agent规划 Prompt 中没有给出 Agent 列表查看plan的返回结果在规划 Prompt 中显式列出可用 Agent 名称和职责Agent 无法执行日志查询报 401/403ES 用户名密码错误或权限不足用 curl 直接调用 ES API 验证检查ES_PASSWORD确认账号只有必要的只读权限子 Agent 之间结果串接错误后续 Agent 的输入没有拼接前序结果查看传入Agent.run的task内容在dispatch中把前序 Agent 的输出写进后续任务的 user promptAgent 需要克隆仓库时失败网络连通、仓库地址错误或权限不足手动执行一次 git clone 复现检查仓库地址、认证方式和网络连通性不要在 Agent 的 Prompt 里硬编码敏感 Token运行后上下文仍然很大子 Agent 调用的工具返回了完整原文检查工具返回值的大小在 Skill 层先做字段裁剪或聚合只返回关键摘要其中“子 Agent 之间结果串接错误”是新手最容易忽略的。多 Agent 不是把任务丢出去就结束了它依赖清晰的输入输出协议。每个 Agent 返回什么结构、下一个 Agent 的输入从哪里拼接都应该在设计阶段定好而不是靠运气让 LLM 自动理解。9. 生产环境使用建议从一个能跑的 Demo 到一个能上线的多 Agent 服务中间还差着不少工程细节。以下几个建议来自实际项目经验优先级从高到低排列。9.1 所有外部调用都需要超时和重试LLM 服务和 ES 服务都可能变慢或抖动。chat和query_es_logs都需要设置超时时间并配合重试策略。重试要带退避不要在同一时刻把所有请求打出去。9.2 最小权限原则LogAgent 只需要 ES 日志的只读权限就不要给它写权限SummaryAgent 不需要访问数据库就不要在它的工作区里配置数据库连接。Agent 的权限边界应该像人类员工一样严格定义。9.3 可观测性是上线前提多 Agent 系统最大的问题是不透明。用户看到的是一个最终结果但开发者需要知道任务被拆成了几步每一步花了多久每一步调用了什么工具因此从第一天就要打印或上报每一层的关键日志包括规划结果、子任务分派、Skill 调用次数、每个 Agent 的返回摘要。9.4 优先保证幂等如果某个 Agent 执行到一半失败重试时会不会产生重复数据比如“创建订单工单”这类有副作用的 Skill重试前要检查是否已经执行过。尽量设计成“先查询再写入”的流程或者让写入操作本身支持去重。9.5 不要一开始就追求复杂编排如果当前任务用 3 个 Agent 能做就不要设计 10 个 Agent。每个 Agent 的增加都意味着新的失败点和新的延迟。先用最少 Agent 跑通主链路再根据真实问题演进。10. 总结与后续学习方向Bot Mode 的价值不在于把系统变得更大更复杂而在于让多个 AI Agent 在合适的边界内协作主 Agent 管编排子 Agent 管执行Skill 管能力。它能有效缓解长链路任务中的上下文污染和错误扩散问题但也会引入新的编排开销和协议设计成本。适合它的场景是复杂、可拆解、可复用的任务不适合单点问答和低延迟交互。如果你正准备上手建议按这个顺序实践先跑通本文的最小版 Bot Mode理解规划、分派、执行、汇总这四个环节然后为你的真实场景写一个 Skill注册到对应的 Agent 上最后再考虑并行执行、结果校验、日志监控这些生产级能力。多 Agent 协作的下一步通常还会涉及到 Agent 之间的“通信协议”设计、工具调用的 schema 校验、记忆共享与隔离机制。这些内容都值得继续深入但在那之前先把一个最小的多 Agent 协作链路跑起来比什么都重要。
返回列表