1. 为什么我要自己撸一个 Agent-Reach先说清楚这东西是干嘛的。Agent-Reach 是我最近折腾出来的一个命令行工具核心定位就一句话让 AI Agent 能真正“够得着”外部世界。你如果用过各种大模型接口肯定遇到过这种尴尬——模型本身很聪明但它被困在对话框里没法读你本地的文件、没法查数据库、没法调你写好的脚本。Agent-Reach 就是来解决这个“最后一公里”问题的。我平时的工作流里Python 脚本一大堆有些是数据清洗的有些是调内部接口的还有些是定时跑的分析任务。以前每次想让 AI 帮我串起来都得手动复制粘贴上下文效率极低。后来我试过不少现成的 Agent 框架要么太重要么绑定特定平台要么配置复杂到劝退。于是干脆自己动手用 Python 写了一个轻量的 CLI 工具把“AI 决策”和“本地执行”这两层彻底打通。这个项目适合谁呢如果你满足下面任意一条那这篇内容应该对你有用写过 Python 但没怎么碰过 AI Agent 的想给自己日常脚本加个“智能调度层”的对 CLI 工具有执念、喜欢在终端里完成一切的或者单纯想看看一个 Agent 项目从零到跑起来到底要踩哪些坑。我会把设计思路、核心代码逻辑、实操步骤、以及我调试过程中遇到的各种幺蛾子都摊开讲尽量让你看完就能复现一个自己的版本。2. Agent-Reach 的整体设计与思路拆解2.1 核心架构三层分离各司其职Agent-Reach 的架构我改过三版前两版都因为耦合太严重被我推倒重来了。现在稳定下来的结构是三层接入层、决策层、执行层。这个分层不是拍脑袋定的而是被实际需求逼出来的。接入层负责跟大模型对话把用户的自然语言指令转成结构化的“行动计划”。决策层拿到计划后判断每一步该调用哪个本地能力比如读文件、跑脚本、查数据。执行层就是真正干活的调用 Python 的 subprocess、文件 IO、或者第三方库去完成任务。三层之间通过统一的 JSON 消息格式通信这样任何一层想换实现都不会影响另外两层。为什么这么分因为我一开始把模型调用和文件操作写在一个函数里结果想换个模型供应商的时候发现要改几十处代码。分层之后换模型只需要动接入层的一个适配器文件执行层完全不用碰。这个教训很值钱Agent 项目的复杂度不在于单个功能而在于功能之间的组合爆炸分层是控制复杂度的唯一手段。2.2 为什么选 CLI 而不是 Web 界面很多人问我都什么年代了还做 CLI我的理由很实在。第一我的使用场景 90% 在终端里写代码、跑脚本、看日志都在终端切到浏览器去跟 AI 对话再切回来这个上下文切换成本太高。第二CLI 天然适合管道操作我可以把 Agent-Reach 的输出直接 pipe 给 grep、jq 或者另一个脚本组合能力是 Web 界面给不了的。第三开发成本低不用折腾前端框架和部署一个人维护完全扛得住。当然 CLI 也有代价比如交互体验不如图形界面直观复杂任务的进度展示比较粗糙。我的取舍是优先保证核心链路的稳定和可组合性交互体验够用就行。如果你要做的是面向非技术用户的产品那 Web 界面可能更合适但如果是给自己或者小团队用的效率工具CLI 的性价比高得多。2.3 技术选型Python 为主关键路径考虑过 Rust主体用 Python这个没什么好纠结的。生态成熟写起来快调试方便而且我的大部分本地脚本本来就是 Python 写的直接复用。但我在设计执行层的时候认真考虑过用 Rust 重写关键模块原因是并发场景下 Python 的 GIL 确实是个瓶颈。具体来说当 Agent 需要同时调用多个外部工具时Python 的多线程在 CPU 密集型任务上基本没优势。我实测过用 threading 跑 10 个并行的数据处理任务总耗时跟串行差不多因为 GIL 把真正的并行锁死了。后来我改用 asyncio 加多进程的组合IO 密集的用 asyncioCPU 密集的丢给 ProcessPoolExecutor。这个方案在大多数场景下够用但如果你的 Agent 要扛很高的并发比如同时处理上百个请求那确实需要考虑用 Rust 写一个高性能的执行引擎通过 FFI 或者子进程的方式跟 Python 主逻辑通信。我目前的判断是先用 Python 把逻辑跑通等真的遇到性能瓶颈再针对性优化。过早用 Rust 重写开发效率会掉一大截而且很多性能问题其实通过架构调整就能解决不一定非要换语言。3. 核心细节解析与实操要点3.1 消息协议设计Agent 各层之间的“普通话”三层之间怎么通信这个协议设计是整个项目的地基。我最终定下来的格式是一个 JSON 对象包含action、params、context三个顶层字段。action是字符串标识要做什么比如read_file、run_script、query_db。params是参数字典具体内容由 action 决定。context用来传递会话状态比如上一步的输出、用户的原始指令等。# 消息协议示例 { action: run_script, params: { script_path: /home/user/scripts/clean_data.py, args: [--input, raw.csv, --output, clean.csv] }, context: { session_id: abc123, step: 2, previous_output: 数据加载完成共 10000 行 } }这个设计的关键在于action 和 params 的分离。一开始我把所有东西都塞进一个扁平字典里结果执行层要写一大堆 if-else 来判断参数含义非常丑陋。分离之后执行层只需要一个 action 到处理函数的映射表每个处理函数只关心自己的 params 结构代码清晰了不止一个档次。注意context 字段不要塞太多东西我踩过的坑是把整个对话历史都放进去结果消息体积膨胀到几 MB序列化和传输都成了瓶颈。后来改成只传必要的状态摘要历史记录单独存到本地 SQLite 里需要的时候再查。3.2 模型接入层的适配器模式接入层我用的是适配器模式每个模型供应商对应一个适配器类统一实现parse_intent和generate_response两个方法。这样换模型的时候只需要写一个新的适配器注册到工厂里就行。class BaseAdapter: def parse_intent(self, user_input: str) - dict: 把自然语言转成结构化行动计划 raise NotImplementedError def generate_response(self, result: dict) - str: 把执行结果转成自然语言回复 raise NotImplementedError class AdapterFactory: _adapters {} classmethod def register(cls, name: str, adapter_cls): cls._adapters[name] adapter_cls classmethod def get(cls, name: str) - BaseAdapter: return cls._adapters[name]()这里有个实操要点parse_intent 的提示词工程是重中之重。我试过直接让模型输出 JSON结果它经常在 JSON 外面包一层 markdown 代码块或者加一些解释性文字导致解析失败。后来我在提示词里明确要求“只输出 JSON不要任何其他内容”并且在解析前加了一个清洗步骤用正则把代码块标记去掉。这个清洗逻辑虽然丑但非常实用建议你也加上。3.3 执行层的安全边界执行层是最危险的地方因为它真的会跑命令、删文件、改数据。我给自己定了三条铁律第一所有文件操作限制在指定的工作目录内用os.path.realpath做路径校验防止../穿越。第二所有 shell 命令必须白名单只允许执行预注册的脚本不允许模型直接生成任意命令。第三所有写操作先备份改文件之前自动复制一份到.backup目录。import os ALLOWED_SCRIPTS { clean_data: /home/user/scripts/clean_data.py, gen_report: /home/user/scripts/gen_report.py, } def run_script(script_name: str, args: list): if script_name not in ALLOWED_SCRIPTS: raise PermissionError(f脚本 {script_name} 不在白名单中) script_path ALLOWED_SCRIPTS[script_name] # 后续执行逻辑...这三条铁律帮我避免了好几次事故。有一次模型理解错了指令想执行一个删除操作因为不在白名单里直接被拦下来了。Agent 的安全不能靠模型自觉必须靠代码层面的硬约束这是我从实际踩坑中得出的血泪教训。4. 实操过程与核心环节实现4.1 环境准备与依赖安装先把环境搭起来。我假设你已经装好了 Python版本建议 3.10 以上因为用到了 match-case 语法和一些新的类型注解特性。如果还没装去 Python 官网下载安装包安装时记得勾选“Add Python to PATH”这个选项能省掉后面很多配置麻烦。依赖方面核心就几个openai或者你用的模型供应商的 SDK、rich用来做终端输出美化、pydantic做数据校验。用 pip 一条命令搞定pip install openai rich pydantic如果你在国内pip 下载慢的话可以临时指定镜像源pip install openai rich pydantic -i https://pypi.tuna.tsinghua.edu.cn/simple提示建议用虚拟环境别直接装在系统 Python 里。python -m venv agent-env然后激活这样依赖隔离不会污染其他项目。4.2 项目骨架搭建目录结构我建议这样组织清晰且易于扩展agent-reach/ ├── main.py # CLI 入口 ├── adapters/ # 模型适配器 │ ├── __init__.py │ └── openai_adapter.py ├── core/ │ ├── __init__.py │ ├── dispatcher.py # 决策层负责路由 │ └── executor.py # 执行层负责干活 ├── utils/ │ ├── __init__.py │ └── logger.py └── config.yaml # 配置文件main.py是入口用argparse或者click解析命令行参数。我用的argparse因为标准库自带不用额外装依赖。核心逻辑是接收用户输入交给 dispatcherdispatcher 调用 adapter 解析意图然后路由到 executor 执行最后把结果返回给用户。4.3 核心调度逻辑实现dispatcher 是整个项目的大脑它的职责是协调 adapter 和 executor。我实现的时候用了简单的状态机IDLE - PARSING - EXECUTING - RESPONDING - IDLE。每个状态转换都有对应的日志记录方便排查问题。class Dispatcher: def __init__(self, adapter, executor): self.adapter adapter self.executor executor self.state IDLE def handle(self, user_input: str) - str: self.state PARSING plan self.adapter.parse_intent(user_input) self.state EXECUTING results [] for step in plan.get(steps, []): result self.executor.execute(step) results.append(result) if result.get(status) error: break self.state RESPONDING response self.adapter.generate_response({results: results}) self.state IDLE return response这里有个细节值得说执行过程中如果某一步失败我是直接中断还是继续我的选择是默认中断因为 Agent 的任务通常有依赖关系前一步失败后面继续跑可能产生错误结果。但我在配置里留了continue_on_error开关某些独立任务可以开启继续执行。4.4 参数计算与性能调优并发处理是 Agent 绕不开的话题。我实测过几种方案这里给一组具体数据。测试场景是同时处理 20 个文件读取加分析任务每个任务耗时约 0.5 秒。方案总耗时CPU 利用率内存占用纯串行10.2s15%80MBthreading9.8s18%95MBasyncio1.2s25%110MBasyncio 多进程0.8s65%320MB结论很清晰IO 密集型任务用 asyncio 收益最大CPU 密集型任务必须上多进程。我最终的方案是混合模式dispatcher 根据任务类型自动选择执行策略。判断逻辑很简单如果任务涉及网络请求或文件 IO走 asyncio如果涉及大量计算丢给进程池。import asyncio from concurrent.futures import ProcessPoolExecutor async def execute_io_task(task): # IO 密集型用 asyncio await asyncio.sleep(0) # 模拟异步操作 return await do_io_work(task) def execute_cpu_task(task): # CPU 密集型用多进程 with ProcessPoolExecutor() as pool: future pool.submit(do_cpu_work, task) return future.result()注意多进程方案下任务函数必须是可序列化的不能是 lambda 或者闭包。我一开始用 lambda 传参直接报 PicklingError排查了半天才发现是这个原因。改成顶层函数或者类的静态方法就好了。5. 常见问题与排查技巧实录5.1 模型输出格式不稳定怎么办这是最高频的问题。模型有时候输出纯 JSON有时候包代码块有时候加解释文字。我的解决方案是三层防护第一层提示词里明确格式要求并给示例第二层解析前用正则清洗第三层解析失败时自动重试并加强提示。import json import re def parse_model_output(text: str) - dict: # 清洗代码块标记 text re.sub(r^(?:json)?\s*, , text.strip()) text re.sub(r\s*$, , text) try: return json.loads(text) except json.JSONDecodeError: # 尝试提取第一个 JSON 对象 match re.search(r\{.*\}, text, re.DOTALL) if match: return json.loads(match.group()) raise实测下来加了清洗逻辑之后解析成功率从 70% 左右提升到 95% 以上。剩下 5% 的失败案例重试一次基本都能解决。5.2 执行超时与资源泄漏Agent 调用外部脚本时如果脚本卡死整个流程就挂了。我给每个执行步骤加了超时控制默认 30 秒超时直接 kill 进程并返回错误。同时用try-finally确保临时文件和数据库连接一定会被清理。import subprocess def run_with_timeout(cmd: list, timeout: int 30): try: result subprocess.run( cmd, capture_outputTrue, textTrue, timeouttimeout ) return {status: ok, output: result.stdout} except subprocess.TimeoutExpired: return {status: error, message: f执行超时{timeout}秒} finally: # 清理逻辑 pass5.3 常见问题速查表问题现象可能原因排查方向解决方案模型返回空结果API 配额耗尽或网络问题检查 API 响应状态码增加重试机制检查账户余额JSON 解析失败模型输出格式不规范打印原始输出查看加强提示词增加清洗逻辑脚本执行报权限错误文件权限或路径问题检查文件权限和路径用绝对路径检查 chmod 设置并发任务互相干扰共享状态未隔离检查全局变量和临时文件每个任务用独立的工作目录内存持续增长资源未释放用 memory_profiler 分析确保连接和文件句柄及时关闭5.4 我踩过的三个大坑第一个坑是路径问题。我在开发机上用相对路径跑得好好的部署到服务器上就找不到文件了。原因是工作目录变了。后来我强制所有路径都用os.path.abspath转成绝对路径问题消失。这个坑很低级但真的很常见。第二个坑是编码问题。Windows 上默认编码是 GBKLinux 上是 UTF-8我处理中文文件的时候在两边表现不一致。解决方案是显式指定encodingutf-8所有文件读写都加上这个参数一劳永逸。第三个坑是模型幻觉。有一次模型编了一个不存在的脚本名执行层直接报错。后来我在执行层加了白名单校验不在白名单里的操作一律拒绝。永远不要相信模型输出的内容所有关键操作都要有代码层面的校验这是做 Agent 项目的铁律。6. 后续扩展方向与个人体会Agent-Reach 目前跑通了我最核心的需求但还有不少可以打磨的地方。我接下来想做的几个方向一是加一个简单的 WebSocket 接口让其他程序也能调用 Agent 的能力二是把执行层的工具注册做成插件化通过配置文件动态加载不用改代码就能加新工具三是加一个任务队列支持长时间运行的任务异步执行完成后通知。如果你也想自己撸一个类似的工具我的建议是从最小可用版本开始别一上来就追求大而全。我第一版只支持读文件和跑脚本两个功能代码不到 200 行但已经能解决我 80% 的问题。后面加的功能都是在这个基础上逐步迭代出来的。Agent 这个领域变化很快保持代码的灵活性和可替换性比追求一步到位重要得多。另外多看看别人的 Agent 项目是怎么设计的但别照搬。每个项目的使用场景不一样适合别人的架构不一定适合你。我的三层分离架构就是参考了几个开源项目之后根据自己的实际需求调整出来的。理解设计背后的“为什么”比记住“怎么做”更有价值。 
长河君 · 2026/10/29 4:06:12 · 阅读 1.2k 从零构建Agent-Reach：打通AI Agent与本地执行的CLI工具
1. 为什么我要自己撸一个 Agent-Reach先说清楚这东西是干嘛的。Agent-Reach 是我最近折腾出来的一个命令行工具核心定位就一句话让 AI Agent 能真正“够得着”外部世界。你如果用过各种大模型接口肯定遇到过这种尴尬——模型本身很聪明但它被困在对话框里没法读你本地的文件、没法查数据库、没法调你写好的脚本。Agent-Reach 就是来解决这个“最后一公里”问题的。我平时的工作流里Python 脚本一大堆有些是数据清洗的有些是调内部接口的还有些是定时跑的分析任务。以前每次想让 AI 帮我串起来都得手动复制粘贴上下文效率极低。后来我试过不少现成的 Agent 框架要么太重要么绑定特定平台要么配置复杂到劝退。于是干脆自己动手用 Python 写了一个轻量的 CLI 工具把“AI 决策”和“本地执行”这两层彻底打通。这个项目适合谁呢如果你满足下面任意一条那这篇内容应该对你有用写过 Python 但没怎么碰过 AI Agent 的想给自己日常脚本加个“智能调度层”的对 CLI 工具有执念、喜欢在终端里完成一切的或者单纯想看看一个 Agent 项目从零到跑起来到底要踩哪些坑。我会把设计思路、核心代码逻辑、实操步骤、以及我调试过程中遇到的各种幺蛾子都摊开讲尽量让你看完就能复现一个自己的版本。2. Agent-Reach 的整体设计与思路拆解2.1 核心架构三层分离各司其职Agent-Reach 的架构我改过三版前两版都因为耦合太严重被我推倒重来了。现在稳定下来的结构是三层接入层、决策层、执行层。这个分层不是拍脑袋定的而是被实际需求逼出来的。接入层负责跟大模型对话把用户的自然语言指令转成结构化的“行动计划”。决策层拿到计划后判断每一步该调用哪个本地能力比如读文件、跑脚本、查数据。执行层就是真正干活的调用 Python 的 subprocess、文件 IO、或者第三方库去完成任务。三层之间通过统一的 JSON 消息格式通信这样任何一层想换实现都不会影响另外两层。为什么这么分因为我一开始把模型调用和文件操作写在一个函数里结果想换个模型供应商的时候发现要改几十处代码。分层之后换模型只需要动接入层的一个适配器文件执行层完全不用碰。这个教训很值钱Agent 项目的复杂度不在于单个功能而在于功能之间的组合爆炸分层是控制复杂度的唯一手段。2.2 为什么选 CLI 而不是 Web 界面很多人问我都什么年代了还做 CLI我的理由很实在。第一我的使用场景 90% 在终端里写代码、跑脚本、看日志都在终端切到浏览器去跟 AI 对话再切回来这个上下文切换成本太高。第二CLI 天然适合管道操作我可以把 Agent-Reach 的输出直接 pipe 给 grep、jq 或者另一个脚本组合能力是 Web 界面给不了的。第三开发成本低不用折腾前端框架和部署一个人维护完全扛得住。当然 CLI 也有代价比如交互体验不如图形界面直观复杂任务的进度展示比较粗糙。我的取舍是优先保证核心链路的稳定和可组合性交互体验够用就行。如果你要做的是面向非技术用户的产品那 Web 界面可能更合适但如果是给自己或者小团队用的效率工具CLI 的性价比高得多。2.3 技术选型Python 为主关键路径考虑过 Rust主体用 Python这个没什么好纠结的。生态成熟写起来快调试方便而且我的大部分本地脚本本来就是 Python 写的直接复用。但我在设计执行层的时候认真考虑过用 Rust 重写关键模块原因是并发场景下 Python 的 GIL 确实是个瓶颈。具体来说当 Agent 需要同时调用多个外部工具时Python 的多线程在 CPU 密集型任务上基本没优势。我实测过用 threading 跑 10 个并行的数据处理任务总耗时跟串行差不多因为 GIL 把真正的并行锁死了。后来我改用 asyncio 加多进程的组合IO 密集的用 asyncioCPU 密集的丢给 ProcessPoolExecutor。这个方案在大多数场景下够用但如果你的 Agent 要扛很高的并发比如同时处理上百个请求那确实需要考虑用 Rust 写一个高性能的执行引擎通过 FFI 或者子进程的方式跟 Python 主逻辑通信。我目前的判断是先用 Python 把逻辑跑通等真的遇到性能瓶颈再针对性优化。过早用 Rust 重写开发效率会掉一大截而且很多性能问题其实通过架构调整就能解决不一定非要换语言。3. 核心细节解析与实操要点3.1 消息协议设计Agent 各层之间的“普通话”三层之间怎么通信这个协议设计是整个项目的地基。我最终定下来的格式是一个 JSON 对象包含action、params、context三个顶层字段。action是字符串标识要做什么比如read_file、run_script、query_db。params是参数字典具体内容由 action 决定。context用来传递会话状态比如上一步的输出、用户的原始指令等。# 消息协议示例 { action: run_script, params: { script_path: /home/user/scripts/clean_data.py, args: [--input, raw.csv, --output, clean.csv] }, context: { session_id: abc123, step: 2, previous_output: 数据加载完成共 10000 行 } }这个设计的关键在于action 和 params 的分离。一开始我把所有东西都塞进一个扁平字典里结果执行层要写一大堆 if-else 来判断参数含义非常丑陋。分离之后执行层只需要一个 action 到处理函数的映射表每个处理函数只关心自己的 params 结构代码清晰了不止一个档次。注意context 字段不要塞太多东西我踩过的坑是把整个对话历史都放进去结果消息体积膨胀到几 MB序列化和传输都成了瓶颈。后来改成只传必要的状态摘要历史记录单独存到本地 SQLite 里需要的时候再查。3.2 模型接入层的适配器模式接入层我用的是适配器模式每个模型供应商对应一个适配器类统一实现parse_intent和generate_response两个方法。这样换模型的时候只需要写一个新的适配器注册到工厂里就行。class BaseAdapter: def parse_intent(self, user_input: str) - dict: 把自然语言转成结构化行动计划 raise NotImplementedError def generate_response(self, result: dict) - str: 把执行结果转成自然语言回复 raise NotImplementedError class AdapterFactory: _adapters {} classmethod def register(cls, name: str, adapter_cls): cls._adapters[name] adapter_cls classmethod def get(cls, name: str) - BaseAdapter: return cls._adapters[name]()这里有个实操要点parse_intent 的提示词工程是重中之重。我试过直接让模型输出 JSON结果它经常在 JSON 外面包一层 markdown 代码块或者加一些解释性文字导致解析失败。后来我在提示词里明确要求“只输出 JSON不要任何其他内容”并且在解析前加了一个清洗步骤用正则把代码块标记去掉。这个清洗逻辑虽然丑但非常实用建议你也加上。3.3 执行层的安全边界执行层是最危险的地方因为它真的会跑命令、删文件、改数据。我给自己定了三条铁律第一所有文件操作限制在指定的工作目录内用os.path.realpath做路径校验防止../穿越。第二所有 shell 命令必须白名单只允许执行预注册的脚本不允许模型直接生成任意命令。第三所有写操作先备份改文件之前自动复制一份到.backup目录。import os ALLOWED_SCRIPTS { clean_data: /home/user/scripts/clean_data.py, gen_report: /home/user/scripts/gen_report.py, } def run_script(script_name: str, args: list): if script_name not in ALLOWED_SCRIPTS: raise PermissionError(f脚本 {script_name} 不在白名单中) script_path ALLOWED_SCRIPTS[script_name] # 后续执行逻辑...这三条铁律帮我避免了好几次事故。有一次模型理解错了指令想执行一个删除操作因为不在白名单里直接被拦下来了。Agent 的安全不能靠模型自觉必须靠代码层面的硬约束这是我从实际踩坑中得出的血泪教训。4. 实操过程与核心环节实现4.1 环境准备与依赖安装先把环境搭起来。我假设你已经装好了 Python版本建议 3.10 以上因为用到了 match-case 语法和一些新的类型注解特性。如果还没装去 Python 官网下载安装包安装时记得勾选“Add Python to PATH”这个选项能省掉后面很多配置麻烦。依赖方面核心就几个openai或者你用的模型供应商的 SDK、rich用来做终端输出美化、pydantic做数据校验。用 pip 一条命令搞定pip install openai rich pydantic如果你在国内pip 下载慢的话可以临时指定镜像源pip install openai rich pydantic -i https://pypi.tuna.tsinghua.edu.cn/simple提示建议用虚拟环境别直接装在系统 Python 里。python -m venv agent-env然后激活这样依赖隔离不会污染其他项目。4.2 项目骨架搭建目录结构我建议这样组织清晰且易于扩展agent-reach/ ├── main.py # CLI 入口 ├── adapters/ # 模型适配器 │ ├── __init__.py │ └── openai_adapter.py ├── core/ │ ├── __init__.py │ ├── dispatcher.py # 决策层负责路由 │ └── executor.py # 执行层负责干活 ├── utils/ │ ├── __init__.py │ └── logger.py └── config.yaml # 配置文件main.py是入口用argparse或者click解析命令行参数。我用的argparse因为标准库自带不用额外装依赖。核心逻辑是接收用户输入交给 dispatcherdispatcher 调用 adapter 解析意图然后路由到 executor 执行最后把结果返回给用户。4.3 核心调度逻辑实现dispatcher 是整个项目的大脑它的职责是协调 adapter 和 executor。我实现的时候用了简单的状态机IDLE - PARSING - EXECUTING - RESPONDING - IDLE。每个状态转换都有对应的日志记录方便排查问题。class Dispatcher: def __init__(self, adapter, executor): self.adapter adapter self.executor executor self.state IDLE def handle(self, user_input: str) - str: self.state PARSING plan self.adapter.parse_intent(user_input) self.state EXECUTING results [] for step in plan.get(steps, []): result self.executor.execute(step) results.append(result) if result.get(status) error: break self.state RESPONDING response self.adapter.generate_response({results: results}) self.state IDLE return response这里有个细节值得说执行过程中如果某一步失败我是直接中断还是继续我的选择是默认中断因为 Agent 的任务通常有依赖关系前一步失败后面继续跑可能产生错误结果。但我在配置里留了continue_on_error开关某些独立任务可以开启继续执行。4.4 参数计算与性能调优并发处理是 Agent 绕不开的话题。我实测过几种方案这里给一组具体数据。测试场景是同时处理 20 个文件读取加分析任务每个任务耗时约 0.5 秒。方案总耗时CPU 利用率内存占用纯串行10.2s15%80MBthreading9.8s18%95MBasyncio1.2s25%110MBasyncio 多进程0.8s65%320MB结论很清晰IO 密集型任务用 asyncio 收益最大CPU 密集型任务必须上多进程。我最终的方案是混合模式dispatcher 根据任务类型自动选择执行策略。判断逻辑很简单如果任务涉及网络请求或文件 IO走 asyncio如果涉及大量计算丢给进程池。import asyncio from concurrent.futures import ProcessPoolExecutor async def execute_io_task(task): # IO 密集型用 asyncio await asyncio.sleep(0) # 模拟异步操作 return await do_io_work(task) def execute_cpu_task(task): # CPU 密集型用多进程 with ProcessPoolExecutor() as pool: future pool.submit(do_cpu_work, task) return future.result()注意多进程方案下任务函数必须是可序列化的不能是 lambda 或者闭包。我一开始用 lambda 传参直接报 PicklingError排查了半天才发现是这个原因。改成顶层函数或者类的静态方法就好了。5. 常见问题与排查技巧实录5.1 模型输出格式不稳定怎么办这是最高频的问题。模型有时候输出纯 JSON有时候包代码块有时候加解释文字。我的解决方案是三层防护第一层提示词里明确格式要求并给示例第二层解析前用正则清洗第三层解析失败时自动重试并加强提示。import json import re def parse_model_output(text: str) - dict: # 清洗代码块标记 text re.sub(r^(?:json)?\s*, , text.strip()) text re.sub(r\s*$, , text) try: return json.loads(text) except json.JSONDecodeError: # 尝试提取第一个 JSON 对象 match re.search(r\{.*\}, text, re.DOTALL) if match: return json.loads(match.group()) raise实测下来加了清洗逻辑之后解析成功率从 70% 左右提升到 95% 以上。剩下 5% 的失败案例重试一次基本都能解决。5.2 执行超时与资源泄漏Agent 调用外部脚本时如果脚本卡死整个流程就挂了。我给每个执行步骤加了超时控制默认 30 秒超时直接 kill 进程并返回错误。同时用try-finally确保临时文件和数据库连接一定会被清理。import subprocess def run_with_timeout(cmd: list, timeout: int 30): try: result subprocess.run( cmd, capture_outputTrue, textTrue, timeouttimeout ) return {status: ok, output: result.stdout} except subprocess.TimeoutExpired: return {status: error, message: f执行超时{timeout}秒} finally: # 清理逻辑 pass5.3 常见问题速查表问题现象可能原因排查方向解决方案模型返回空结果API 配额耗尽或网络问题检查 API 响应状态码增加重试机制检查账户余额JSON 解析失败模型输出格式不规范打印原始输出查看加强提示词增加清洗逻辑脚本执行报权限错误文件权限或路径问题检查文件权限和路径用绝对路径检查 chmod 设置并发任务互相干扰共享状态未隔离检查全局变量和临时文件每个任务用独立的工作目录内存持续增长资源未释放用 memory_profiler 分析确保连接和文件句柄及时关闭5.4 我踩过的三个大坑第一个坑是路径问题。我在开发机上用相对路径跑得好好的部署到服务器上就找不到文件了。原因是工作目录变了。后来我强制所有路径都用os.path.abspath转成绝对路径问题消失。这个坑很低级但真的很常见。第二个坑是编码问题。Windows 上默认编码是 GBKLinux 上是 UTF-8我处理中文文件的时候在两边表现不一致。解决方案是显式指定encodingutf-8所有文件读写都加上这个参数一劳永逸。第三个坑是模型幻觉。有一次模型编了一个不存在的脚本名执行层直接报错。后来我在执行层加了白名单校验不在白名单里的操作一律拒绝。永远不要相信模型输出的内容所有关键操作都要有代码层面的校验这是做 Agent 项目的铁律。6. 后续扩展方向与个人体会Agent-Reach 目前跑通了我最核心的需求但还有不少可以打磨的地方。我接下来想做的几个方向一是加一个简单的 WebSocket 接口让其他程序也能调用 Agent 的能力二是把执行层的工具注册做成插件化通过配置文件动态加载不用改代码就能加新工具三是加一个任务队列支持长时间运行的任务异步执行完成后通知。如果你也想自己撸一个类似的工具我的建议是从最小可用版本开始别一上来就追求大而全。我第一版只支持读文件和跑脚本两个功能代码不到 200 行但已经能解决我 80% 的问题。后面加的功能都是在这个基础上逐步迭代出来的。Agent 这个领域变化很快保持代码的灵活性和可替换性比追求一步到位重要得多。另外多看看别人的 Agent 项目是怎么设计的但别照搬。每个项目的使用场景不一样适合别人的架构不一定适合你。我的三层分离架构就是参考了几个开源项目之后根据自己的实际需求调整出来的。理解设计背后的“为什么”比记住“怎么做”更有价值。