Kimi Code CLI 不是简单的"请求-响应"模式,而是一个迭代的、目标导向的过程:
Run (运行)
├── Turn (回合) - 对应一次用户输入
│ ├── Step (步骤) - 对应一次 LLM 调用
│ │ ├── LLM API 调用
│ │ ├── 工具调用(可能有多个,并行执行)
│ │ └── 上下文更新
│ │ └── (可能重复多个 Step)
│ └── 回合结束
└── (可能重复多个 Turn)
async def _agent_loop(self) -> TurnOutcome:
step_no = 0
while True:
step_no += 1
if step_no > self._loop_control.max_steps_per_turn:
raise MaxStepsReached(...)
# 上下文压缩检查
if self._context.token_count + reserved >= max_context_size:
await self.compact_context()
step_outcome = await self._step()
# 处理 D-Mail 回溯
if back_to_the_future:
await self._context.revert_to(checkpoint_id)
class KimiToolset:
def __init__(self):
self._tool_dict: dict[str, ToolType] = {}
self._mcp_servers: dict[str, MCPServerInfo] = {}
工具通过构造函数声明依赖,系统自动装配:
class Shell:
def __init__(self, config: Config, approval: Approval):
# 自动注入 Config 和 Approval 实例
...
| 工具 | 功能 |
|---|---|
| ReadFile/WriteFile | 文件读写 |
| Shell | 安全执行命令 |
| Search/FetchURL | 网络搜索和获取 |
| Task | 创建子代理任务 |
| SetTodoList | 管理待办事项 |
MCP(Model Context Protocol)是开放协议,支持集成外部工具服务:
{
"mcpServers": {
"my-server": {
"url": "https://api.example.com/mcp",
"headers": {"Authorization": "Bearer token"}
}
}
}
class LaborMarket:
def __init__(self):
self.fixed_subagents: dict[str, Agent] = {} # 固定子代理
self.dynamic_subagents: dict[str, Agent] = {} # 动态子代理
主代理分配任务给子代理:
async def __call__(
self,
description: str,
subagent_name: str | None = None,
prompt: str | None = None,
) -> ToolReturnValue:
独特的 D-Mail 机制,灵感来自《命运石之门》:
class DenwaRenji:
def send_dmail(self, checkpoint_id: int, message: str):
"""发送 D-Mail(只能调用一次)"""
def fetch_pending_dmail(self) -> DMail | None:
"""获取待处理的 D-Mail"""
子代理可以向父代理"发送消息到过去",触发上下文回溯。
book/ 目录。
还没有人回复