我是张大鹏做了十多年人工智能带过不少项目。说实话最难的不是把六边形架构画在PPT上是让一个784行的单体文件安全地拆成适配器还不让任何一行测试变红。这篇文章记录我如何在TDD的约束下把ga.py里的文件系统和代码执行功能完整抽离出来。一、上一篇文章埋的坑这周填上了如果你看过我上一篇《如意Agent六边形架构改造一从单体巨石到端口适配器》应该记得这张表阶段内容预计工时状态① 核心数据模型StepOutcome / MockResponse 提取0.5d✅已完成② 文件系统和代码执行适配器code_run / file_* → 适配器2d⏳ 本篇③ LLM 适配器llmcore.py 拆分为模块化 Session3-4d④ 浏览器和记忆适配器web_scan / memory 提取2d⑤ Agent ServiceGenericAgentHandler 注入化3-4d⑥ 装配层 DIContainer配置管理2d⑦ CLI 桌面适配器main.py / agentmain.py 变薄2-3d⑧ 清理收尾移除转发层1d第二阶段来了。这周的目标把ga.py里文件读写和代码执行这两个独立能力拆出来变成正经的适配器类。为什么要先动这两块两个原因它们是独立的技术能力——文件系统和代码执行不依赖 Agent 的业务逻辑天然适合定义在端口边界上代码味儿最重——file_read函数里有模糊匹配、关键词搜索、文件推荐code_run里有线程管理、超时控制、流式输出——这都是值得独立测试的逻辑二、刀尖上的手术三步走策略这次改造我严格遵循了三步走策略端口先行 → TDD实现 → 绞杀藤替换 定义接口 测试适配器 老函数变薄转发每一步都必须满足pytest tests/全量测试通过不能有任何回归。2.1 第一步定义端口Ports六边形架构的核心是端口——一个纯抽象的接口定义要做什么不定义怎么做。文件系统端口定义在src/core/port/for_file_system.py# src/core/port/for_file_system.pyfrom__future__importannotationsfromcollections.abcimportGeneratorclassForFileSystem:defread_file(self,path,start1,keywordNone,count200,show_linenosTrue):...defpatch_file(self,path:str,old_content:str,new_content:str)-str|dict:...defexpand_file_refs(self,text,base_dirNone):...defconsume_file(self,dr,file):...deffile_write(self,path,content,modeoverwrite):...代码执行端口定义在src/core/port/for_code_execution.py# src/core/port/for_code_execution.pyfrom__future__importannotationsfromcollections.abcimportGeneratorclassForCodeExecution:defrun_code(self,code:str,code_type:strpython,timeout:int60,cwd:str|NoneNone,code_cwd:str|NoneNone,stop_signal:list|NoneNone,)-Generator[str|dict,None,None]:...注意run_code的返回类型——它是一个Generator会先yield中间状态字符串给用户看执行进度最后return一个包含完整结果的dict。Generator 的函数体里yield和return并用这是 Python generator 的一个暗坑。2.2 第二步TDD 实现适配器先写测试确认失败 Red以CodeExecutionAdapter为例我写了 11 个测试覆盖所有路径# tests/adapter/driven/test_code_execution_adapter.pypytest.fixture()defadapter(tmp_path):returnCodeExecutionAdapter(data_dirstr(tmp_path))deftest_run_simple_python(adapter,tmp_path):resultslist(adapter.run_code(print(hello from adapter),timeout10,cwdstr(tmp_path)))lastresults[-1]assertlast[status]successasserthello from adapterinlast[stdout]deftest_run_with_timeout(adapter,tmp_path):codeimport time; time.sleep(30); print(should not reach)resultslist(adapter.run_code(code,timeout3,cwdstr(tmp_path)))lastresults[-1]assertlast[status]errordeftest_run_with_stop_signal(adapter,tmp_path):stop_signal[]codeimport time; time.sleep(30)resultslist(adapter.run_code(code,timeout10,cwdstr(tmp_path),stop_signalstop_signal))lastresults[-1]assertlast[status]errorstop_signal.append(True)# 触发 killfor_ingen:pass# 消费剩余deftest_code_with_unicode_output(adapter,tmp_path):resultslist(adapter.run_code(print(你好世界),timeout10,cwdstr(tmp_path)))lastresults[-1]assert你好世界inlast[stdout]测试覆盖正常执行、语法错误、不支持的代码类型、超时终止、用户手动终止、Generator 中间态、PowerShell、临时文件清理、Unicode 输出、空代码。然后实现适配器 GreenCodeExecutionAdapter的核心逻辑# src/adapter/driven/code_execution_adapter.pyclassCodeExecutionAdapter(ForCodeExecution):def__init__(self,data_dir:str|NoneNone,script_dir:str|NoneNone)-None:self._data_dirdata_dir self._script_dirscript_dirdefrun_code(self,code,code_typepython,timeout60,cwdNone,code_cwdNone,stop_signalNone)-Generator[str|dict,None,None]:ifstop_signalisNone:stop_signal[]preview(code[:60].replace(\n, )...)iflen(code)60elsecode.strip()yieldf[Action] Running{code_type}in{os.path.basename(cwdor)}:{preview}\ncwdcwdoros.path.join(self._get_data_dir(),temp)# Python → 临时文件执行ifcode_typein(python,py):tmp_filetempfile.NamedTemporaryFile(suffix.ai.py,deleteFalse,modew,encodingutf-8,dircode_cwd)cr_headeros.path.join(self._get_script_dir(),assets,code_run_header.py)ifos.path.exists(cr_header):tmp_file.write(open(cr_header,encodingutf-8).read())tmp_file.write(code)tmp_pathtmp_file.name tmp_file.close()cmd[sys.executable,-X,utf8,-u,tmp_path]# Shell → 直接命令执行elifcode_typein(powershell,bash,sh,shell,ps1,pwsh):ifos.nament:cmd[powershell,-NoProfile,-NonInteractive,-Command,code]else:cmd[bash,-c,code]# ... 线程化 stdout 读取、超时控制、kill 逻辑 ...FileSystemAdapter同样有 23 个测试覆盖正常读取、行号显示、关键词搜索、文件未找到时的模糊推荐、patch_file的安全性校验等。2.3 第三步绞杀藤替换——老函数变薄最微妙的一步——如何在不破坏任何导入的情况下让老函数指到新代码答案是ga.py变成转发层每个旧函数只保留签名 一行委托。我写了一个transform_ga.py脚本来自动完成这件事ga.py 改造前后对比 改造前784行 def code_run(code, ...): 代码执行器 python: 运行复杂的 .py 脚本 ... tmp_file tempfile.NamedTemporaryFile(...) process subprocess.Popen(...) t threading.Thread(targetstream_reader, ...) # ... 约 90 行原生实现 ... 改造后1 行 def code_run(code, ...): 代码执行器 python: 运行复杂的 .py 脚本 ... yield from _code_adapter.run_code(code, code_type, timeout, cwd, code_cwd, stop_signal)同时在ga.py底部注入适配器实例# ga.py — 在 _data_dir 初始化之后插入fromsrc.core.util.textimportformat_erroras_format_error_adapterfromsrc.core.util.textimportsmart_formatas_smart_format_adapterfromsrc.adapter.driven.file_system_adapterimportFileSystemAdapterfromsrc.adapter.driven.code_execution_adapterimportCodeExecutionAdapter _file_adapterFileSystemAdapter()_code_adapterCodeExecutionAdapter(data_dir_data_dir,script_dirscript_dir)这样做的好处from ga import code_run→仍然可用from ga import file_read→仍然可用from src.adapter.driven import FileSystemAdapter→新的标准路径三、这中间踩了三个坑3.1 坑一Generator 的yieldreturn混合code_run是一个 Generator 函数——先用yield输出中间状态字符串最后用return返回结果字典。端口接口必须准确反映这个类型# 错误丢失了 yield 信息defrun_code(...)-dict:...# 正确Generator 类型标注defrun_code(...)-Generator[str|dict,None,None]:...3.2 坑二docstring 拷贝的边界问题transform_ga.py脚本要识别函数签名后的 docstring 并保留它。但有些函数有 docstring有些没有比如file_read就直接以try:开头而file_read之后的GenericAgentHandler类的 docstring 包含。最初的实现用 in line检测结果file_read的docstring把整个类定义都吞了进去。修复方案只检查第一行是否以开头而不是搜索任意位置的。3.3 坑三子进程的当前工作目录CodeExecutionAdapter的cwd回退逻辑原本是os.path.join(self._get_data_dir(), temp)。但_data_dir在 PyInstaller 打包环境下指向sys.executable目录开发环境下又可能指向不适用的目录。修复方案显式传入cwdstr(tmp_path)给测试回退到os.getcwd()。四、成果量化改造完成后全量验证结果如下指标改造前改造后变化测试总数256 pass / 7 fail256 pass/ 7 fail0 回归适配器测试N/A34 个新测试✅ 全部通过ga.py行数784 行245 行-539 行-69%ruff 新代码N/AAll checks passed✅ga.pyruff 存量54 项46 项仅清理不可达 import新建文件08 个端口 适配器 测试目录结构的变化GenericAgent/ ├── src/ │ ├── core/ │ │ ├── port/ │ │ │ ├── for_file_system.py ← ★ 文件系统端口 │ │ │ └── for_code_execution.py ← ★ 代码执行端口 │ │ └── util/ │ │ └── text.py ← ★ smart_format / format_error │ └── adapter/ │ └── driven/ │ ├── file_system_adapter.py ← ★ 文件系统适配器23 测试 │ └── code_execution_adapter.py ← ★ 代码执行适配器11 测试 ├── ga.py ← 转发层245 行-69% └── tests/ ├── adapter/ │ └── driven/ │ ├── test_file_system_adapter.py ← 23 个测试 │ └── test_code_execution_adapter.py ← 11 个测试 └── core/ └── util/ └── test_text.py ← 16 个测试smart_format / format_error五、从绞杀藤到织锦第二阶段完成了回顾一下我的绞杀藤策略执行情况第一阶段核心数据模型StepOutcome / MockResponse→src/core/model/第二阶段文件系统和代码执行适配器 →src/adapter/driven/✅本篇下一阶段LLM 适配器 →src/adapter/driven/llm_adapter.py每一步都在不中断服务的前提下把一块能力从巨石里绞杀出来迁移到新的架构位置上。旧文件越来越薄新目录越来越丰满。如果说第一阶段是热身那第二阶段就是正式开刀。经过了第二阶段我更有信心了——这套方法可行。关键是每一步都要可验证测试通过、可回滚旧转发层还在、可独立合并不依赖后续阶段。总结维度内容核心思路端口-适配器分离 绞杀藤迁移策略ga.py函数变薄为单向转发第二阶段成果文件系统23 测试和代码执行11 测试适配器完全独立开发方式TDDRed → Green → Refactor34 个新测试全覆盖向后兼容from ga import code_run等所有导入路径不变迁移铁律256 测试全量通过0 回归539 行删除ruff 新代码零告警参考资料上一期如意Agent六边形架构改造一——从单体巨石到端口适配器Ports Adapters Architecture — Alistair CockburnStrangler Fig Application — Martin Fowler如意Agent 项目源码作者张大鹏日期2026-05-03系列如意Agent六边形架构改造