MetaGPT：面向工程落地的多角色AI协作操作系统

1. MetaGPT 是什么？它不是另一个大模型，而是一套让 AI“团队协作”的操作系统

你有没有试过让 ChatGPT 写一份完整的商业计划书？它能写出漂亮的执行摘要、市场分析段落，甚至财务预测的模板——但当你要求它“把这份计划书转成带动画的 PPT，自动配图，再生成一份给投资人看的 3 分钟语音脚本，并同步发到你的邮箱”，它大概率会卡在第一步：它不知道 PowerPoint 怎么调用，不清楚图片版权怎么查，更没法真正点击“发送邮件”按钮。这不是模型能力不够，而是它的角色定位太单一：它是个“超级文员”，不是“项目经理”。

MetaGPT 就是为解决这个根本矛盾而生的。它不训练新模型，也不追求参数规模更大；它干的是更底层、更务实的事——给大语言模型（LLM）装上组织架构、岗位职责、工作流程和协作协议。你可以把它理解成一套“AI 团队的操作系统”：当你要开发一个新网站，MetaGPT 不是让一个模型硬扛所有活，而是自动拉起一支虚拟小队——产品经理负责拆解需求、写 PRD；架构师设计技术方案；前端工程师写 React 代码；测试工程师写单元测试用例；最后由项目经理统一对齐进度、协调交付。每个角色都是同一个基础大模型（比如 Qwen 或 Llama3）驱动的独立 Agent，它们之间通过结构化消息（JSON 格式）沟通，像真实团队一样开站会、传文档、互相 Review。

这背后的关键突破在于“角色即能力封装”。传统 Agent 框架里，一个 Agent 往往对应一个工具调用函数（比如“查天气”或“搜网页”），而 MetaGPT 把“角色”本身变成了可复用的能力单元。产品经理 Agent 的核心不是调用某个 API，而是内嵌了需求分析方法论（比如 KANO 模型）、PRD 写作规范、用户故事地图模板；测试工程师 Agent 则内置了等价类划分、边界值分析等测试设计思维。这些不是 prompt 工程的临时技巧，而是被固化在角色定义里的专业认知框架。所以它解决的从来不是“能不能做”，而是“能不能像专业人士那样有章法地做”。如果你正被“LLM 输出很泛、落地很散、多步骤任务容易断链”这类问题困扰，MetaGPT 提供的不是又一个玩具 Demo，而是一条通向工程化 AI 应用的清晰路径——它把“让 AI 做事”这件事，从单点突破升级成了系统作战。

2. 项目整体设计与思路拆解：为什么必须是“多角色协同”，而不是“单模型增强”

2.1 核心设计哲学：用组织分工替代能力堆砌

很多人初看 MetaGPT，第一反应是：“这不就是把 ChatGPT 包装成多个马甲？” 实际上完全相反。它的设计起点恰恰是否定了“靠一个模型越变越强就能解决一切”的幻想。我们来算一笔账：假设你要做一个电商后台管理系统，需求包括用户管理、订单处理、库存预警、数据报表四个模块。如果用单个 LLM Agent 硬啃，它得同时掌握：

• 用户权限模型（RBAC/ABAC）
• 订单状态机流转逻辑（待支付→已发货→已完成→已退款）
• 库存扣减的并发控制（Redis 分布式锁 or 数据库乐观锁）
• BI 工具的 SQL 聚合语法（GROUP BY + HAVING + 窗口函数）

这已经远超任何当前开源模型的知识覆盖边界。更致命的是，当它在写订单模块时，可能因过度关注库存逻辑而漏掉支付回调的幂等性设计；写报表时又可能为了炫技用复杂窗口函数，却忽略了前端渲染性能。单点智能的代价是全局失焦。

MetaGPT 的破局点在于“责任隔离”。它把上述四个模块直接映射为四个角色：

• Product Manager：只关心“用户要什么”，输出带验收标准的用户故事（如“作为仓库管理员，我需要看到库存低于安全阈值的商品列表，以便及时补货”）；
• Software Architect：只响应 PM 的 PRD，输出技术选型对比表（Spring Boot vs FastAPI）、数据库 ER 图、API 接口契约（OpenAPI 3.0 格式）；
• Frontend Engineer：只读取 Architect 输出的接口契约，生成 Vue 组件代码+Mock 数据；
• QA Engineer：只基于 PM 的用户故事和 Architect 的状态机，生成包含异常流的测试用例（如“模拟网络中断时重复提交订单，验证是否触发幂等校验”）。

每个角色的输入/输出都被严格约束，形成天然的“认知防火墙”。这不仅是工程实践，更是对人类专业分工本质的复刻——医生不会同时操刀、开药、写病历、管药房，因为深度专注才能保证每个环节的可靠性。MetaGPT 把这套逻辑搬进了 AI 世界。

2.2 架构分层：从“消息总线”到“知识基座”的四层设计

MetaGPT 的代码结构看似简单，实则暗藏精妙的分层逻辑。我把它拆解为四个不可跳过的层级，每一层都解决了特定维度的协作难题：

第一层：Role Layer（角色层）
这是最直观的层面，定义了ProductManager、Engineer等类。但关键细节在于：每个角色类都强制继承Role基类，并重写run()方法。这个run()不是自由发挥的 prompt，而是遵循固定模式：

1. 读取self.rc.memory.get()获取最新消息（来自其他角色或用户）
2. 执行角色专属动作（如 PM 调用self._write_prd()，Architect 调用self._design_system()）
3. 将结果封装为Message(content=..., role="ProductManager", cause_by="UserRequirement")发送

提示：cause_by字段是协作的灵魂。它记录消息的原始触发源（如"UserRequirement"表示用户输入，"CodeReview"表示被工程师质疑），让每个角色能追溯决策链条。没有这个字段，整个系统会退化成无序聊天。

第二层：Message Bus（消息总线）
很多人忽略 MetaGPT 最反直觉的设计：它没有中心化的“调度器”。角色间通信走的是异步消息队列（默认用内存队列asyncio.Queue）。当 PM 生成 PRD 后，它不直接调用 Architect 的方法，而是await self.send(Message(...))。Architect 的run()方法则持续监听队列，收到cause_by="UserRequirement"的消息才启动。这种松耦合设计带来两个硬收益：

• 故障隔离：若 Architect 崩溃，PM 仍可继续接收用户需求并存档，不影响前端工程师处理已有任务；
• 可插拔性：你想把 Architect 替换为 Claude 3，只需改写其run()中的模型调用逻辑，其他角色完全无感。

第三层：Memory & Context（记忆与上下文）
MetaGPT 的RoleContext类是真正的黑科技。它不像普通 Agent 那样用长文本拼接历史，而是构建了三重记忆索引：

• 短期记忆：当前任务的对话流（类似 Chat History）
• 中期记忆：该角色在本次任务中产出的所有工件（PRD 文档、ER 图、测试用例），按类型打标签存储
• 长期记忆：跨任务的领域知识（如“电商系统中订单状态机有 7 种状态”），通过向量数据库（Chroma）持久化

这意味着当 QA 工程师需要设计测试用例时，它不仅能读取 PM 的最新 PRD，还能自动关联 Architect 之前画过的状态机图，甚至检索出历史项目中同类订单模块的 Bug 清单——这才是专业工程师的真实工作方式。

第四层：Action Layer（动作层）
这是连接 AI 与现实世界的桥梁。MetaGPT 预置了WriteFile、RunCode、SearchWeb等基础 Action，但真正体现设计功力的是它的“动作组合范式”。例如RunCode动作并非简单执行subprocess.run()，而是：

1. 先调用WriteFile把代码写入临时目录
2. 启动沙箱环境（Docker 容器）执行
3. 捕获 stdout/stderr 并结构化解析（区分编译错误、运行时异常、正常输出）
4. 将结果连同代码文件哈希值存入 Memory

这种“原子动作+组合协议”的设计，让 MetaGPT 天然支持安全审计——你可以随时回溯：“这个报错是哪段代码在哪次执行中产生的？当时输入的测试数据是什么？”

2.3 为什么不用 LangChain / LlamaIndex？—— 工程视角下的框架选型逻辑

当我在团队内部推动 MetaGPT 落地时，第一个被挑战的问题就是：“既然 LangChain 生态这么成熟，为什么还要自己造轮子？” 这个问题问到了根子上。我用一次真实踩坑经历来说明：

我们曾尝试用 LangChain 的AgentExecutor改造一个客服工单处理系统。当用户投诉“订单 12345 未发货”，LangChain Agent 会：

1. 用Tool查询订单状态 → 返回“已支付，未发货”
2. 用Tool查询物流单号 → 返回“空”
3. 用Tool触发补发流程 → 返回“成功”

表面看很流畅。但当用户追问“补发的快递单号是多少？”，系统就懵了——因为前三个 Tool 调用是孤立的，没有共享上下文。LangChain 的AgentExecutor本质是“单次决策循环”，它无法记住“刚才我查过订单 12345，现在要查它的物流单号”，每次都要重新解析用户意图。

MetaGPT 的解法是彻底重构交互范式：

• 用户输入 → PM 角色解析为结构化工单（含订单号、投诉类型、期望结果）
• PM 将工单存入 Memory → Architect 基于工单生成处理流程图（含分支判断：若物流单号为空，则触发补发；若补发失败，则升级人工）
• 所有后续操作都基于这张流程图执行，状态全程可追溯

这背后是两种哲学的差异：LangChain 在解决“如何调用工具”，MetaGPT 在解决“如何构建业务逻辑”。前者是战术级工具链，后者是战略级操作系统。当你面对的是“开发一个完整 SaaS 应用”这种复杂任务时，前者会让你陷入 endless loop 的 prompt 调试，后者则给你一张清晰的作战地图。

3. 核心细节解析与实操要点：从零部署一个可工作的 MetaGPT 电商项目

3.1 环境准备：避开 Python 版本与依赖冲突的深坑

MetaGPT 对环境的要求看似宽松（Python 3.9+），但实际部署中 80% 的失败都源于依赖冲突。我整理了经过生产验证的最小可行环境配置：

# 创建干净的虚拟环境（强烈建议不用 conda，MetaGPT 的 asyncio 依赖与 conda 的 event loop 有兼容问题） python3.10 -m venv metagpt-env source metagpt-env/bin/activate # 安装核心依赖（注意版本锁定！） pip install "metagpt==0.7.10" # 必须指定版本，0.8.x 引入了 breaking change pip install "openai==1.12.0" # 与 MetaGPT 0.7.10 兼容的最佳版本 pip install "pydantic==1.10.12" # 关键！高版本 pydantic v2 会导致 Role 初始化失败 pip install "chromadb==0.4.22" # 向量数据库，0.4.x 系列最稳定

注意：不要用pip install metagpt[all]。这个命令会安装所有可选依赖（包括 Docker、Kubernetes 相关包），但其中docker-py与kubernetes在 macOS 上存在二进制冲突，导致import metagpt直接报错。生产环境应坚持“按需安装”，只加你需要的扩展。

最关键的环境变量配置（.env文件）：

# OPENAI_API_KEY 必须设置，否则启动就报错 OPENAI_API_KEY=sk-xxx # 设置工作空间根目录（所有生成的代码/文档都放这里） WORKSPACE_ROOT=/path/to/your/workspace # 启用 ChromaDB 持久化（默认是内存模式，重启就丢数据） CHROMA_PERSIST_DIR=/path/to/chroma/data # 日志级别设为 DEBUG，否则你看不到角色间的消息流转细节 LOG_LEVEL=DEBUG

实操心得：第一次运行时，务必在终端开启tail -f logs/metagpt.log。MetaGPT 的日志设计非常友好，每条消息都标注了role: ProductManager,cause_by: UserRequirement,sent_to: SoftwareArchitect，这是你理解协作流程的唯一可靠依据。很多新手抱怨“系统没反应”，其实只是日志级别太低，看不到后台消息在静默流转。

3.2 角色定制：如何让 Product Manager 真正懂你的业务

MetaGPT 自带的角色模板（metagpt/roles/product_manager.py）是通用电商场景，但你的业务一定有特殊规则。比如我们做跨境 SaaS 时，PM 必须强制检查“海关申报要素”是否在 PRD 中体现。改造方法如下：

1. 创建自定义角色类（my_roles/custom_pm.py）：

from metagpt.roles import ProductManager from metagpt.schema import Message from metagpt.actions import WritePRD class CrossBorderPM(ProductManager): def __init__(self, **kwargs): super().__init__(**kwargs) # 注入领域知识：海关申报的 6 大核心字段 self.custom_knowledge = { "hs_code": "商品海关编码，6-10位数字", "country_of_origin": "原产国，ISO 3166-1 alpha-2 格式", "export_license_required": "是否需要出口许可证" } async def _write_prd(self, requirements: list[str]) -> Message: # 在原始 PRD 生成逻辑前，插入业务校验 if not any("海关" in req or "申报" in req for req in requirements): requirements.append("必须包含海关申报要素（HS编码、原产国、出口许可证要求）") # 调用父类方法生成 PRD prd_msg = await super()._write_prd(requirements) # 对生成的 PRD 内容做后处理 prd_content = prd_msg.content if "海关申报" not in prd_content: prd_content += "\n\n## 海关申报要求\n" for field, desc in self.custom_knowledge.items(): prd_content += f"- `{field}`: {desc}\n" return Message(content=prd_content, role=self.profile, cause_by=WritePRD)

2. 在启动脚本中注册新角色（run_crossborder.py）：

from metagpt.software_company import SoftwareCompany from my_roles.custom_pm import CrossBorderPM # 创建公司实例时，用自定义角色替换默认角色 company = SoftwareCompany( roles=[ CrossBorderPM(), # 其他角色保持默认 ] ) # 启动时指定初始需求 await company.run("开发一个跨境订单管理后台，支持 HS 编码自动匹配")

这个改造的关键在于：不修改 MetaGPT 源码，只通过继承和重写实现业务定制。所有业务规则都封装在角色类内部，与框架解耦。当你需要切换到医疗 SaaS 场景时，只需新建HealthcarePM类，完全不影响现有代码。

3.3 消息协议详解：读懂角色间“加密通话”的真实含义

MetaGPT 的协作本质是消息驱动，而消息格式就是它的“协议标准”。一个典型的消息对象长这样：

{ "content": "用户需求：开发一个支持实时库存预警的电商后台。\n验收标准：当商品库存低于 50 件时，自动发送企业微信通知。", "role": "User", "cause_by": "UserRequirement", "sent_from": "User", "sent_to": "ProductManager", "timestamp": "2024-06-15T10:23:45.123Z", "id": "msg_abc123" }

但真正决定协作质量的是cause_by和sent_to字段的组合。我整理了生产环境中最常出现的 7 种消息类型及其处理逻辑：

提示：cause_by字段是调试协作流的黄金线索。当你发现某个角色没响应时，先检查日志中它收到的消息cause_by是什么——如果cause_by是"CodeReview"但角色是Engineer，那说明流程已进入返工阶段，而非初始开发。

3.4 工件生成规范：让 AI 输出符合工程交付标准

MetaGPT 生成的代码/文档常被吐槽“看着很美，用不了”。根本原因在于它默认输出的是“演示级工件”，而非“生产级工件”。我们必须通过配置强制提升交付标准。以下是针对电商项目的 4 项关键配置：

1. 代码生成强制规范（在metagpt/actions/write_code.py中修改）：

# 原始代码只生成 .py 文件，我们增加： def _generate_file_structure(self, code: str) -> dict: # 强制生成标准 Django 项目结构 return { "models.py": self._generate_models(code), "views.py": self._generate_views(code), "serializers.py": self._generate_serializers(code), # 新增：必须生成序列化器 "tests/test_models.py": self._generate_model_tests(code), # 新增：必须生成单元测试 "docs/api_spec.md": self._generate_openapi_spec(code) # 新增：必须生成 OpenAPI 文档 }

2. PRD 文档强制字段（在metagpt/actions/write_prd.py中）：

# 在 PRD 模板中硬编码必填章节 PRD_TEMPLATE = """ # {title} ## 1. 背景与目标 {background} ## 2. 用户故事 {user_stories} ## 3. 验收标准（AC） {acceptance_criteria} ## 4. 非功能需求（NFR） - 性能：首页加载 < 1s（P95） - 安全：所有 API 必须 JWT 认证 - 合规：符合 GDPR 数据删除要求 # 新增：强制合规条款 ## 5. 数据字典 {data_dictionary} # 新增：必须定义所有字段类型、长度、约束 """

3. 测试用例生成策略（在metagpt/actions/write_testcase.py中）：

# 不再随机生成，而是基于 PRD 中的 AC 自动生成 def _generate_test_cases_from_ac(self, ac_list: list[str]) -> list[str]: test_cases = [] for i, ac in enumerate(ac_list, 1): # 每个 AC 生成 3 个测试用例：正常流、边界值、异常流 test_cases.extend([ f"TC{i}-1 正常流：{ac} → 预期：成功", f"TC{i}-2 边界值：输入临界值（如库存=50）→ 预期：触发预警", f"TC{i}-3 异常流：输入非法值（如库存=-1）→ 预期：返回 400 错误" ]) return test_cases

4. 输出产物打包逻辑（在metagpt/software_company.py中重写）：

def _package_output(self, output_dir: Path): # 不再只打包代码，而是生成标准交付包 deliverables = { "1_PRD": (output_dir / "PRD.md"), "2_Technical_Design": (output_dir / "DESIGN.md"), "3_Source_Code": (output_dir / "src/"), "4_Test_Report": (output_dir / "TEST_REPORT.md"), "5_Deployment_Guide": self._generate_deployment_guide() # 新增：必须提供部署指南 } # 生成 ZIP 包并计算 SHA256 校验码（用于审计） zip_path = output_dir / "deliverables.zip" with zipfile.ZipFile(zip_path, 'w') as zf: for name, path in deliverables.items(): if path.is_file(): zf.write(path, f"deliverables/{name}") else: for file in path.rglob("*"): if file.is_file(): zf.write(file, f"deliverables/{name}/{file.relative_to(path)}") # 生成校验文件 with open(output_dir / "SHA256SUMS", "w") as f: f.write(f"{hashlib.sha256(zip_path.read_bytes()).hexdigest()} deliverables.zip\n")

这套规范带来的改变是质的：生成的代码可以直接git clone进团队仓库，PRD 文档能直接作为需求评审材料，测试报告可导入 Jira。它把 AI 从“灵感助手”升级为“交付伙伴”。

4. 实操过程与核心环节实现：手把手完成一个库存预警系统的端到端开发

4.1 启动命令与初始需求输入

我们以开发“电商库存实时预警系统”为例，这是最能体现 MetaGPT 协作价值的典型场景。启动命令极其简洁：

# 进入 MetaGPT 根目录 cd /path/to/metagpt # 设置环境变量（确保 .env 已配置） source .env # 运行单次任务（不进入交互模式） python main.py --req "开发一个库存预警系统：当商品库存低于安全阈值时，自动发送企业微信通知。安全阈值需在商品详情页可配置。"

注意：--req参数必须是完整句子，不能是关键词。MetaGPT 的 PM 角色对输入格式敏感，"库存预警"会被识别为模糊需求，而"当商品库存低于安全阈值时，自动发送企业微信通知"包含明确的触发条件、动作、对象，能直接映射为用户故事。

启动后，你会看到终端滚动输出（截取关键片段）：

[INFO] ProductManager received message from User (cause_by=UserRequirement) [DEBUG] ProductManager is writing PRD for: 开发一个库存预警系统... [INFO] ProductManager sent PRD to SoftwareArchitect (cause_by=PRD) [INFO] SoftwareArchitect received message from ProductManager (cause_by=PRD) [DEBUG] SoftwareArchitect is designing system for inventory alert... [INFO] SoftwareArchitect sent SystemDesign to Engineer (cause_by=SystemDesign)

这个日志流就是协作的“心跳”。只要看到sent_to和received成对出现，说明消息总线工作正常。如果卡在某一步超过 30 秒，基本可以判定是模型调用超时（检查 OPENAI_API_KEY 是否有效，或网络是否通畅）。

4.2 PRD 生成环节：如何让 AI 写出可评审的需求文档

PM 角色生成的 PRD 会保存在workspace/PRD.md。我们来看一个真实生成的片段（已脱敏）：

# 库存预警系统 PRD ## 1. 背景与目标 解决运营人员无法及时发现低库存商品的问题，将库存预警响应时间从小时级缩短至秒级。 ## 2. 用户故事 - 作为运营人员，我需要在商品详情页设置安全库存阈值，以便灵活调整预警标准。 - 作为系统管理员，我需要查看所有触发预警的商品列表，并能手动关闭误报。 - 作为企业微信用户，我需要收到包含商品名称、当前库存、安全阈值的预警消息。 ## 3. 验收标准（AC） - AC1：商品详情页新增“安全库存”输入框，默认值为 50，支持整数输入（1-10000）。 - AC2：当商品库存 ≤ 安全库存时，系统在 5 秒内触发预警。 - AC3：预警消息发送至企业微信指定群组，包含商品ID、名称、当前库存、安全阈值、预警时间。 - AC4：管理员可在后台“预警中心”查看所有活跃预警，并点击“关闭”按钮停止推送。 ## 4. 非功能需求（NFR） - 性能：预警延迟 ≤ 5 秒（P99） - 安全：安全库存阈值修改需二次确认（输入管理员密码） - 合规：预警消息中不包含用户手机号等 PII 信息 ## 5. 数据字典 | 字段名 | 类型 | 长度 | 约束 | 描述 | |--------|------|------|------|------| | product_id | VARCHAR | 32 | NOT NULL, PK | 商品唯一标识 | | safety_stock | INTEGER | 4 | DEFAULT 50, CHECK(safety_stock > 0) | 安全库存阈值 | | current_stock | INTEGER | 4 | NOT NULL | 当前实时库存 | | last_alert_time | DATETIME | - | NULLABLE | 上次预警时间 |

这份 PRD 的价值在于：

• AC 条款可直接转化为测试用例（QA 角色会为每个 AC 生成 3 个测试用例）
• 数据字典可直接导入数据库建表语句（Architect 角色会据此生成 SQL）
• NFR 明确了安全与合规红线（避免后期返工）

实操心得：如果 PRD 中缺少关键字段（比如没定义last_alert_time），不要手动修改.md文件。正确的做法是重新运行main.py --req，并在需求描述中强调：“必须记录每次预警的触发时间，以便做时效性分析”。MetaGPT 的 PM 会根据新提示重新生成 PRD。

4.3 技术方案设计：Architect 如何输出可落地的架构图

Architect 角色生成的DESIGN.md是技术决策的核心。它不会画 UML 图（那是设计师的工作），而是输出可执行的技术契约。以下是从真实输出中提取的关键部分：

## 系统架构设计 ### 1. 技术选型 - 后端框架：FastAPI（轻量、异步、OpenAPI 原生支持） - 数据库：PostgreSQL 14（支持 JSONB 字段存储预警配置） - 消息队列：Redis Streams（低延迟、支持消费者组） - 通知服务：企业微信 Webhook（官方 API，无需 OAuth） ### 2. 核心流程图 ```mermaid graph LR A[商品库存变更] --> B{库存 ≤ 安全阈值?} B -->|Yes| C[写入 Redis Stream: inventory_alert] B -->|No| D[结束] C --> E[Alert Consumer 从 Stream 读取] E --> F[调用企业微信 Webhook] F --> G[更新 last_alert_time]

3. API 接口契约（OpenAPI 3.0）

/openapi.yaml paths: /products/{product_id}/safety-stock: put: summary: 更新商品安全库存阈值 parameters: - name: product_id in: path required: true schema: {type: string} requestBody: required: true content: application/json: schema: type: object properties: safety_stock: type: integer minimum: 1 maximum: 10000 responses: '200': description: 更新成功

4. 数据库 ER 图

-- products 表（新增 safety_stock 字段） ALTER TABLE products ADD COLUMN safety_stock INTEGER NOT NULL DEFAULT 50, ADD COLUMN last_alert_time TIMESTAMP; -- 预警日志表（用于审计） CREATE TABLE alert_logs ( id SERIAL PRIMARY KEY, product_id VARCHAR(32) NOT NULL, current_stock INTEGER NOT NULL, safety_stock INTEGER NOT NULL, alert_time TIMESTAMP NOT NULL, status VARCHAR(20) DEFAULT 'sent' -- 'sent', 'failed', 'manually_closed' );

这份设计的价值在于： - **所有技术选型都有明确理由**（FastAPI 因“异步”被选中，而非“流行”） - **流程图用 Mermaid 语法，可直接粘贴到 Confluence 渲染** - **OpenAPI 契约可一键生成客户端 SDK**（用 `openapi-generator`） - **SQL 语句可直接在数据库执行**（无需 DBA 二次审核） > 提示：如果 Architect 输出的架构过于复杂（比如提议用 Kafka 替代 Redis），说明你的初始需求描述太宽泛。回到 `--req` 参数，加上约束：“使用最简技术栈，避免引入新中间件”。 ### 4.4 代码生成与测试：Engineer 与 QA 的攻防演练 Engineer 角色生成的代码位于 `workspace/src/`，结构如下：

src/ ├── main.py # FastAPI 入口 ├── models.py # Pydantic 模型定义 ├── database.py # PostgreSQL 连接池 ├── redis_client.py # Redis Streams 封装 ├── wecom_notifier.py # 企业微信通知类 ├── services/ │ ├── inventory_service.py # 库存变更核心逻辑 │ └── alert_service.py # 预警触发与去重 ├── tests/ │ ├── test_inventory_service.py │ └── test_alert_service.py └── docs/ └── openapi.yaml

我们重点看 `services/inventory_service.py` 的核心逻辑（已简化）： ```python from sqlalchemy import text from src.database import db_session from src.redis_client import redis_client from src.wecom_notifier import WecomNotifier class InventoryService: @staticmethod async def update_stock(product_id: str, delta: int): """更新库存并触发预警""" # 1. 原子性更新库存 async with db_session() as session: result = await session.execute( text("UPDATE products SET current_stock = current_stock + :delta WHERE product_id = :pid RETURNING current_stock, safety_stock"), {"delta": delta, "pid": product_id} ) row = result.fetchone() if not row: raise ValueError(f"Product {product_id} not found") current_stock, safety_stock = row # 2. 检查是否触发预警（关键：去重逻辑） if current_stock <= safety_stock: # 使用 Redis SETNX 防止重复预警（同一商品 1 小时内只预警一次） alert_key = f"alert:{product_id}:{int(time.time() // 3600)}" if await redis_client.set(alert_key, "1", ex=3600, nx=True): # 3. 写入预警流 await redis_client.xadd( "inventory_alerts", {"product_id": product_id, "current_stock": current_stock, "safety_stock": safety_stock} )

QA 角色生成的tests/test_inventory_service.py则精准打击：

import pytest from unittest.mock import AsyncMock, patch from src.services.inventory_service import InventoryService @pytest.mark.asyncio async def test_update_stock_triggers_alert(): """测试库存低于阈值时触发预警""" # Mock 数据库返回：更新后库存=45，安全阈值=50 mock_session = AsyncMock() mock_session.execute.return_value.fetchone.return_value = (45, 50) # Mock Redis SETNX 返回 True（首次预警） mock_redis = AsyncMock() mock_redis.set.return_value = True # Patch 依赖 with patch('src.services.inventory_service.db_session', mock_session), \ patch('src.services.inventory_service.redis_client', mock_redis): await InventoryService.update_stock("prod_001", -10) # 验证是否调用了 xadd mock_redis.xadd.assert_called_once() call_args = mock_redis.xadd.call_args assert call_args[0][0] == "inventory_alerts" assert "prod_001" in str(call_args[0][1]) @pytest.mark