Learn Claude Code
约 3610 个字 885 行代码 预计阅读时间 23 分钟
1 Core Loop
1.1 最小 Agent Loop
把 “LLM + Tools” 连接成一个能持续推进任务的 Main Loop
把工具输出回头喂给 LLM 用于下一步推理，而不是仅仅输出一段文本
名词解释
-
turn：Agent 的一轮行动，在我们的最小版本中 -
tool_result_block：工具执行结果被打包回工作流的形态 -
state：主循环需要一直携带的 Context，在最小版本中包含：state = { 'messages': [ {"role": "user", "content": "..."}, {"role": "assistant", "content": [...]}, ... ], 'turn_count': 1, # 当前轮次 'transition_reason': None # 这一轮结束后，为什么要继续下一轮 }msgs不是聊天记录展示，而是 LLM 下一轮需要读取的 Ctx- MVP 中的
transition_reason只有 "tool_result" 一种取值，即：刚执行完工具，需要继续
实现
# 基于用户请求初始化 msgs
messages = [{"role": "user", "content": query}]
# MVP 实现
def agent_loop(state):
while True:
# 调用 LLM
response = client.messages.create(
model=MODEL,
system=SYSTEM,
messages=state["messages"],
tools=TOOLS,
max_tokens=8000,
)
# 将 response 写回 Context
state["messages"].append({
"role": "assistant",
"content": response.content,
})
# 不调用工具 => 退出
if response.stop_reason != "tool_use":
state["transition_reason"] = None
return
# 调用工具
results = []
for block in response.content:
if block.type == "tool_use":
output = run_tool(block)
results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": output,
})
# 将工具调用结果写回 Context
state["messages"].append({"role": "user", "content": results})
# 更新轮次信息
state["turn_count"] += 1
state["transition_reason"] = "tool_result"
1.2 Tool-using
追加工具不需要修改 Main Loop，只需要在 handler map + schema 中进行注册
handler map 是一个由 tool_name: handler_func 构成的 dict
对 1.1 中的工具调用 block 进行优化，使其能从 handler map 中调用多种工具
# handler map
TOOL_HANDLERS = {
"bash": lambda **kw: run_bash(kw["command"]),
"read_file": lambda **kw: run_read(kw["path"], kw.get("limit")),
...
}
# tool schema（这部分其实会通过 tools=TOOLS 的形式塞在 prompt 里）
TOOLS = [
{
"name": "bash",
"description": "Run a shell command.",
"input_schema": {
"type": "object",
"properties": {"command": {"type": "string"}},
"required": ["command"]
}
},
{
"name": "read_file",
"description": "Read file contents.",
"input_schema": {
"type": "object",
"properties": {
"path": {"type": "string"},
"limit": {"type": "integer"}
},
"required": ["path"]
}
},
...
]
# mod: tool-using block
for block in response.content:
if block.type == "tool_use":
handler = TOOL_HANDLERS.get(block.name)
output = handler(**block.input) if handler else f"Unknown tool: {block.name}"
results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": output,
})
handler map 好像不能完全兜住工具使用
- 是的，在更完备的实现中，Tool-using 应该是一个 Layer，而不仅仅是 Handler Map + Schema Map
- 你还需要往里塞进：鉴权、curMsg + AppState、MCP Client、文件读取缓存、通知 + query 追踪 ...
如何防止 read_file / write_file 等工具从指定工作区中逃逸
添加一个 safe_path 作为沙箱（其实只是 check 一下指定文件的 path 是否在工作区中）
def safe_path(p: str) -> Path: # p 在理想状态下为 相对路径
path = (WORKDIR / p).resolve() # 拼接 WORKDER + p，并转换为绝对路径
if not path.is_relative_to(WORKDIR): # 防止 p 通过 ../ 或者 绝对路径 跳出工作区
raise ValueError(f"Path escapes workspace: {p}")
return path
def run_read(path: str, limit: int = None) -> str:
text = safe_path(path).read_text() # 在此处连接路径
lines = text.splitlines()
if limit and limit < len(lines):
lines = lines[:limit]
return "\n".join(lines)[:50000]
如何处理内部 messages 消息列表格式与 LLM API 协议不兼容
-
LLM 的 API 协议给出的消息存在 3 点约束
- 每个
tool_useblock 必须存在对应的tool_result（通过tool_use_id关联） user - assitant必须交替出现（两条同角色消息不得连续）- 只接受协议定义的字段（ msg 中的一些内部字段会导致 400 错误）
- 每个
-
Solution：加一个函数，在 发送请求前 做一次规范化
messages 列表是系统的内部表示, API 看到的是规范化后的副本，两者不是一个东西
def normalize_messages(messages: list) -> list:
normalized = []
for msg in messages:
# Step 1: 去除内部字段
clean = {"role": msg["role"]}
if isinstance(msg.get("content"), str):
clean["content"] = msg["content"]
elif isinstance(msg.get("content"), list):
clean["content"] = [
{
k: v for k, v in block.items() \
if k not in ("_internal", "_source", "_timestamp")
} for block in msg["content"]
]
normalized.append(clean)
# Step 2: tool_result 配对补齐
# 收集所有已有的 tool_result ID
existing_results = set()
for msg in normalized:
if isinstance(msg.get("content"), list):
for block in msg["content"]:
if block.get("type") == "tool_result":
existing_results.add(block.get("tool_use_id"))
# 找出缺失配对的 tool_use (被取消), 插入占位 result
for msg in normalized:
if msg["role"] == "assistant" and isinstance(msg.get("content"), list):
for block in msg["content"]:
if (block.get("type") == "tool_use"
and block.get("id") not in existing_results):
# 在下一条 user 消息中补齐
normalized.append({"role": "user", "content": [{
"type": "tool_result",
"tool_use_id": block["id"],
"content": "(cancelled)",
}]})
# Step 3: 合并连续同角色消息
merged = [normalized[0]] if normalized else []
for msg in normalized[1:]:
if msg["role"] == merged[-1]["role"]:
# 合并内容
prev = merged[-1]
prev_content = prev["content"] if isinstance(prev["content"], list) \
else [{"type": "text", "text": prev["content"]}]
curr_content = msg["content"] if isinstance(msg["content"], list) \
else [{"type": "text", "text": msg["content"]}]
prev["content"] = prev_content + curr_content
else:
merged.append(msg)
return merged
1.3 Session 待办更新
LLM 一口气列出很多步骤后，跑一半自己搞忘了怎么办
外显当前会话计划并随任务推进不断更新（⚠️ 不是跨对话的持久化任务系统）
-
Pipeline
其中，
complted* N +in_progress* 1 +pending* M -
数据结构
-
PlanItem：每一步在干嘛 -
PlanningState：计划整体执行状态
-
TodoManager 如何与 Main Loop 交互
- 把 TodoManager 也作为一个 Tool 加入 Tools Map
- 此后，Main Loop 将在
messages列表的基础上额外对PlanningState进行维护
class TodoManager:
def __init__(self):
self.state = PlanningState()
def render(self) -> str:
"""把 Todo 以更可读的形式输出"""
lines = []
for item in self.state.items:
marker = {
"pending": "[ ]",
"in_progress": "[>]",
"completed": "[x]",
}[item.status]
line = f"{marker} {item.content}"
if item.status == "in_progress" and item.active_form:
line += f" ({item.active_form})"
lines.append(line)
completed = sum(1 for item in self.state.items if item.status == "completed")
lines.append(f"\n({completed}/{len(self.state.items)} completed)")
return "\n".join(lines)
def update(self, items: list) -> str:
"""允许更新 Todo"""
validated = []
in_progress_count = 0
for item in items:
status = item.get("status", "pending")
if status == "in_progress":
in_progress_count += 1
validated.append({
"content": item["content"],
"status": status,
"activeForm": item.get("activeForm", ""),
})
if in_progress_count > 1: # 约束 check
raise ValueError("Only one plan item can be in_progress")
self.state.items = validated
self.state.rounds_since_update = 0
return self.render()
# 连续 PLAN_REMINDER_INTERVAL 轮没有更新计划，进行提醒
def note_round_without_update(self) -> None:
self.state.rounds_since_update += 1
def reminder(self) -> str | None:
if not self.state.items:
return None
if self.state.rounds_since_update < PLAN_REMINDER_INTERVAL:
return None
return "<reminder>Refresh your current plan before continuing.</reminder>"
1.4 SubAgent
看起来还是 顺序执行，只是把 dirty work 外包了
把探索性任务丢给 SubAgent，让父 Agent 在干净的 Context 中盯紧主目标
在组会上汇报你的实验结果就可以了，不要和我讲你的失败小故事
-
隔离
-
上下文隔离：SubAgent 有自己的独立
messages（基础实现为 “白手起家”）拓展：fork => 继承 Parent 的已有 msgs（Ctx）、并追加子任务 Prompt
-
工具隔离：只获取必要工具、不允许继续派生 SubSubAgent
-
-
保护措施：防止无限循环
- 最大允许运行轮次
- 工具调用出错时如何退出
-
SubAgent 只把 最终摘要/结果 写回父智能体，中间不会改动 Parent Msgs
-
task tool：让 Parent 能够主动外包任务TOOLS += { "name": "task", "description": "Run a subtask in a clean context and return a summary.", "input_schema": { "type": "object", "properties": { "prompt": {"type": "string"} }, "required": ["prompt"] } }对应的 handler 如下，此处为 “白手起家” 模式
def run_subagent(prompt: str) -> str: # 白手起家 sub_messages = [{"role": "user", "content": prompt}] # 最大轮次限制：只操作 SubAgent msgs for _ in range(30): response = client.messages.create( model=MODEL, system=SUBAGENT_SYSTEM, messages=sub_messages, tools=CHILD_TOOLS, max_tokens=8000, # 工具隔离 ) sub_messages.append({"role": "assistant", "content": response.content}) if response.stop_reason != "tool_use": break results = [] for block in response.content: if block.type == "tool_use": handler = TOOL_HANDLERS.get(block.name) output = handler(**block.input) if handler else f"Unknown tool: {block.name}" results.append({"type": "tool_result", "tool_use_id": block.id, "content": str(output)[:50000]}) sub_messages.append({"role": "user", "content": results}) # 只返回摘要 return "".join(b.text for b in response.content if hasattr(b, "text")) or "(no summary)"
1.5 Skills
Skill 是一种支持按需加载的领域知识（system prompt），解决 '怎么做' 的问题
Skill - Memory - CLAUDE.md
- Skill: 可选的知识包，仅在完成特定任务时 按需加载
- Memory：系统记住的 跨对话 信息（通常是一些 fact / preference），不是任务 SOP
- CLAUDE.md：全局、稳定、长期的规则说明
名词解释
-
skill：用于解决特定任务的说明书（可复用），通常包含以下内容- 什么时候用
- 解决任务的具体步骤
- 有哪些注意事项
-
actions
-
discovery：列出可用的 skills、以及这些 skill 能干什么（ name + description ） -
loading：按需注入 —— 真正将完整的 skill 内容放进当前上下文
-
实现
-
如何存放
skills：一般来说会按照以下目录来塞 md 文档 -
代码实现：
# 被塞进 sysPrompt 只有格式化为 “- name: desc” 的 manifest 列表 SKILL_REGISTRY = SkillRegistry(SKILLS_DIR) SYSTEM = f"""You are a coding agent at {WORKDIR}. Use load_skill when a task needs specialized instructions before you act. Skills available: {SKILL_REGISTRY.describe_available()} """ # 提供用于加载特定技能的 tool：对应的 result 为 skill 的完整正文 TOOL_HANDLERS = { "load_skill": lambda **kw: SKILL_REGISTRY.load_full_text(kw["name"]), } # 从 `skills` 路径下整理所有可用技能 class SkillRegistry: def __init__(self, skills_dir): self.skills = {} self._load_all() def _load_all(self): for path in skills_dir.rglob("SKILL.md"): meta, body = parse_frontmatter(path.read_text()) name = meta.get("name", path.parent.name) self.skills[name] = { "manifest": { "name": name, "description": meta.get("description", ""), }, "body": body, } def load_full_text(self, name: str) -> str: skill = self.skills.get(name) if not skill: known = ", ".join(sorted(self.skills)) or "(none)" return f"Error: Unknown skill '{name}'. Available skills: {known}" return ( f"<skill name=\"{skill['manifest']name}\">\n" f"{skill['body']}\n" "</skill>" )
1.6 Ctx 压缩
Ctx 不是越多越好，只在活跃区域保留有价值的部分
tool output
|
+-- 太大 -----------------------> 保存到磁盘 + 留预览
|
v
messages
|
+-- 太旧 -----------------------> 替换成简短的占位提示
|
+-- 整体还是太长 / 手动 compact --> 把历史信息压缩成摘要
|
v
neo_history
-
需要维护的压缩状态 Meta：
-
长
tool_resultDump 工具：返回存储路径 + 开头的 previewdef persist_large_output(tool_use_id: str, output: str) -> str: if len(output) <= PERSIST_THRESHOLD: return output stored_path = save_to_disk(tool_use_id, output) preview = output[:2000] return ( "<persisted-output>\n" f"Full output saved to: {stored_path}\n" f"Preview:\n{preview}\n" "</persisted-output>" ) -
把太旧的
tool_result替换成占位符：此处仅保留最近 3 次工具调用结果 -
整体历史压缩工具：需要保留 当前目标、已完成/未完成的事项、被改动的文件、需要保留的决定
# 具体的压缩函数 def summarize_history(messages: list) -> str: conversation = json.dumps(messages, default=str)[:80000] prompt = ( "Summarize this coding-agent conversation so work can continue.\n" "Preserve:\n" "1. The current goal\n" "2. Important findings and decisions\n" "3. Files read or changed\n" "4. Remaining work\n" "5. User constraints and preferences\n" "Be compact but concrete.\n\n" f"{conversation}" ) response = client.messages.create( model=MODEL, messages=[{"role": "user", "content": prompt}], max_tokens=2000, ) return response.content[0].text.strip() # 保留 focus 目标 + 最近改动的 files def compact_history(messages, state, focus): # 先保存以下压缩前的内容（万一之后还要找回） transcript_path = write_transcript(messages) print(f"[transcript saved: {transcript_path}]") summary = summarize_history(messages) if focus: summary += f"\n\nFocus to preserve next: {focus}" if state.recent_files: recent_lines = "\n".join(f"- {path}" for path in state.recent_files) summary += f"\n\nRecent files to reopen if needed:\n{recent_lines}" state.has_compacted = True state.last_summary = summary return [{ "role": "user", "content": ( "This conversation was compacted so the agent can continue working.\n\n" f"{summary}" ), }] -
接入 Main Loop：插在 LLM Call 之前
2 系统安全性
2.1 鉴权
不是所有 '意图' 都应该被执行
一个最小的权限管理系统遵循以下风格（顺序检查即可）：
-
Deny Rules：命中了就直接拒绝（提前拒绝看起来就不对劲的命令）
-
Mode Check：当前系统风格（决定特定分组下的 tool 是否执行）
模式 含义 适用场景 default未命中规则时问用户 日常交互 plan只允许读，不允许写 计划、审查、分析 auto简单安全操作自动过，危险操作再问 高流畅度探索 -
Allow Rules：命中了就直接通过（跳过 常见/重复 操作的检查）
-
Ask User：剩下不确定的都问 User
-
Rule 结构定义
-
MVP 实现
def check_permission(tool_name: str, tool_input: dict) -> dict: # 1. deny rules for rule in deny_rules: if matches(rule, tool_name, tool_input): return {"behavior": "deny", "reason": "matched deny rule"} # 2. mode if mode == "plan" and tool_name in WRITE_TOOLS: return {"behavior": "deny", "reason": "plan mode blocks writes"} if mode == "auto" and tool_name in READ_ONLY_TOOLS: return {"behavior": "allow", "reason": "auto mode allows reads"} # 3. allow rules for rule in allow_rules: if matches(rule, tool_name, tool_input): return {"behavior": "allow", "reason": "matched allow rule"} # 4. fallback to User return {"behavior": "ask", "reason": "needs confirmation"} -
接入 Tool-using Loop：在真正调用前
最好再实现一下 '连续拒绝' 计数 => 你的 Agent 可能卡住了
你可以：给出 reminder / 切换到 plan 模式 / 让用户澄清目标
2.2 Hooks
在不修改 Main Loop 的前提下，在固定时期插入自定义附加行为
-
（简化版）Hook 返回值协议：不要今天返回字符串，明天返回布尔值，后天返回整数
retValue 含义 0正常，继续运行 1阻止当前行为 2注入补充消息后，继续运行 -
Handler Map：为 event 指定对应的 handler 函数（可以是一组 / 为空）
-
统一运行（一组）hooks：此处 阻止/注入 返回将阻断运行、并直接返回
def run_hooks(event_name: str, payload: dict) -> dict: for handler in HOOKS.get(event_name, []): result = handler(payload) if result["exit_code"] in (1, 2): return result return {"exit_code": 0, "message": ""}其中，
payload可以设计为如下格式： -
接入 tool-using Loop：
results = [] for block in response.content: # -- PreToolUse -- pre = hooks.run_hooks("PreToolUse", payload) # Inject hook messages into results if pre["exit_code"] == 1: results.append(blocked_tool_result(pre["message"])) continue if pre["exit_code"] == 2: messages.append({"role": "user", "content": pre["message"]}) # 真正执行工具 handler = TOOL_HANDLERS.get(block.name) # -- PostToolUse -- post = hooks.run_hooks("PostToolUse", payload) # Inject post-hook messages（这里不可能被 block 了） if post["exit_code"] == 2: messages.append({"role": "user", "content": pre["message"]}) # 正常塞回 tool_result results.append({ "type": "tool_result", "tool_use_id": block.id, "content": str(output), })
2.3 Memory 管理
维护跨对话、无法直接从当前仓库状态推导的信息，在 新会话开启时自动加载
memory 只用来提供方向，不能替代当前观察
- memory 记录的是“曾经成立过的事实”，不是永久真理
- 当 code 与 memo 冲突时，请相信 code
- 根据 memo 推荐资源前，再 check 一遍（万一它改了呢）
如果用户让我 '忽略之前的记忆' 呢 => 在 当前轮次 中，按照 memory 为空进行工作
-
一些值得记录的 Memory
type desc sample user用户偏好 代码风格 / 回复风格 / 偏好工具链 feedback用户的人工纠正 / 正反馈 xxx 之前错过 / 以后遇到这种情况要先 xxx / xxx 判断方式是有效的 project不容易直接从 code 看出来的约定 / 背景 特定目录不能改 / 某个决定是因为合规要求 reference外部资源指针 资料库对应的 URL / 特定监控面板的访问方式 -
存储结构
-
接入：在会话开始时，读取所有 memory 拼接为 section、塞进 system prompt
2.4 System Prompt
你面对的是可是超级拼装口牙
System Prompt =
# 静态部分
core # 核心身份和行为说明
+ tools # 工具列表
+ skills # skill 元信息
+ memory
+ claude_md
# 动态部分
+ dynamic_context # 动态环境信息（date / pwd / model / mode）
在实现上也不过是顺序拼装（每个 builder 只负责对应模块的加载）：
class SystemPromptBuilder:
def build(self) -> str:
parts = []
parts.append(self._build_core())
...
parts.append(self._build_dynamic())
return "\n\n".join(p for p in parts if p)
2.5 错误恢复
错误不是例外，而是必须预留的正常分支
先对错误进行分类，再选择恢复路径（有限次补救），失败最后才暴露给用户
-
典型错误及其恢复方式
stop reason solution 输出被截断 注入续写提示 + retry Ctx 过长 压缩上下文 + retry 临时连接失败 等一会儿 + retry -
数据结构
-
（Meta）Recovery State：（分别）统计每种恢复路径重试次数，避免死循环
-
恢复决策：
-
恢复选择器：类似于 Mapper
def choose_recovery(stop_reason: str | None, error_text: str | None) -> dict: if stop_reason == "max_tokens": return {"kind": "continue", "reason": "output truncated"} if error_text and "prompt" in error_text and "long" in error_text: return {"kind": "compact", "reason": "context too large"} if error_text and any(word in error_text for word in [ "timeout", "rate", "unavailable", "connection" ]): return {"kind": "backoff", "reason": "transient transport failure"} return {"kind": "fail", "reason": "unknown or non-recoverable error"}
-
-
接入 Main Loop
- 在 LLM Call 外层处理：API + 网络错误
- 在拿到 Response 后处理：超出 max_token
while True: response = None for attempt in range(MAX_RECOVERY_ATTEMPTS + 1): try: # LLM 请求 break # success except APIError as e: error_body = str(e).lower() # 压缩 Ctx if "overlong_prompt" in error_body or ("prompt" in error_body and "long" in error_body): messages[:] = auto_compact(messages) continue # 请求错误 if attempt < MAX_RECOVERY_ATTEMPTS: delay = backoff_delay(attempt) time.sleep(delay) continue return # 超出最大重试 except (ConnectionError, TimeoutError, OSError) as e: # 连接错误 if attempt < MAX_RECOVERY_ATTEMPTS: delay = backoff_delay(attempt) time.sleep(delay) continue return # 超出最大重试 if response is None: return messages.append({"role": "assistant", "content": response.content}) # 输出被截断 if response.stop_reason == "max_tokens": max_output_recovery_count += 1 if max_output_recovery_count <= MAX_RECOVERY_ATTEMPTS: messages.append({"role": "user", "content": CONTINUATION_MESSAGE}) continue else: return # 重置对应 counter max_output_recovery_count = 0 # 正常工具处理
3 Task Runtime
3.1 Task System
Todo 只能提供 '有序列表' 形式的任务拆分，无法提供任务依赖拓扑
- Todo 对于 简单的、可被顺序拆分的 目标已经足够
- Task 是一个可以被跟踪、被分配、被完成、被阻塞的工作目标（不是某个正在执行的线程）
- 让 Task System 帮你把 Todos 升级成持久化的 Task Graph
-
任务状态
- pending => in_progress => completed
- deleted：逻辑删除
- 仅当
task["status"] == "pending" && not task["blockedBy"]时，可被执行（Ready）
-
任务持久化：不止是塞在
messages中一个任务存一个文件 =>
tasks/task_idx.json -
依赖关系维护
-
添加依赖关系：同时维护 A => B + B => A
-
解锁依赖关系：从所有直接后继任务的 blockedBy 中移除
-
-
接入主循环：提供以下工具
tool_name desc tool_name desc task_create新建任务 task_update更新任务状态 task_get查看单一任务 task_list查看完整任务板
3.2 后台任务
把慢命令移到后台，至少让主循环看上去没在傻等
主循环仍然只有一条，并行的是等待，不是主循环本身
Main Loop 已在后台
| |
+-- background_run("pytest") ---> 执行 pytest
| -> 立刻返回 task_id -> 完成后，写 notification
|
+-- 继续别的工作
|
v
下一轮 LLM call 前：查看所有 notifications、把摘要注入 messages
-
序列化
-
runtime-tasks/tid.json：运行状态（只有 prview 会被写回 msgs） -
runtime-tasks/tid.log：完整输出
-
-
Notification：只是告诉主循环有结果了（不是完整的日志）
-
接入主循环：提供两个 Tool
tool_name desc background_run在后台执行 runtime task background_check检查特定 runtime task / 罗列所有 runtime task 状态
class BackgroundManager:
def __init__(self):
self.tasks = {} # 所有的后台任务
self._notification_queue = [] # 已经有返回结果的任务
self._lock = threading.Lock() # 并发控制
def run(self, command: str) -> str:
"""启动后台执行线程"""
task_id = str(uuid.uuid4())[:8]
self.tasks[task_id] = {
"id": task_id,
"status": "running",
"result": None,
"command": command,
"started_at": time.time(),
"finished_at": None,
"result_preview": "",
"output_file": str(self._output_path(task_id)),
}
thread = threading.Thread(
target=self._execute, args=(task_id, command), daemon=True
)
thread.start()
return task_id
def _execute(self, task_id: str, command: str):
"""运行，完成后写 notification（自行负责）"""
try:
r = subprocess.run(
command, shell=True, cwd=WORKDIR,
capture_output=True, text=True, timeout=300
)
output = (r.stdout + r.stderr).strip()[:50000]
status = "completed"
except subprocess.TimeoutExpired:
output = "Error: Timeout (300s)"
status = "timeout"
# 记录完整输出
output_path = self._output_path(task_id)
output_path.write_text(output)
# 更新状态 + 记录 preview
self.tasks[task_id]["status"] = status
self.tasks[task_id]["result"] = final_output
self.tasks[task_id]["finished_at"] = time.time()
self.tasks[task_id]["result_preview"] = self._preview(output)
with self._lock:
self._notification_queue.append({
"task_id": task_id,
"status": status,
"command": command[:80],
"preview": preview,
"output_file": str(output_path.relative_to(WORKDIR)),
})
def drain_notifications(self) -> list:
"""下一次 LLM Call 前，处理所有 notifications"""
with self._lock:
notifs = list(self._notification_queue)
self._notification_queue.clear()
return notifs
3.3 定时任务
等待开始：在未来的特定时间做某事
- schedule_create => 把定时调度记录写到列表或文件里
- 后台定时检查器每分钟 check 一次“现在是否匹配”
-> 匹配，把 prompt 放进 nofitication
-> 下一轮 当成 User Msg 喂给 LLM
-
从创建一条新的调度记录开始：durable 需要 dump，session only 可以不管
-
定时检查 Loop：需要开一条线程持续跑，到点就写 notification
def check_loop(self): while True: now = datetime.now() self.check_jobs(now) time.sleep(60) def check_jobs(self, now): for job in self.jobs: if cron_matches(job["cron"], now): self.queue.put({ "type": "scheduled_prompt", "schedule_id": job["id"], "prompt": job["prompt"], }) job["last_fired_at"] = now.timestamp() -
接入主循环：最终还是让 LLM 来判断定时任务怎么执行
4 Multi Agents
4.1 Agents Team
SubAgent 用完一次就丢，Team 支持长期在线接活 + 协作
-
团队名册：队伍里有谁、分别担任什么角色、目前状态（空闲 / 工作中 / 下线）
-
teammate-
一个拥有独立名字、角色、消息入口和生命周期的持久 agent
-
新锅收取：邮箱（在下一轮工作前整理信息）
最简单的实现是一人一个 jsonl 文件（
.team/inbox/NAME.jsonl）、来锅直接 append -
上下文压缩后，队友有时会忘记自己是谁。最小补法是重新注入一段身份提示：
-
目前的实现仍由 leader 手动进行（创建队友、分派任务）
是的，我们还没实现自主找活、自动恢复工作的全自治牛马
class TeammateManager:
# 1 老大说，首先要有一份名册
def __init__(self, team_dir: Path):
self.dir = team_dir
self.config_path = self.dir / "config.json"
self.config = self._load_config()
def _load_config(self) -> dict:
if self.config_path.exists():
return json.loads(self.config_path.read_text())
return {"team_name": "default", "members": []}
# 2 捏一个新的牛马来干活吧
def spawn(self, name: str, role: str, prompt: str) -> str:
member = {"name": name, "role": role, "status": "working"}
self.config["members"].append(member)
self._save_config()
thread = threading.Thread(
target=self._teammate_loop, # 略（和 main loop 差不多、但需要 check 邮箱）
args=(name, role, prompt),
daemon=True,
)
thread.start()
# 3 谁家牛马天天看邮箱找活干
def _teammate_loop(self, name: str, role: str, prompt: str):
messages = [{"role": "user", "content": prompt}]
while True:
inbox = bus.read_inbox(name) # 读全部信息 => 解析为列表 => 清空邮箱
for item in inbox:
messages.append({"role": "user", "content": json.dumps(item)})
response = client.messages.create(...)
...
4.2 团队协议
Protocol 只负责协作
协议请求不是任务本身，请求状态表也不是任务板，任务系统才负责 “工作推进”
-
协议消息
-
请求追踪：最好还是一个个 dump 为
.team/requests/req_xxx.json -
运行协议
4.3 自主代理
嗨嗨嗨，两眼一睁就是自己找活干的全自动牛马来辣！！！
当可汗点不动兵成为瓶颈，牛就要学会自己认领磨拉
私戳和公屏都要看捏
活跃牛马
|
| 当前轮工作做完，或者主动进入 idle
v
绝赞摸鱼
|
+-- 看邮箱，有新消息 -> 回到 WORK
|
+-- 看任务板，有 ready task -> 认领 -> 回到 WORK
|
+-- 长时间什么都没有 -> shutdown
-
如何判断一个锅能不能被自主认领
-
认领后需要修改
-
任务状态
-
需要落盘的自主拿锅记录
-
-
接入 teammate_loop
# 支持 work -> idle 状态流转 while True: run_work_phase(...) should_resume = run_idle_phase(...) if not should_resume: break # idle：从 邮箱 + 告示板 找活干，循环若干轮后关闭自己 def idle_phase(name: str, messages: list) -> bool: # 邮箱被点名，有活干了 inbox = bus.read_inbox(name) if inbox: messages.append({ "role": "user", "content": json.dumps(inbox), }) return True # 没人找我？看看公告栏 unclaimed = scan_unclaimed_tasks(role) # 支持接收角色过滤 if unclaimed: task = unclaimed[0] claim_result = claim_task( # clain_task 通过锁实现原子操作 task["id"], name, role=role, source="auto", ) # 默念自己的角色再开始干活 ensure_identity_context(messages, name, role, team_name) messages.append({ "role": "user", "content": f"<auto-claimed>Task #{task['id']}: {task['subject']}</auto-claimed>", }) messages.append({ "role": "assistant", "content": f"{claim_result}. Working on it.", }) return True
4.4 💀 Worktree
让两个 teammate 在同一个 code 版本上独立修改、互不干扰（不共享未 commit 的改动）
-
worktree 注册表：你这个任务到底在哪条车道上创人啊
-
TaskRecord：新增与 worktree 相关字段
-
WorkTreeRecord：路径映射 + 最近操作 + 收尾操作
-
CloseoutRecord：收尾行为 => 保留 / 回收
-
EventRecord：worktree 具有复杂的生命周期
创建 => 进入 => 执行命令 => 保留 / 删除 => 删除失败 ...
-
接入思路：先开任务、再指定 worktree
- 创建任务
- 分配 worktree、写入注册表
- 更新 TaskRecord 关于 worktree 的字段
- 显示切换工作目录、在对应目录下执行命令
- 收尾：显式通过
worktree_closeout处理，同意兜住keep / remove
删除 worktree 前应该检查 是否有脏改动
4.5 MCP
不是所有活都需要本机干，搞一些可插拔的外包员工进来吧
启动时
-> PluginLoader 找到 server 配置
-> MCP client 连接 server
-> list_tools 并标准化 tool_name
-> 与 native tools 一起合并进同一个工具池
运行时
-> LLM 产出 tool_use
-> 统一权限处理
-> 工具路由：native / mcp
-> 标准化 tool_result
-> 回到主循环
-
MCP：让 agent 和外部工具程序对话的统一协议
-
MCP Server 会首先告诉你 "我能提供什么工具"(还能提供一些 resource)
-
当你想要借用工具时、把请求转发给它，它会向你返回执行结果
-
-
Plugin 配置：有哪些可用的外部工具、通过什么命令调用
-
MCP Client：支持连接 MCP Server、列出可用工具、调用特定工具