★ FEATURED ARTICLE 1. 十万星标背后这个项目到底做对了什么第一次看到这个项目的星标数时我的反应和大多数人一样一个命令行工具凭什么没有炫酷的界面没有铺天盖地的宣传就是一个跑在终端里的 AI Agent却能在短时间内积累到十万级别的关注度。后来我花了两周时间把它的源码从头到尾读了一遍又自己动手复刻了一个简化版本才真正理解了一件事这个项目的价值根本不在 AI 本身而在于它用一套极其克制的软件工程方法把一个天然充满不确定性的 LLM 应用做成了可测试、可回滚、可扩展的工程系统。这恰恰是当前 AI Agent 开发领域最稀缺的东西。我见过太多团队做 Agent 项目Demo 阶段惊艳四座一上生产就各种翻车——工具调用死循环、上下文爆炸、状态丢失、错误无法复现。问题的根源往往不是模型不够强而是软件工程的底子没打牢。这个十万星项目最值得学的地方就是它把不确定性当作一等公民来对待用事件溯源、分层解耦、CLI 优先等策略把 LLM 的随机性关进了一个可控的工程框架里。这篇文章适合两类人看一是正在做 AI Agent 开发、被各种工程问题折磨的从业者二是想从真实项目中理解软件工程到底怎么落地的学习者。我不会泛泛而谈什么架构设计原则而是会把这个项目里几个关键决策拆开讲清楚它为什么这么做、不这么做会怎样、你自己动手时该怎么抄作业。读完之后你应该能对如何从 0 到 1 搭建一个靠谱的 AI Agent有一套清晰的工程思路。2. 事件溯源为什么 Agent 的状态管理不能用传统 CRUD2.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(frozenTrue) class UserMessage: content: str timestamp: datetime dataclass(frozenTrue) class ToolCall: tool_name: str arguments: dict call_id: str timestamp: datetime dataclass(frozenTrue) class ToolResult: call_id: str output: str is_error: bool timestamp: datetime dataclass(frozenTrue) 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(contentuser_input)) # 2. Agent 执行循环 while True: context build_context(event_store.all_events()) response await call_llm(context) event_store.append(AssistantMessage( contentresponse.content, tool_callsresponse.tool_calls )) print(fagent {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_idcall.call_id, outputresult.output, is_errorresult.is_error )) print(f [result] {result.output[:200]})这个循环看起来简单但包含了 Agent 的核心输入 → 模型 → 工具 → 模型 → ... → 输出。所有复杂功能都是在这个骨架上叠加的。3.4 CLI 交互设计中的几个实用技巧技巧一用颜色区分信息层级。用户输入用默认色模型输出用白色工具调用用青色错误用红色。这样一眼就能看出执行到哪一步了。不要用太多颜色三到四种足够。技巧二工具调用要显示参数摘要。不要只显示正在调用工具而是显示工具名和关键参数。比如[tool] read_file(path./src/main.py)这样你能立刻判断这个调用是否合理。技巧三长输出要截断。工具返回的内容可能很长全部打印会刷屏。我的做法是默认只显示前 200 个字符加一个--verbose参数可以看完整内容。技巧四支持中断和恢复。用户按 CtrlC 时不要直接退出而是中断当前执行把已产生的事件保存下来下次可以继续。这个功能在调试长任务时特别有用。4. 工具调用的边界控制Agent 最容易失控的地方4.1 工具调用为什么会失控Agent 最危险的地方不是模型说错话而是它反复调用工具却得不到有效结果。我见过最夸张的案例是一个 Agent 在读取文件失败后连续调用了 47 次同一个工具每次都传相同的参数直到把 token 耗尽。这不是模型笨而是工程上没有设置边界。失控通常有三种模式死循环反复调用同一工具、无限递归工具 A 调用工具 BB 又调用 A、资源耗尽单次工具返回内容过大撑爆上下文。这三种问题都必须用工程手段解决不能指望模型自己收敛。4.2 三层防护机制这个项目用了三层防护来控制工具调用边界我觉得这个设计非常值得借鉴。第一层单轮调用次数限制。模型在一次响应里最多触发 N 个工具调用超过就截断。N 一般设成 5 到 10。这个限制防止模型一次性触发大量调用。第二层单会话调用总数限制。整个会话里工具调用总数不超过 M 次超过就强制结束。M 取决于任务复杂度我一般设成 50。这个限制防止死循环。第三层重复调用检测。如果连续三次调用的工具名和参数完全相同直接判定为死循环中断执行并返回错误。这个检测要基于参数的哈希值而不是字符串比较避免格式差异导致漏检。class ToolCallGuard: def __init__(self, max_per_turn10, max_per_session50): 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_k5) 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 里多了个逗号、工具名拼错了、参数类型不对、该调用工具的时候直接输出了文字。如果你的解析逻辑是假设模型一定输出正确格式那生产环境会教你做人。正确的做法是防御性解析。每一步解析都要有 fallbackJSON 解析失败就尝试提取代码块、工具名不匹配就做模糊匹配、参数类型不对就尝试转换。转换不了就返回一个明确的错误信息给模型让它重新生成。def parse_tool_call(raw: str) - ToolCall | None: # 尝试直接解析 try: data json.loads(raw) return ToolCall( tool_namedata[name], argumentsdata.get(arguments, {}), call_iddata.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_nametool_name, arguments{}, call_idgenerate_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_exceptionsTrue) # 按原始顺序排序 ordered [] for call, result in zip(calls, results): if isinstance(result, Exception): ordered.append(ToolResult( call_idcall.call_id, outputf工具执行异常{str(result)}, is_errorTrue )) 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), timeouttimeout) except asyncio.TimeoutError: return ToolResult( call_idcall.call_id, outputf工具 {call.tool_name} 执行超时{timeout}秒请检查参数或稍后重试, is_errorTrue )8. 这套工程方法能迁移到哪些场景事件溯源加 CLI 优先加边界控制这套组合拳不只适用于通用 Agent。我在几个不同场景里验证过它的可迁移性。代码助手场景工具是读文件、写文件、执行测试。事件溯源让你能精确回放Agent 为什么改了这行代码边界控制防止它反复改同一个文件。CLI 形态天然适配开发者的终端工作流。数据分析场景工具是查询数据库、执行计算、生成图表。事件溯源记录了每一步的数据变换方便审计和复现。上下文管理策略让长会话不会因为数据量太大而崩溃。运维自动化场景工具是执行命令、查看日志、重启服务。危险工具的分级确认机制在这里尤其重要避免 Agent 误操作生产环境。事件日志本身就是一份完整的操作审计记录。知识库问答场景工具是检索文档、读取片段。检索式上下文召回在这里是核心事件溯源让为什么召回了这些文档变得可追溯。每个场景的具体工具不同但底层的工程框架是一样的用事件记录一切、用 CLI 保证核心逻辑纯粹、用边界控制防止失控、用分层上下文控制成本。把这四件事做好你的 Agent 就从能跑的 Demo变成了能上生产的系统。我在实际项目里最大的体会是AI Agent 的难点从来不是 AI而是工程。模型能力是给定的你能控制的是怎么组织代码、怎么管理状态、怎么处理错误、怎么控制边界。这个十万星项目之所以值得学正是因为它把这些工程问题解决得足够干净干净到你可以直接借鉴它的思路用到自己的项目里。
阅读完成 · 觉得有帮助？
