CoreCoder源码精读系列：逐文件拆解agent.py、llm.py、context.py，看懂生产级Agent的每个设计决策

CoreCoder 是一个极简单文件视角的 AI coding agent 开源项目：用约 1,322 行引擎代码，把生产级编码 Agent 的核心机制诚实地写了出来。本篇源码精读逐文件拆解引擎的三个心脏文件agent.py、llm.py、context.py，讲清楚每一处设计决策背后的"为什么"：主循环如何兜底、LLM 调用如何重试与记账、有限上下文窗口里如何扛住长任务。读完你能看懂一个 coding agent 的完整骨架。

三个文件的分工：agent 主循环全景图

在钻进代码之前，先建立一张地图。整个引擎就是三块拼图：

数据流一句话：用户消息进agent.py的循环 →llm.py把消息流式发给模型 → 模型要工具就执行、结果回填 → 历史太长时context.py分层压缩 → 直到模型不再要工具、吐出答案为止。

想边读边断点，五分钟先跑起来：

git clone https://gitcode.com/gh_mirrors/co/CoreCoder cd CoreCoder pip install -e . # 任意 OpenAI 兼容模型都能接，只需两个环境变量 export OPENAI_API_KEY=sk-... OPENAI_BASE_URL=https://api.deepseek.com CORECODER_MODEL=deepseek-chat corecoder

agent.py（240行）：主循环只有二十行，其余全是护栏

主循环骨架：问模型、跑工具、再问

打开 agent.py，核心就是chat()方法（完整实现 L70-L124），骨架长这样：

def chat(self, user_input): self.messages.append({"role": "user", "content": user_input}) self.context.maybe_compress(self.messages, self.llm) for _ in range(self.max_rounds): # 默认 50 轮，硬上限 resp = self.llm.chat(self._full_messages(), self._tool_schemas()) if not resp.tool_calls: # 模型不再要工具 self.messages.append(resp.message) return resp.content # -> 任务完成，交还答案 self.messages.append(resp.message) results = run_tools(resp.tool_calls) # 多个则并行执行 self.messages += tool_replies(results) # 结果回填，进入下一轮 return "(reached maximum tool-call rounds)"

值得停一下的细节：模型自己决定何时收手。没有任何外部规则判断"任务完成了没有"，它读完文件、跑完测试、觉得够了，就只回一段文本。这种"让模型自己判断收敛"是所有 agent 的共同假设——你不在写 if-else，你在和一个会自己拿主意的东西协作。

刹车设计：为什么用for而不是while(true)

循环写成for _ in range(self.max_rounds)，默认 50 轮。这不是风格偏好，是最便宜的保险：模型可能陷入"读文件→发现不对→再读→还是不对"的死循环，没有上限就会一直烧 token。真撞上限，循环克制地返回一句(reached maximum tool-call rounds)，把控制权交还给你。

系统提示词每轮重新拼装

_full_messages()不是直接返回缓存的 system prompt，而是每一轮都重新拼：plan 模式的追加指令、agent 自己维护的 TODO 任务列表，都在每次请求前现渲染。好处是两点——会话中途切换/plan，下一个请求立即生效；模型看到的任务清单永远是当前状态，而不是一份埋在旧工具输出里的过期副本。

两段式 try：区分"参数填错"与"工具自己炸了"

_exec_tool里藏着一个很值得偷的工程判断：

try: inspect.signature(tool.execute).bind(**tc.arguments) # 先只做参数绑定 except TypeError as e: return f"Error: bad arguments for {tc.name}: {e}" try: return tool.execute(**tc.arguments) except Exception as e: return f"Error executing {tc.name}: {e}"

如果直接调用，TypeError分不清是模型给的参数对不上签名（它的错，该让它改）还是工具内部 bug（和参数无关）。bind()不执行函数，只校验实参能否绑定形参，先把"参数问题"单独拎出来判。模型拿到一句精确的错误反馈，下一轮就能自我修正；拿到误导性的反馈，会朝错误方向越改越远。

Ctrl+C 的半截状态：给中断补上占位回复

OpenAI 兼容协议有条硬约束：assistant 消息里每个tool_calls必须有tool_call_id一一配对的tool回复，否则下一次请求直接被 API 拒绝。而用户随时可能 Ctrl+C，正好打在"模型返回了一批工具调用、工具还没跑完"的瞬间。

_answer_pending_tool_calls的处理：给每个还没拿到回复的调用补一条[interrupted]占位，让历史重新合法，再抛出异常。这样中断之后还能接着聊，会话不被"弄脏"。这是 demo 和可交付 agent 的距离之一——循环不仅要处理"正常走完"，还要处理"在任意一步被掐断"。

并行执行：线程池干活，主线程"问路"

模型一次返回多个工具调用时，_exec_tools_parallel用 8 线程的线程池并发执行。两个容易忽略的设计：

• hooks 和权限确认全部在主线程提前结算。如果放到池子里，多个 worker 会在同一个终端上交错弹出一串"允许吗？"，体验灾难。
• 被追踪的 cwd 是线程局部的：worker 拿不到会话当前目录，所以要显式传入；某个 worker 里cd之后，主线程按调用顺序把目录变化合并回来，让一批并行命令的行为等价于顺序执行。

🛡️ Plan 模式：一个布尔量压过--yes

_permit里，plan 模式优先于一切同意层——连脚本用的--yes都压不过。开启期间所有写操作当场被拒，拒绝信息作为普通工具结果喂回模型，告诉它"只用只读工具继续调研，然后给出编号计划"。用户输入approve后才放行执行。机械上就是Agent上的一个布尔量加一个拒绝分支，主循环和权限层都不用动：

不想配 API key 也能体验这套流程，仓库自带离线演示 examples/plan_hooks_demo.py。

llm.py（349行）：引擎最大的文件，全在做脏活

调用模型本身不难，难的是流式响应会把每个工具调用的参数撕成碎片、provider 会给你半个 JSON 或 null 的 usage 字段、429 和超时要退避重试而 4xx 该直接抛。llm.py整篇就是在处理这些。

一个 OpenAI 兼容层接住全世界

DeepSeek、Qwen、Kimi、GLM、本地 Ollama 都暴露 OpenAI 兼容端点，所以LLM直接复用 openai SDK，换 provider 只是换base_url和 key 两个环境变量。对不提供兼容端点的（AWS Bedrock、Google Vertex 等），LiteLLM子类走统一接口路由到 100+ 家，设CORECODER_PROVIDER=litellm即可。

错误三分法：重试、适配、抛出

两层重试结构把错误分成三种命运：

其中"方言适配"是容易被忽略的巧思：新版 OpenAI 模型只认max_completion_tokens且不接受自定义 temperature，_adapt_rejected_param解析 400 报错里被引号点名的参数，自动改名或丢弃；还有一层兜底是某些服务器会拒掉stream_options扩展字段，那就丢一次再发。注意引号匹配是关键——max_completion_tokens的报错文本里也含max_tokens，不加引号限定就会误触发转换。

另一个设计：流式中途中断时，重试的是整个请求而非连接。因为此时工具还没执行，整包重发是安全的，已经流式显示出的半截文本会被重新生成，不会产生副作用。

碎片化工具调用的拼接

流式协议下，一个tool_call的 id、函数名、参数 JSON 会分散在多个 chunk 里。_drain用一个tc_map按index归位：

for tc_delta in delta.tool_calls: idx = tc_delta.index if idx not in tc_map: tc_map[idx] = {"id": "", "name": "", "args": ""} if tc_delta.id: tc_map[idx]["id"] = tc_delta.id if tc_delta.function: if tc_delta.function.name: tc_map[idx]["name"] = tc_delta.function.name if tc_delta.function.arguments: tc_map[idx]["args"] += tc_delta.function.arguments # 逐片累加

整条流走完后才json.loads完整参数串，解析失败兜底为空 dict 而不是崩溃。思考模型（deepseek-reasoner、kimi 系）的reasoning_content也在这里单独接住：只用于展示，永不进对话历史，因为很多 provider 把它发回去会直接拒绝请求。

💰 诚实记账：可覆盖的价格表

_PRICING内建了 GPT、Claude、DeepSeek、Qwen、Kimi 每百万 token 的 (输入, 输出) 价格，_load_pricing允许用~/.corecoder/pricing.json覆盖任一条目——新模型上市或调价，不用等发版。/tokens命令背后就是这张表加上累计 token 数。

context.py（220行）：用有限窗口扛住长任务

上下文窗口是 agent 的硬约束。核心问题不是"满了怎么办"，而是"满之前，先让掉什么"。

先算对账：token 估算要懂中文

[ _approx_tokens](https://link.gitcode.com/i/07f5e66ce423fb4cfb1541fdf8e4f1e5#L28-L35)没有拍脑袋按"每 3 字符 1 token"，而是分类估算：CJK 字符约 1.5 字符/token（汉字信息密度高），符号密集的代码约 2.8，普通英文散文约 3.4。按固定 3 字符估算，中文会话的真实体积会被读成一半，压缩触发就会严重偏晚——等发现满了，窗口已经爆了。

50% / 70% / 90% 三层递进：从最便宜的让起

阈值定义在 L51-L54，maybe_compress按层触发：

设计哲学是廉价的先让：一刀切的截断往往会扔掉长任务最依赖的早期决策；分层让掉最不值钱的部分，让重要信息活得更久。第 2、3 层的总结提示词还专门要求"保留文件路径、关键决策、遇到的错误、当前任务状态；丢弃冗长命令输出和代码清单"，总结失败（比如没配 llm）时还有纯正则的兜底——从历史里抠出文件路径和 error 行拼一段摘要（_extract_key_info）。

必须后退的边界：孤儿 tool 消息

_safe_split是全文最容易被忽视、删掉一定出 bug 的函数：

split = max(0, len(messages) - keep_recent) while split > 0 and messages[split].get("role") == "tool": split -= 1

切分点如果恰好落在一批tool结果上，这些结果就会和产生它们的 assistanttool_calls消息分离——孤儿 tool 消息，OpenAI 兼容 API 见到必拒。所以边界要一路后退，直到落回tool消息之前。压缩和中断（agent.py 里的回填）都撞在同一条协议约束上，这条线守不住，整个会话立刻报错。

可以带走的设计决策清单

把这 800 行读完后，真正能搬进你自己项目的，是这十条判断：

1. 循环必须带硬上限——for range(N)比while(true)便宜且安全得多
2. 错误是普通返回值——工具炸了、被拒了，都变成文本喂回模型，循环从不被外围杀死
3. 参数错和内部错要分开报——精确的错误反馈是模型自我修正的前提
4. 半截状态必须回填——任何"每个 tool_call 要有配对回复"的系统，中断处理都是必修课
5. 确认在主线程，执行在 worker——交互类操作放进线程池必乱
6. 重试只给 5xx 和瞬时错误，4xx 直接抛——用重试次数掩盖真正的错误是最贵的 bug
7. 流式工具调用按 index 拼接——参数 JSON 会在碎片里跨 chunk 到达
8. 价格表内建 + 本地文件覆盖——记账功能不需要等发版就能跟上市场
9. token 估算按内容类型加权——中英混排场景下"每 3 字符 1 token"是失真的
10. 压缩分层且切分点必须避开 tool 消息——先让最便宜的，保住最关键的

想继续深入，仓库自带 8 篇双语源码精读系列，本篇拆解的每个细节在对应篇目里都有更完整的展开：系列总目录、主循环篇、LLM 与成本篇、上下文篇。