1. 十万星标背后,这个项目到底做对了什么
第一次看到这个项目的星标数时,我的反应和大多数人一样:一个命令行工具,凭什么?没有炫酷的界面,没有铺天盖地的宣传,就是一个跑在终端里的 AI Agent,却能在短时间内积累到十万级别的关注度。后来我花了两周时间把它的源码从头到尾读了一遍,又自己动手复刻了一个简化版本,才真正理解了一件事:这个项目的价值根本不在 AI 本身,而在于它用一套极其克制的软件工程方法,把一个天然充满不确定性的 LLM 应用,做成了可测试、可回滚、可扩展的工程系统。
这恰恰是当前 AI Agent 开发领域最稀缺的东西。我见过太多团队做 Agent 项目,Demo 阶段惊艳四座,一上生产就各种翻车——工具调用死循环、上下文爆炸、状态丢失、错误无法复现。问题的根源往往不是模型不够强,而是软件工程的底子没打牢。这个十万星项目最值得学的地方,就是它把"不确定性"当作一等公民来对待,用事件溯源、分层解耦、CLI 优先等策略,把 LLM 的随机性关进了一个可控的工程框架里。
这篇文章适合两类人看:一是正在做 AI Agent 开发、被各种工程问题折磨的从业者;二是想从真实项目中理解"软件工程到底怎么落地"的学习者。我不会泛泛而谈什么"架构设计原则",而是会把这个项目里几个关键决策拆开,讲清楚它为什么这么做、不这么做会怎样、你自己动手时该怎么抄作业。读完之后,你应该能对"如何从 0 到 1 搭建一个靠谱的 AI Agent"有一套清晰的工程思路。
2. 事件溯源:为什么 Agent 的状态管理不能用传统 CRUD
2.1 传统状态管理的死穴在哪里
大部分人在做 AI Agent 时,第一反应是设计一张会话表,存 session_id、messages、status 这些字段,每次对话就 update 一下。这个思路在普通 Web 应用里没问题,但放到 Agent 场景里会立刻暴露三个致命问题。
第一个问题是不可复现。Agent 执行到第三步时调用了一个工具,返回了错误结果,然后模型基于这个错误结果做出了一个奇怪的决策。你想 debug,但数据库里只存了最终的 messages 数组,中间的工具调用参数、返回值、模型当时的完整上下文全丢了。你根本不知道它为什么走到那一步。
第二个问题是状态覆盖。Agent 的执行是流式的,模型可能在一次响应里同时触发多个工具调用,也可能在等待工具返回时被用户打断。用 update 覆盖状态,很容易出现竞态条件——后一次写入把前一次的关键信息冲掉了。
第三个问题是无法回滚。Agent 走错了一步,你想让它回到上一步重新决策,但传统 CRUD 只保留最终状态,没有历史轨迹,回滚无从谈起。
2.2 事件溯源的核心思路
这个项目采用的是**事件溯源(Event Sourcing)**模式。简单说就是:不存最终状态,只存导致状态变化的所有事件。每一次用户输入、每一次模型响应、每一次工具调用、每一次错误,都是一个不可变的事件,按时间顺序追加到事件日志里。当前状态是通过重放这些事件计算出来的。
用一个生活化的类比:传统 CRUD 像是你只保存了银行账户的当前余额,而事件溯源像是保存了所有的存取款流水。余额可以随时通过流水算出来,但流水本身包含了余额无法提供的信息——什么时候、因为什么、发生了多少钱的变化。
在代码层面,这个模式通常长这样:
from dataclasses import dataclass from datetime import datetime from typing import Union @dataclass(frozen=True) class UserMessage: content: str timestamp: datetime @dataclass(frozen=True) class ToolCall: tool_name: str arguments: dict call_id: str timestamp: datetime @dataclass(frozen=True) class ToolResult: call_id: str output: str is_error: bool timestamp: datetime @dataclass(frozen=True) class AssistantMessage: content: str tool_calls: list[ToolCall] timestamp: datetime Event = Union[UserMessage, ToolCall, ToolResult, AssistantMessage]每个事件都是 frozen 的,一旦创建就不能修改。Agent 的运行时状态是一个事件列表,任何状态查询都是对这个列表的 fold 操作。
2.3 事件溯源给 Agent 带来的三个实际好处
第一,完整的可观测性。出问题时,你可以把整个事件流 dump 出来,逐步回放,精确定位是哪一步的哪个事件导致了异常。我在复刻版本里加了一个 replay 命令,输入 session_id 就能把整个执行过程像录像一样重放出来,debug 效率提升了不止一个量级。
第二,天然支持回滚和分支。因为事件是不可变的,你可以从任意一个事件点 fork 出一个新的执行分支。比如 Agent 在第五步走错了,你可以从第四步的事件点重新开始,换一个 prompt 或者换一个模型,看看会不会走出更好的路径。这个能力在做 prompt 调优时特别有用。
第三,状态一致性有保障。事件是追加写入的,不存在 update 覆盖的问题。即使多个工具并发返回,每个结果都是独立的事件,按到达顺序追加,不会互相干扰。
注意:事件溯源不是银弹。它会让存储体积膨胀,查询当前状态需要重放事件(通常用快照来优化)。对于简单的问答机器人,用传统 CRUD 完全够用。但只要你做的是多步骤、多工具、长会话的 Agent,事件溯源带来的收益远超成本。
2.4 落地时的两个关键细节
第一个细节是事件粒度。不要把整个模型响应作为一个事件存,而是拆成 AssistantMessage 和它触发的多个 ToolCall。这样回放时能精确到每一次工具调用。粒度太粗会丢失信息,太细又会增加重放开销,我的经验是以"一次原子操作"为单位——一次用户输入、一次模型输出、一次工具执行,各算一个事件。
第二个细节是快照策略。事件日志会越来越长,每次查询都从头重放不现实。常见做法是每 N 个事件打一个快照,查询时从最近的快照开始重放。N 的取值取决于单个事件的平均大小和查询频率,我一般设成 50 到 100 之间。
3. CLI 优先:为什么终端界面反而是 Agent 的最佳载体
3.1 图形界面在 Agent 场景下的三个尴尬
很多人觉得 CLI 是"简陋"的代名词,做 Agent 就应该配一个漂亮的 Web 界面。但这个十万星项目偏偏选择了 CLI 优先,而且我认为这是它最聪明的决策之一。原因在于,图形界面在 Agent 场景下有三个绕不开的尴尬。
尴尬一:Agent 的输出是流式且不确定的。模型可能输出一段文字,然后调用工具,然后继续输出,中间还可能插入错误信息。Web 界面要处理这种混合流,需要复杂的状态管理和渲染逻辑,稍不注意就会出现"文字闪一下又消失"或者"工具结果渲染错位"的问题。而 CLI 天然就是流式的,逐行输出,天然适配。
尴尬二:Agent 需要频繁的开发者介入。调试 Agent 时,你需要看完整的 prompt、看工具调用的原始参数、看模型的原始响应。这些在 Web 界面里通常被"美化"掉了,反而增加了调试难度。CLI 里所有东西都是明文,一眼看穿。
尴尬三:Agent 的交互模式还在快速演化。今天流行这种确认机制,明天可能就换成那种。Web 界面每次改交互都要动前端,而 CLI 改起来就是改几行输出逻辑,迭代速度快得多。
3.2 CLI 优先背后的工程哲学
这个选择背后其实是一种**"先保证核心逻辑正确,再考虑交互体验"**的工程哲学。CLI 强制你把 Agent 的核心循环——接收输入、调用模型、执行工具、返回结果——做得干干净净,不掺杂任何 UI 逻辑。等核心稳定了,再在上面套 Web 界面或者 IDE 插件,都是水到渠成的事。
我在自己的项目里也验证了这一点。第一版直接上 Web,结果前端状态和 Agent 状态老是不同步,debug 花了一周。后来推倒重来,先做 CLI,核心逻辑两天就跑通了,再套 Web 界面只花了三天,而且稳定得多。
3.3 一个最小可用的 CLI Agent 循环
下面是一个简化版的 CLI Agent 主循环,展示了核心逻辑应该长什么样:
import asyncio from prompt_toolkit import PromptSession async def agent_loop(session_id: str): session = PromptSession() event_store = EventStore(session_id) while True: # 1. 读取用户输入 user_input = await session.prompt_async("you> ") if user_input.strip() in ("/exit", "/quit"): break event_store.append(UserMessage(content=user_input)) # 2. Agent 执行循环 while True: context = build_context(event_store.all_events()) response = await call_llm(context) event_store.append(AssistantMessage( content=response.content, tool_calls=response.tool_calls )) print(f"agent> {response.content}") if not response.tool_calls: break # 3. 执行工具调用 for call in response.tool_calls: print(f" [tool] {call.tool_name}({call.arguments})") result = await execute_tool(call) event_store.append(ToolResult( call_id=call.call_id, output=result.output, is_error=result.is_error )) print(f" [result] {result.output[:200]}")这个循环看起来简单,但包含了 Agent 的核心:输入 → 模型 → 工具 → 模型 → ... → 输出。所有复杂功能都是在这个骨架上叠加的。
3.4 CLI 交互设计中的几个实用技巧
技巧一:用颜色区分信息层级。用户输入用默认色,模型输出用白色,工具调用用青色,错误用红色。这样一眼就能看出执行到哪一步了。不要用太多颜色,三到四种足够。
技巧二:工具调用要显示参数摘要。不要只显示"正在调用工具",而是显示工具名和关键参数。比如[tool] read_file(path="./src/main.py"),这样你能立刻判断这个调用是否合理。
技巧三:长输出要截断。工具返回的内容可能很长,全部打印会刷屏。我的做法是默认只显示前 200 个字符,加一个--verbose参数可以看完整内容。
技巧四:支持中断和恢复。用户按 Ctrl+C 时,不要直接退出,而是中断当前执行,把已产生的事件保存下来,下次可以继续。这个功能在调试长任务时特别有用。
4. 工具调用的边界控制:Agent 最容易失控的地方
4.1 工具调用为什么会失控
Agent 最危险的地方不是模型说错话,而是它反复调用工具却得不到有效结果。我见过最夸张的案例是一个 Agent 在读取文件失败后,连续调用了 47 次同一个工具,每次都传相同的参数,直到把 token 耗尽。这不是模型笨,而是工程上没有设置边界。
失控通常有三种模式:死循环(反复调用同一工具)、无限递归(工具 A 调用工具 B,B 又调用 A)、资源耗尽(单次工具返回内容过大,撑爆上下文)。这三种问题都必须用工程手段解决,不能指望模型自己收敛。
4.2 三层防护机制
这个项目用了三层防护来控制工具调用边界,我觉得这个设计非常值得借鉴。
第一层:单轮调用次数限制。模型在一次响应里最多触发 N 个工具调用,超过就截断。N 一般设成 5 到 10。这个限制防止模型一次性触发大量调用。
第二层:单会话调用总数限制。整个会话里工具调用总数不超过 M 次,超过就强制结束。M 取决于任务复杂度,我一般设成 50。这个限制防止死循环。
第三层:重复调用检测。如果连续三次调用的工具名和参数完全相同,直接判定为死循环,中断执行并返回错误。这个检测要基于参数的哈希值,而不是字符串比较,避免格式差异导致漏检。
class ToolCallGuard: def __init__(self, max_per_turn=10, max_per_session=50): self.max_per_turn = max_per_turn self.max_per_session = max_per_session self.session_count = 0 self.recent_calls = [] def check(self, tool_name: str, arguments: dict) -> tuple[bool, str]: if self.session_count >= self.max_per_session: return False, "会话工具调用次数已达上限" call_hash = hash((tool_name, frozenset(arguments.items()))) self.recent_calls.append(call_hash) if len(self.recent_calls) >= 3: last_three = self.recent_calls[-3:] if len(set(last_three)) == 1: return False, "检测到重复调用,疑似死循环" self.session_count += 1 return True, ""4.3 工具返回内容的截断策略
工具返回内容过大是另一个常见问题。一个read_file工具读取了一个 10MB 的日志文件,直接塞进上下文,token 瞬间爆炸。解决办法是在工具层面做截断,而不是在模型层面。
具体策略是:每个工具定义自己的max_output_size,超过就截断,并在末尾加上[内容已截断,共 X 字符,显示前 Y 字符]的提示。这样模型知道内容被截断了,可以选择用更精确的参数重新调用,而不是傻傻地基于不完整信息做决策。
提示:截断阈值不要设得太小。太小会导致模型频繁重新调用,反而增加总 token 消耗。我的经验值是单次工具返回不超过 4000 个字符,大约 1000 到 1500 个 token。
4.4 工具权限的分级设计
不是所有工具都应该无条件可用。这个项目把工具分成了三个权限级别:
| 权限级别 | 典型工具 | 执行策略 |
|---|---|---|
| 只读 | read_file, list_dir, search | 自动执行,无需确认 |
| 写入 | write_file, edit_file | 首次执行需用户确认,可设置信任 |
| 危险 | execute_command, delete | 每次执行都需确认,不可信任 |
这个分级的意义在于,把用户的注意力集中在真正有风险的操作上。如果每个工具调用都要确认,用户很快就会疲劳,然后无脑点"同意",反而失去了防护意义。
5. 上下文管理:LLM 应用最烧钱也最容易做错的部分
5.1 上下文窗口不是越大越好
很多人有个误区:既然模型支持 128K 甚至 200K 的上下文,那就把所有历史都塞进去呗。这个想法在实际项目中会带来两个问题。
问题一:成本。上下文越长,每次调用的 token 消耗越大。一个 100K token 的上下文,每次调用可能就要几毛钱,一个会话几十次调用下来,成本相当可观。而且大部分历史信息对当前决策是无关的。
问题二:效果。上下文太长会导致"中间遗忘"现象——模型对开头和结尾的信息记得清楚,中间部分容易被忽略。塞得越多,关键信息反而越容易被淹没。
5.2 分层上下文策略
这个项目采用的是分层上下文策略,把上下文分成几个层次,按需加载。
第一层:系统提示词。定义 Agent 的角色、能力边界、输出格式。这部分永远保留,不参与裁剪。
第二层:当前任务上下文。与当前任务直接相关的事件,比如最近几轮对话、当前正在处理的文件内容。这部分完整保留。
第三层:历史摘要。更早的对话不保留原文,而是用模型生成一段摘要。摘要只保留关键决策和结论,丢弃过程细节。
第四层:检索式召回。当需要历史信息时,通过关键词或向量检索从事件日志里召回相关片段,而不是全部加载。
def build_context(events: list[Event], current_task: str) -> list[dict]: system_prompt = get_system_prompt() recent_events = events[-20:] # 最近 20 个事件完整保留 older_events = events[:-20] # 对更早的事件生成摘要 if older_events: summary = summarize_events(older_events) summary_message = {"role": "system", "content": f"历史摘要:{summary}"} else: summary_message = None # 检索与当前任务相关的事件 relevant = retrieve_relevant(older_events, current_task, top_k=5) context = [{"role": "system", "content": system_prompt}] if summary_message: context.append(summary_message) context.extend(format_events(relevant)) context.extend(format_events(recent_events)) return context5.3 摘要生成的时机和粒度
摘要不是每轮都生成,那样太浪费。我的做法是当事件数量超过阈值时触发摘要,比如超过 30 个事件,就把最早的 10 个事件压缩成一段摘要,替换掉原文。这样上下文长度始终维持在一个可控范围内。
摘要的粒度也很关键。太粗会丢失关键信息,太细又起不到压缩作用。我的经验是:保留决策和结论,丢弃过程和细节。比如"用户要求重构 auth 模块,Agent 读取了 auth.py,发现使用了过时的 API,决定改用新的认证方式",而不是把读取文件的完整内容都写进摘要。
5.4 一个容易被忽略的细节:工具结果的缓存
同一个工具用相同参数调用多次,结果应该是一样的(对于只读工具)。这个项目做了一个工具结果缓存,key 是工具名加参数的哈希,value 是返回结果。这样即使模型重复调用,也不会重复执行,直接返回缓存结果。
这个优化看起来小,但实际效果显著。我在一个代码分析任务里测试,加了缓存之后,工具调用次数减少了约 30%,总 token 消耗降低了 25%。因为模型经常会"忘记"自己已经读过某个文件,然后又读一遍。
6. 从 Demo 到生产:那些只有踩过才知道的坑
6.1 模型输出的解析不能太乐观
Demo 阶段,模型输出基本都符合预期格式。但一上生产,各种奇葩输出就来了:JSON 里多了个逗号、工具名拼错了、参数类型不对、该调用工具的时候直接输出了文字。如果你的解析逻辑是"假设模型一定输出正确格式",那生产环境会教你做人。
正确的做法是防御性解析。每一步解析都要有 fallback:JSON 解析失败就尝试提取代码块、工具名不匹配就做模糊匹配、参数类型不对就尝试转换。转换不了就返回一个明确的错误信息给模型,让它重新生成。
def parse_tool_call(raw: str) -> ToolCall | None: # 尝试直接解析 try: data = json.loads(raw) return ToolCall( tool_name=data["name"], arguments=data.get("arguments", {}), call_id=data.get("id", generate_id()) ) except (json.JSONDecodeError, KeyError): pass # 尝试从代码块提取 match = re.search(r'```(?:json)?\s*(\{.*?\})\s*```', raw, re.DOTALL) if match: try: data = json.loads(match.group(1)) return ToolCall(...) except json.JSONDecodeError: pass # 尝试模糊匹配工具名 for tool_name in AVAILABLE_TOOLS: if tool_name in raw: return ToolCall(tool_name=tool_name, arguments={}, call_id=generate_id()) return None6.2 错误信息要写给模型看,不是写给人看
这是一个反直觉的点。传统软件里,错误信息是给开发者看的,越详细越好。但在 Agent 里,错误信息主要是给模型看的,因为模型要根据错误信息决定下一步怎么做。
所以错误信息要满足三个条件:说清楚哪里错了、给出可能的修正方向、不要包含无关的技术细节。比如工具调用失败,不要返回一堆 Python traceback,而是返回"参数 path 指向的文件不存在,请检查路径是否正确,或先用 list_dir 查看目录内容"。
6.3 并发工具调用的顺序问题
模型可能一次触发多个工具调用,这些调用如果并发执行,返回顺序是不确定的。但事件日志要求顺序一致,否则回放时结果会不同。解决办法是给每个工具调用分配一个序号,结果按序号排序后再追加到事件日志。
async def execute_tools_parallel(calls: list[ToolCall]) -> list[ToolResult]: tasks = [execute_tool(call) for call in calls] results = await asyncio.gather(*tasks, return_exceptions=True) # 按原始顺序排序 ordered = [] for call, result in zip(calls, results): if isinstance(result, Exception): ordered.append(ToolResult( call_id=call.call_id, output=f"工具执行异常:{str(result)}", is_error=True )) else: ordered.append(result) return ordered6.4 日志和事件的区别
很多人会把日志和事件混为一谈,其实它们是两个东西。事件是业务状态的一部分,日志是运维观测的一部分。事件要持久化、要可回放、要参与状态计算;日志可以随时丢弃、不需要回放、不参与业务逻辑。
这个项目里,事件存在事件存储里,日志输出到标准错误流。两者分开,互不干扰。我见过一些项目把 debug 信息也塞进事件里,结果事件日志膨胀得飞快,回放时还要过滤掉这些噪音,非常痛苦。
6.5 测试策略:怎么测一个不确定的系统
Agent 的测试是最头疼的,因为模型输出不确定。这个项目的做法是分层测试:
- 单元测试:测工具函数、事件存储、上下文构建这些确定性逻辑,用 mock 替代模型调用。
- 集成测试:用固定的模型响应(录制好的)测整个 Agent 循环,确保流程正确。
- 评估测试:用真实模型跑一批标准任务,用规则或另一个模型来评分,关注通过率而不是单次结果。
关键是不要把不确定性引入单元测试。单元测试里模型必须是 mock 的,否则测试永远不稳定。
7. 我自己复刻时踩过的三个坑
7.1 事件存储用 JSON 文件,结果并发写入损坏
第一版我图省事,把事件直接追加到一个 JSON 文件里。单线程跑没问题,一开并发就出事了——两个工具同时返回,同时写文件,结果 JSON 格式损坏,整个会话读不出来了。后来改成每个事件一个文件,文件名用时间戳加序号,彻底解决了并发问题。虽然文件多了点,但胜在简单可靠。
7.2 上下文裁剪裁掉了系统提示词
有一次我实现上下文裁剪时,简单粗暴地"保留最近 N 条消息",结果把系统提示词也裁掉了。Agent 瞬间"失忆",不知道自己是谁、能做什么,开始胡言乱语。这个 bug 找了半天才定位到。教训是:系统提示词必须单独管理,永远不参与裁剪。
7.3 工具超时没处理,整个 Agent 卡死
有个工具是调用外部命令,正常情况下几百毫秒返回。但有一次命令卡住了,Agent 就一直等,整个会话冻结。后来给所有工具加了超时机制,默认 30 秒,超时就返回错误。这个错误信息会告诉模型"工具执行超时",模型可以选择重试或者换一种方式。
async def execute_tool_with_timeout(call: ToolCall, timeout: float = 30.0): try: return await asyncio.wait_for(execute_tool(call), timeout=timeout) except asyncio.TimeoutError: return ToolResult( call_id=call.call_id, output=f"工具 {call.tool_name} 执行超时({timeout}秒),请检查参数或稍后重试", is_error=True )8. 这套工程方法能迁移到哪些场景
事件溯源加 CLI 优先加边界控制这套组合拳,不只适用于通用 Agent。我在几个不同场景里验证过它的可迁移性。
代码助手场景:工具是读文件、写文件、执行测试。事件溯源让你能精确回放"Agent 为什么改了这行代码",边界控制防止它反复改同一个文件。CLI 形态天然适配开发者的终端工作流。
数据分析场景:工具是查询数据库、执行计算、生成图表。事件溯源记录了每一步的数据变换,方便审计和复现。上下文管理策略让长会话不会因为数据量太大而崩溃。
运维自动化场景:工具是执行命令、查看日志、重启服务。危险工具的分级确认机制在这里尤其重要,避免 Agent 误操作生产环境。事件日志本身就是一份完整的操作审计记录。
知识库问答场景:工具是检索文档、读取片段。检索式上下文召回在这里是核心,事件溯源让"为什么召回了这些文档"变得可追溯。
每个场景的具体工具不同,但底层的工程框架是一样的:用事件记录一切、用 CLI 保证核心逻辑纯粹、用边界控制防止失控、用分层上下文控制成本。把这四件事做好,你的 Agent 就从"能跑的 Demo"变成了"能上生产的系统"。
我在实际项目里最大的体会是:AI Agent 的难点从来不是 AI,而是工程。模型能力是给定的,你能控制的是怎么组织代码、怎么管理状态、怎么处理错误、怎么控制边界。这个十万星项目之所以值得学,正是因为它把这些工程问题解决得足够干净,干净到你可以直接借鉴它的思路,用到自己的项目里。