Kimi Code CLI 的多 Agent 架构: 一场精心设计的分工协作

> 当你用 Kimi CLI 写代码时, 有没有想过: 如果有一个专门的 Agent 负责写代码, 另一个专门负责审查, 还有一个专门写测试, 它们之间如何协作? 本文将带你走进 Kimi CLI 的多 Agent 世界. --- ## 从一个场景说起 假设你要开发一个新功能. 理想的工作流可能是这样: 1. **需求分析 Agent**: 理解需求, 拆解任务 2. **代码编写 Agent**: 实现核心逻辑 3. **代码审查 Agent**: 检查质量和潜在问题 4. **测试编写 Agent**: 编写单元测试和集成测试 5. **文档编写 Agent**: 更新相关文档 这不是科幻. Kimi Code CLI 从设计之初就考虑了这种多 Agent 协作场景. --- ## 核心概念: LaborMarket (劳动力市场) Kimi CLI 把子 Agent 看作"劳动力", 主 Agent 是"雇主". 所有可用的子 Agent 都注册在一个叫做 `LaborMarket` 的地方: ```python class LaborMarket: def __init__(self): # 预定义的固定子 Agent self.fixed_subagents: dict[str, Agent] = {} # 运行时创建的动态子 Agent self.dynamic_subagents: dict[str, Agent] = {} ``` 这个命名很有意思 -- 它暗示了 Agent 之间的关系: **雇佣与被雇佣**. --- ## 两种子 Agent 模式 ### 1. Fixed Subagent (固定子 Agent) **定义方式**: 在 `agent.yaml` 中预定义 ```yaml agent: name: "main" system_prompt_path: "prompt.md" tools: ["shell", "file", "task"] subagents: code_reviewer: path: "agents/reviewer/agent.yaml" description: "专门审查代码质量和风格, 擅长发现潜在 bug" test_writer: path: "agents/tester/agent.yaml" description: "编写全面的单元测试和集成测试" doc_writer: path: "agents/doc/agent.yaml" description: "编写技术文档和注释" ``` **特点**: - 随主 Agent 一起加载 - 有独立的系统提示词和工具集 - 适合**长期存在的专业角色** ### 2. Dynamic Subagent (动态子 Agent) **创建方式**: 运行时通过代码动态添加 ```python # 主 Agent 可以在执行过程中创建临时子 Agent runtime.labor_market.add_dynamic_subagent( name="temp_analyzer", agent=Agent(...) ) ``` **特点**: - 生命周期由任务决定 - 适合**临时性、一次性的任务** - 与主 Agent 共享 LaborMarket (可以看到其他动态子 Agent) --- ## 如何调用子 Agent: Task 工具 主 Agent 通过 `Task` 工具"雇佣"子 Agent: ```python class Task(CallableTool2[Params]): # Task 工具: 委派任务给子 Agent class Params(BaseModel): description: str # 任务简短描述 (3-5 词) subagent_name: str # 子 Agent 名称 prompt: str # 详细任务描述 ``` **使用示例**: ``` 用户: "帮我实现一个用户登录功能" 主 Agent 思考后决定: 1. 先让 code_writer 写核心代码 2. 再让 code_reviewer 审查 3. 最后让 test_writer 写测试 于是它调用: - Task(description="编写登录代码", subagent_name="code_writer", prompt="...") - Task(description="审查代码", subagent_name="code_reviewer", prompt="...") - Task(description="编写测试", subagent_name="test_writer", prompt="...") ``` --- ## 上下文隔离: 完全独立的"世界" **关键设计**: 每个子 Agent 有**完全独立的上下文**. ```python async def _run_subagent(self, agent: Agent, prompt: str): # 1. 创建独立的上下文文件 subagent_context_file = await self._get_subagent_context_file() # 结果: ~/.kimi/sessions/xxx_sub_1.jsonl context = Context(file_backend=subagent_context_file) # 2. 创建独立的 Soul soul = KimiSoul(agent, context=context) # 3. 运行 (完全隔离) await run_soul(soul, prompt, ...) ``` **这意味着**: - 子 Agent **看不到**主 Agent 的上下文 - 子 Agent **看不到**其他子 Agent 的上下文 - 必须通过 `prompt` 参数传递所有必要信息 **为什么这样设计?** 1. **避免干扰**: 子 Agent 的试错不会污染主 Agent 的上下文 2. **可重入**: 同一个子 Agent 可以被多次调用 3. **可调试**: 每个子 Agent 的上下文文件独立保存, 便于排查问题 --- ## 事件透传: 子 Agent 的"声音"如何被听到 虽然上下文隔离, 但子 Agent 的事件可以透传到主 Agent 的 UI: ```python def _super_wire_send(msg: WireMessage) -> None: if isinstance(msg, ApprovalRequest | ApprovalResponse | ToolCallRequest): # 审批请求直接发送到主 Wire super_wire.soul_side.send(msg) return # 其他事件包装成 SubagentEvent event = SubagentEvent( task_tool_call_id=current_tool_call_id, event=msg, ) super_wire.soul_side.send(event) ``` **效果**: 你可以在主界面看到子 Agent 的执行过程, 就像在看一场"直播". --- ## 一个完整的例子 假设有这样的 Agent 配置: ```yaml # main/agent.yaml agent: name: "project_manager" system_prompt_path: "pm_prompt.md" tools: ["shell", "file", "task", "todo"] subagents: backend_dev: path: "backend/agent.yaml" description: "后端开发专家, 擅长 Python/FastAPI" frontend_dev: path: "frontend/agent.yaml" description: "前端开发专家, 擅长 React/TypeScript" qa_engineer: path: "qa/agent.yaml" description: "QA 工程师, 擅长测试用例设计" ``` **工作流**: ``` 用户: "开发一个 Todo 应用" project_manager: 1. 分析需求, 拆解任务 2. 创建 Todo 列表 3. 调用 Task(subagent_name="backend_dev", prompt="设计 API 接口...") 4. 调用 Task(subagent_name="frontend_dev", prompt="设计 UI 组件...") 5. 调用 Task(subagent_name="qa_engineer", prompt="设计测试用例...") 6. 整合结果, 向用户汇报 ``` --- ## 限制与边界 ### 1. LLM 共享 **当前实现**: 所有 Agent 共享同一个 LLM 实例. ```python # Runtime 复制时 llm=self.llm, # 同一个对象 ``` **这意味着**: - 不能让 backend_dev 用 GPT-4, frontend_dev 用 Claude - 所有 Agent 使用相同的模型参数 ### 2. 并行限制 **当前实现**: Task 工具是**顺序执行**的. ```python # 伪代码 result1 = await task(subagent1, prompt1) # 等待完成 result2 = await task(subagent2, prompt2) # 再执行 ``` **不能**同时调用多个子 Agent. ### 3. 状态共享 子 Agent 之间**无法直接通信**, 必须通过主 Agent 中转: ``` subagent_A -> main_agent -> subagent_B ``` --- ## 与 OpenClaw 的对比 | 特性 | Kimi CLI | OpenClaw | |------|----------|----------| | **设计理念** | 层级分工 (主-子) | 会话隔离 (独立 sessions) | | **子 Agent 创建** | 配置预定义 / 动态添加 | `sessions_spawn` | | **上下文隔离** | 文件级隔离 | Session 级隔离 | | **LLM 配置** | 共享 LLM | 每个 session 独立配置 | | **并行执行** | 顺序执行 | 支持并行 spawn | | **通信方式** | Task 工具 + Wire 透传 | `sessions_send` | | **适用场景** | 专业化分工 | 多模型对比、并行任务 | **简单比喻**: - Kimi CLI 像是一个**项目经理**带几个**专业团队成员** - OpenClaw 像是你同时开了几个**独立的聊天窗口** --- ## 实际应用场景 ### 场景 1: 代码审查流水线 ```yaml subagents: linter: description: "检查代码风格和格式" security_scanner: description: "扫描安全漏洞" performance_analyzer: description: "分析性能瓶颈" ``` ### 场景 2: 多语言项目 ```yaml subagents: rust_expert: description: "Rust 核心模块开发" python_expert: description: "Python 绑定和脚本" js_expert: description: "前端 JavaScript 部分" ``` ### 场景 3: 复杂任务拆解 ```yaml subagents: researcher: description: "研究技术方案和最佳实践" architect: description: "设计系统架构" implementer: description: "具体实现代码" ``` --- ## 总结 Kimi Code CLI 的多 Agent 架构是一个**精心设计的分工协作系统**: **核心设计**: - LaborMarket: 统一管理可用 Agent - Task 工具: 标准化的"雇佣"接口 - 上下文隔离: 完全独立的执行环境 - 事件透传: 可见但不可干扰 **适用场景**: - 需要多个专业角色的复杂任务 - 代码审查、测试、文档等流水线 - 多语言、多模块的项目开发 **不适用场景**: - 需要不同 LLM 模型的对比 - 高度并行的独立任务 - Agent 之间需要频繁通信 理解这套架构, 你就能像搭积木一样, 组建自己的"AI 开发团队". --- ## 参考 - 源码: `src/kimi_cli/soul/agent.py` (LaborMarket, Runtime) - 源码: `src/kimi_cli/tools/multiagent/task.py` (Task 工具) - 源码: `src/kimi_cli/agentspec.py` (Agent 配置) - 配置示例: `src/kimi_cli/agents/default/agent.yaml` --- *"一个人走得快, 一群人走得远." -- 在 AI 编程的世界里, 这句话同样适用.*