软件开发领域这几年有个词被反复提起软件工厂Software Factory。它并不是一个新概念但伴随着大模型和 AI Agent 的爆发这个老概念又被赋予了新的内涵。最近 HumanLayer 发布了一期关于“软件工厂设计模式”的播客讨论了很多 Agent 化开发、多智能体协作、主从模式、工具调用等话题。这篇文章不打算只做播客笔记而是结合设计模式、多 Agent 架构和实际工程落地一起聊聊我对软件工厂设计模式的理解并给出可运行的示例和踩坑记录。文章会从三个层面展开软件工厂设计模式是什么为什么现在讨论它。多 Agent 系统中的几种关键设计模式包括主从模式、工具调用模式、流水线模式。如何用 Python 实现一个极简的软件工厂原型并给出工程建议。无论你是刚接触 AI Agent 的开发者还是已经在做多智能体系统的工程师这篇文章都值得收藏备用。1. 背景与核心概念1.1 软件工厂不是一个新词“软件工厂”这个概念最早可以追溯到 1968 年 NATO 软件工程会议当时人们希望像工业生产一样用标准化、流水线、可复用的方式来生产软件。后来中间件、组件化、微服务、低代码平台本质上都是软件工厂思想的延续。不过传统软件工厂的“工人”是人工具是 IDE、编译器、持续集成系统产品是代码和二进制包。到了 AI 时代软件工厂的“工人”开始变成智能体Agent工具变成了模型 API、函数调用、外部服务产品可能是代码、文案、分析报告甚至是一个完整的小型应用。所以现在聊“软件工厂设计模式”本质上是在聊如何用一套可复用的架构套路把多个 AI Agent 组织起来像工厂流水线一样稳定地产出结果。1.2 HumanLayer 在讨论什么HumanLayer 是一家面向 AI Agent 开发者工具方向的公司它发布的播客内容通常会聊 Agent 工程的工程化问题。从“软件工厂设计模式”这期节目能看出他们关注的重点不是某个模型的参数而是Agent 之间如何协作、如何分工、如何管理状态、如何保证输出质量。这些话题在开发圈里越来越热原因很简单单个 Agent 做简单任务没问题但一旦涉及复杂业务比如“从需求分析到生成代码再到测试”单 Agent 很容易出现上下文丢失、任务中断、输出不稳定等问题。多智能体系统把任务拆开让不同 Agent 各司其职就需要设计模式来支撑。1.3 为什么开发者需要掌握这些模式我在实际项目里的感受是没有设计模式的多 Agent 系统前期跑 Demo 很爽后期维护就是灾难。你分不清某个 Agent 到底该负责什么。状态管理混乱A Agent 生成的结果B Agent 拿不到。出了问题不知道是模型的问题还是编排的问题。想加一个新功能牵一发动全身。设计模式解决的正是这些问题。它不是规定你“必须这么写”而是给出经过验证的、可复用的方案。2. 环境准备与版本说明本文的示例代码以 Python 3.10 为基础使用开源的 Agent 编排思路不依赖特定商业框架方便你理解底层逻辑。如果你已经在使用 LangChain、LlamaIndex 或自研的 Agent 框架核心思想是通用的。2.1 推荐环境组件建议版本/方案操作系统Windows 10/11、macOS 12、Ubuntu 20.04Python3.10 及以上包管理pip 或 poetry大模型 APIOpenAI 兼容接口或本地 Ollama开发工具VS Code、PyCharm 均可2.2 安装依赖本文示例主要用到以下 Python 库pip install pydantic openaipydantic用于结构化数据校验在 Agent 之间传递消息时非常有用。openai用于调用大模型接口测试时可以用本地模型替换。版本需要注意openai1.0.0 pydantic2.0.0如果你的网络环境无法访问外部 API可以使用 Ollama 拉起本地模型接口保持 OpenAI 兼容。下面的示例代码中我会把模型调用封装成一个函数方便替换。3. 核心设计模式拆解在 HumanLayer 的讨论里软件工厂设计模式可以被拆成多个子模式。这里我挑了三个最核心的主从模式、工具调用模式、流水线模式。3.1 主从模式本质上是把 Subagent 当作另一种 Tool最近社区里讨论“多 Agent 设计里的主从模式”时有一个观点很受认可主从模式本质上把 Subagent 当作另一种 Tool 进行调用。这个观点非常关键。传统编程里一个函数可以调用另一个函数Agent 编程里主 Agent 可以调用一个子 Agent就像调用一个函数一样。子 Agent 接收输入返回结果主 Agent 决定下一步做什么。这样做的好处是主 Agent 不需要关心子 Agent 的内部实现。子 Agent 可以被其他主 Agent 复用。状态管理变得更清晰主 Agent 只管编排子 Agent 只管执行。这也解释了为什么很多框架把 Agent 注册成 Tool。你调用一个“代码生成 Agent”本质上和调用一个“天气查询 API”没有区别都是输入输出。下面是一个极简的代码示意# 文件路径agent_patterns/master_slave.py from typing import Dict, Any class BaseAgent: def __init__(self, name: str): self.name name def run(self, task: str) - str: raise NotImplementedError class SubAgent(BaseAgent): def run(self, task: str) - str: # 实际项目中这里会调用大模型或本地模型 return f[{self.name}] 完成子任务: {task} class ToolWrapper: 把 Subagent 包装成 Tool 的形式 def __init__(self, agent: BaseAgent): self.agent agent def execute(self, task: str) - str: return self.agent.run(task) class MasterAgent(BaseAgent): def __init__(self, name: str): super().__init__(name) self.tools: Dict[str, ToolWrapper] {} def register_tool(self, tool_name: str, agent: BaseAgent): self.tools[tool_name] ToolWrapper(agent) def run(self, task: str) - str: # 主 Agent 的编排逻辑 if 编写代码 in task: sub_task task.replace(编写代码, ).strip() return self.tools[coder].execute(sub_task) elif 分析数据 in task: sub_task task.replace(分析数据, ).strip() return self.tools[analyst].execute(sub_task) else: return f[{self.name}] 无法处理: {task} if __name__ __main__: coder SubAgent(代码生成器) analyst SubAgent(数据分析师) master MasterAgent(主控Agent) master.register_tool(coder, coder) master.register_tool(analyst, analyst) print(master.run(编写代码 实现一个冒泡排序)) print(master.run(分析数据 统计销量))运行结果[代码生成器] 完成子任务: 实现一个冒泡排序 [数据分析师] 完成子任务: 统计销量这段代码演示了主从模式的核心主 Agent 内部持有多个子 Agent通过统一接口调用子 Agent 对主 Agent 透明。3.2 工具调用模式给 Agent 安装“手脚”工具调用模式是 AI Agent 的基础能力也是软件工厂里的“机床”。没有工具调用的 Agent 只能聊天有了工具调用的 Agent 才能执行动作。常见的工具包括代码执行器。数据库查询器。HTTP API 调用器。文件读写器。在 Python 中我们可以用普通的函数来表示一个工具并通过字典注册给 Agent。# 文件路径agent_patterns/tool_pattern.py import json def execute_python_code(code: str) - str: 执行一段 Python 代码返回结果。仅供测试生产环境需要沙箱。 try: exec_globals {} exec(code, exec_globals) return 执行成功 except Exception as e: return f执行失败: {e} def query_database(sql: str) - str: 模拟数据库查询实际项目中会连接真实数据库。 # 这里只做演示实际开发要防止 SQL 注入 return f查询结果: {sql} TOOL_REGISTRY { python_executor: execute_python_code, database_query: query_database, } def call_tool(tool_name: str, tool_input: str) - str: if tool_name not in TOOL_REGISTRY: return f找不到工具: {tool_name} func TOOL_REGISTRY[tool_name] return func(tool_input) if __name__ __main__: print(call_tool(python_executor, print(1 2))) print(call_tool(database_query, SELECT * FROM users WHERE id 1))这段代码展示了如何把工具注册、调用、错误处理统一起来。真正的 Agent 框架里工具描述会被序列化后交给模型让模型决定调用哪个工具。这里要特别提醒工具调用一定要注意安全边界。上面的代码执行器如果部署到生产环境必须使用沙箱如 Docker、Firecracker 或专用云函数否则会出现严重的安全漏洞。3.3 流水线模式把复杂任务拆成工序软件工厂最常见的组织方式就是流水线。一个需求从进入到交付要经过多个工序需求分析 - 方案设计 - 代码生成 - 测试 - 文档编写。在 Agent 系统里流水线模式让每个 Agent 只负责一个工序前一个 Agent 的输出是后一个 Agent 的输入。需求Agent - 设计Agent - 编码Agent - 测试Agent - 文档Agent这种模式的优点是职责单一、易于追踪缺点是如果某个环节出错后续环节会被连累。所以在流水线的每个环节之间最好加上“质量门禁”Quality Gate。# 文件路径agent_patterns/pipeline.py from typing import List, Callable class Pipeline: def __init__(self): self.steps: List[Callable[[str], str]] [] self.quality_gates: List[Callable[[str], bool]] [] def add_step(self, step: Callable[[str], str]): self.steps.append(step) def add_quality_gate(self, gate: Callable[[str], bool]): self.quality_gates.append(gate) def run(self, initial_input: str) - str: current initial_input for i, step in enumerate(self.steps): current step(current) # 质量门禁检查 for gate in self.quality_gates: if not gate(current): raise RuntimeError(f第 {i 1} 步输出未通过质量检查) return current def requirement_agent(req: str) - str: return f需求确认: {req} def design_agent(req: str) - str: return f设计文档: {req} - 模块A, 模块B def code_agent(design: str) - str: return f代码: {design} 已生成 def test_agent(code: str) - str: return f测试: {code} 通过 def not_empty(text: str) - bool: return len(text) 0 if __name__ __main__: p Pipeline() p.add_step(requirement_agent) p.add_step(design_agent) p.add_step(code_agent) p.add_step(test_agent) p.add_quality_gate(not_empty) result p.run(开发一个登录功能) print(result)流水线模式适合任务流程稳定、输入输出边界清晰的场景。如果你的任务经常要动态调整流程流水线就会显得僵硬这时可以考虑使用“规划器 执行器”的模式让规划 Agent 动态编排。3.4 设计模式的核心结构化输入输出不管哪种模式有一点是共通的Agent 之间传递的数据必须是结构化的。用自然语言直接传递结果短任务还行长任务很容易丢失信息。更可靠的做法是定义统一的数据结构用 pydantic 或 dataclass 传递。# 文件路径agent_patterns/schemas.py from pydantic import BaseModel from typing import List, Optional class Requirement(BaseModel): title: str description: str acceptance_criteria: List[str] [] class DesignDoc(BaseModel): module_list: List[str] architecture: str risks: List[str] [] class CodeResult(BaseModel): files: List[str] language: str description: str tests: Optional[List[str]] None class TestReport(BaseModel): passed: bool summary: str failed_cases: List[str] []这样每个 Agent 的输出都经过校验下游 Agent 可以安全地读取字段而不是靠正则表达式去“猜”上游结果。4. 完整实战实现一个迷你软件工厂现在我们把上面的模式组合起来实现一个迷你软件工厂。这个工厂接收一个需求描述自动完成需求分析 - 方案设计 - 代码生成 - 测试报告。为了可运行性我们使用模拟数据代替真实大模型调用但保留模型调用接口你可以直接替换为 OpenAI 或本地模型。4.1 项目结构software_factory/ ├── main.py ├── agents.py ├── pipeline.py └── schemas.py4.2 定义数据结构# 文件路径software_factory/schemas.py from pydantic import BaseModel from typing import List class Requirement(BaseModel): title: str description: str class DesignDoc(BaseModel): modules: List[str] data_flow: str class CodeResult(BaseModel): file_path: str code: str class TestReport(BaseModel): passed: bool summary: str4.3 定义各个 Agent# 文件路径software_factory/agents.py from schemas import Requirement, DesignDoc, CodeResult, TestReport def requirement_agent(user_input: str) - Requirement: 需求分析 Agent将用户的自然语言需求转换为结构化 Requirement。 实际项目可以调用大模型这里做简单的模拟。 # 模拟大模型返回 return Requirement( titleuser_input, descriptionf根据用户需求 {user_input} 生成的功能模块 ) def design_agent(requirement: Requirement) - DesignDoc: 方案设计 Agent根据需求生成模块设计和数据流。 return DesignDoc( modules[api_layer, business_logic, data_access], data_flowf用户请求 - api_layer - business_logic - data_access - {requirement.title} ) def code_agent(design: DesignDoc) - CodeResult: 代码生成 Agent根据设计方案生成代码文件。 code f 模块: {, .join(design.modules)} 数据流: {design.data_flow} def handle_request(request): # 业务逻辑层 result process(request) return result def process(request): # 数据访问层 return f处理请求: {{request}} return CodeResult( file_pathgenerated_app.py, codecode ) def test_agent(code_result: CodeResult) - TestReport: 测试 Agent对生成的代码进行静态检查和模拟测试。 # 模拟测试逻辑 if def in code_result.code: return TestReport(passedTrue, summary静态检查通过包含核心函数定义) else: return TestReport(passedFalse, summary未找到函数定义)4.4 实现流水线编排# 文件路径software_factory/pipeline.py from typing import Callable, Any from schemas import Requirement, DesignDoc, CodeResult, TestReport class SoftwareFactory: 软件工厂编排多个 Agent模拟从需求到测试的完整流程。 def __init__(self): self.steps [] def add_step(self, step: Callable[[Any], Any]): self.steps.append(step) def run(self, user_input: str): result user_input for step in self.steps: result step(result) return result4.5 主程序入口# 文件路径software_factory/main.py from agents import requirement_agent, design_agent, code_agent, test_agent from pipeline import SoftwareFactory def main(): factory SoftwareFactory() factory.add_step(requirement_agent) factory.add_step(design_agent) factory.add_step(code_agent) factory.add_step(test_agent) user_input 实现一个用户登录接口支持用户名密码校验 report factory.run(user_input) print(f测试结果: {report.passed}) print(f测试摘要: {report.summary}) if __name__ __main__: main()4.6 运行与验证在software_factory目录下执行cd software_factory python main.py预期输出测试结果: True 测试摘要: 静态检查通过包含核心函数定义这个迷你示例演示了软件工厂的核心流程需求结构化 - 设计 - 编码 - 测试。每一步之间传递的都是真实的 Python 对象而不是松散的自然语言字符串。4.7 如果接入真实大模型把agents.py里的模拟函数替换为真实模型调用即可。以 OpenAI 兼容接口为例# 文件路径software_factory/llm.py from openai import OpenAI client OpenAI( base_urlhttp://localhost:11434/v1, # 以 Ollama 为例 api_keyollama ) def chat(messages): response client.chat.completions.create( modelqwen2.5:7b, messagesmessages, temperature0.2 ) return response.choices[0].message.content然后在agents.py中使用chat函数把自然语言结果解析成 pydantic 对象。解析时建议使用 JSON 输出格式并让模型严格按照 schema 返回。5. 常见问题与排查思路在实际搭建 Agent 系统时常见问题不少。这里整理一个排查清单。问题现象常见原因解决思路主 Agent 不调用子 Agent工具描述不清晰或模型未启用工具调用检查工具描述是否包含完整的功能说明和参数说明子 Agent 输出和预期格式不一致没有强制结构化输出使用 pydantic 或 JSON Schema 约束模型输出流水线中断上个环节输出为空或异常增加质量门禁和异常捕获重试或终止状态丢失多个 Agent 共享可变状态使用不可变数据结构或引入独立的状态存储同一个请求重复调用大模型编排逻辑写错打印日志检查每一步输入输出是否重复上下文窗口超限传递的内容过长截断摘要或使用长期记忆存储几个容易踩的坑单独说一下。5.1 不要用自然语言传对象如果 A Agent 返回一段文字“我生成了代码代码在 /tmp/a.py内容是……”B Agent 解析这段文字就是灾难。格式稍微变化解析就挂了。解决办法Agent 之间传 pydantic 对象或 JSON。每个 Agent 的输入输出都定义清楚。5.2 注意 Agent 的“幻觉”扩散在流水线模式里如果需求 Agent 就理解错了用户的意图后面的设计、编码、测试全都会跟着错而且越往后错误越隐蔽。解决办法在每个阶段都加“确认点”。比如设计完成后可以让用户或审核 Agent 确认设计是否合理再进入编码阶段。HumanLayer 在讨论里强调 human-in-the-loop就是这个意思。5.3 重试要有上限大模型接口偶尔会超时、报错但无限重试只会加重问题。建议给每个 Agent 设置最大重试次数通常 2 到 3 次超过次数直接进入降级流程。MAX_RETRY 3 def run_with_retry(func, *args, **kwargs): for attempt in range(MAX_RETRY): try: return func(*args, **kwargs) except Exception as e: if attempt MAX_RETRY - 1: raise print(f第 {attempt 1} 次失败: {e})6. 最佳实践与工程建议6.1 从简单模式开始很多团队一上来就搞复杂的多 Agent 协作结果调试成本巨大。我建议从单 Agent 工具开始跑通之后再升级为主从模式或流水线模式。设计模式是解决问题的不是用来炫技的。6.2 重视可观测性Agent 系统的一大难点是“不可控”。传统程序有清晰的调用栈Agent 系统的调用链经常是黑盒。所以必须打日志记录每个 Agent 的输入摘要。每个 Agent 的输出摘要。每次大模型调用的 token 消耗。每个环节的耗时。有了日志你才能在出问题时快速定位。6.3 安全边界要前置设计前面提到了代码执行、数据库操作这些工具一定要加权限控制。哪怕是内部系统也不要随便让 Agent 执行任意代码。建议代码执行使用沙箱容器。数据库账号使用只读权限除非明确需要写操作。敏感操作增加人工审批环节。6.4 成本控制多 Agent 系统往往意味着多次大模型调用。一个看似简单的任务背后可能消耗了上万 token。建议在 Agent 之间传递内容时尽量压缩长度。使用价格更低的模型处理简单任务。为每个 Agent 设置预算上限。6.5 版本化你的 Agent 配置Agent 的提示词、工具列表、模型参数都应该像代码一样纳入版本管理。推荐把 Agent 配置写成 YAML 文件方便评审和回滚。# 文件路径config/agent_config.yaml master_agent: model: gpt-4o-mini temperature: 0.2 tools: - name: coder_agent description: 生成代码的子Agent - name: test_agent description: 负责测试的子Agent7. 总结与学习路线这篇文章从 HumanLayer 的软件工厂设计模式播客出发梳理了多 Agent 系统的三个核心设计模式主从模式、工具调用模式、流水线模式并用 Python 实现了一个迷你软件工厂。你需要记住的关键点主从模式本质上是把 Subagent 当作 Tool 来调用。Agent 之间传递结构化对象而不是自然语言字符串。流水线模式适合稳定流程动态流程要引入规划器。可观测性、安全边界、成本控制是生产级 Agent 系统的三座大山。下一步的学习建议如果你正在用 LangChain 或 LlamaIndex去阅读它们的 Agent 和 Tool 源码理解它们是如何实现主从模式和工具调用的。尝试把本文的迷你软件工厂接入真实大模型看结构化输出解析会遇到什么问题。研究多 Agent 框架如 AutoGen、CrewAI的 GroupChat 和 Manager 机制它们把主从模式工程化了。如果你的工作中已经用到了多智能体架构可能对“职责边界划分”和“状态同步”体会更深也欢迎在评论区聊聊你自己踩过的坑。这篇文章值得你收藏起来等真正构建 Agent 系统的时候拿出来对照着做设计。