staged-system-prompt

第10章 · 多 Agent 协作 · 配套项目 chapter10/staged-system-prompt

项目说明

实验 10-1：根据执行阶段决定系统提示词（Staged System Prompt）

《深入理解 AI Agent》配套实验代码。

实验目的

同一个 Coding Agent，在任务的不同执行阶段加载不同的系统提示词 + 不同的工具集， 从而在同一段对话里扮演不同角色、表现出不同的行为模式；同时让对话历史与任务状态在阶段间连续共享。

本实验用一个「Coding Agent」串起三个阶段：

阶段	角色	系统提示词强调	配套工具集	触发进入下一阶段的工具
1 需求澄清	需求分析师	只提问确认、不写代码	ask_clarifying_question / save_requirement / complete_requirements_analysis	complete_requirements_analysis → 阶段2
2 代码实现	软件工程师	按已确认需求写高质量 Python	write_file / read_file / execute_code / submit_for_review	submit_for_review → 阶段3
3 代码审查	代码审查员	批判性把关质量	run_linter / run_tests / analyze_complexity / request_revision / approve_code	request_revision → 回退阶段2；approve_code → 完成

架构

demo.py                入口：一条命令跑通三阶段（任务 = “写一个整理下载文件夹的 Python 脚本”）
agent.py               StagedAgent：阶段状态机 + 工具调用循环 + 跨阶段共享上下文 + 执行日志
tools.py               三套工具的 Schema 与真实实现（虚拟工作区 / 真实执行代码 / linter / 复杂度分析）
simulated_user.py      模拟用户：需求澄清阶段自动回答 Agent 的提问（预设答案），实现无人值守
config.py              从环境变量读取 API Key / base_url / model

关键设计：

• 共享上下文：StagedAgent.history 是一条贯穿始终的消息列表，切换阶段时只替换 system 提示词、只切换传给模型的 tools，历史消息（需求、代码、审查意见）全部保留。每次请求都是 [system(当前阶段)] + history。
• 阶段转换由工具调用触发：主循环识别到 complete_requirements_analysis / submit_for_review / request_revision / approve_code 这些「信号工具」被调用时，注入一条跨阶段「交接」消息并切换阶段。
• 回退机制：审查阶段发现问题时调用 request_revision(issues)，把问题清单退回实现阶段；设有 max_revisions 安全阀，避免无限循环烧 token。
• 真实执行：execute_code / run_tests 会把代码写入临时目录并用子进程真实运行；run_linter / analyze_complexity 基于 ast 做真实静态分析，不是假返回。

如何运行

pip install -r requirements.txt

# 配置（二选一）
export OPENAI_API_KEY=sk-...           # 方式 A：直接 export
cp env.example .env && vi .env         # 方式 B：写到 .env

python demo.py

# 离线查看三阶段配置（角色 / 系统提示词 / 工具集 / 转换信号），无需 API Key
python demo.py --list-stages

# 查看可选参数（不影响默认行为）
python demo.py --help

可选命令行参数（默认值与不加参数完全一致）：

参数	默认值	说明
--task	整理下载文件夹的任务	覆盖交给 Agent 的用户任务
--start-stage	requirements	从哪个阶段开始。选 implementation 会预置一份等价于需求澄清产物的已确认需求、直接从实现阶段起步，便于单独调试后两个阶段（review 依赖实现阶段的代码，不能作为起点）
--interactive	关	需求澄清阶段改由真人从标准输入回答 Agent 的提问（默认用 simulated_user.py 的模拟用户自动回答，可无人值守跑通全流程）
--max-revisions	3	审查阶段允许的最大回退次数，超过则强制结束演示
--model	环境变量 OPENAI_MODEL	覆盖使用的模型名
--list-stages	—	离线打印三阶段配置后退出，不调用任何 API（适合无 Key 时先看清机制）

可配环境变量（见 env.example）：OPENAI_API_KEY、OPENAI_BASE_URL（默认官方）、 OPENAI_MODEL（默认 gpt-5.6-luna，当前便宜旗舰）、OPENAI_TEMPERATURE（默认 0.3）。 也可切到兼容 OpenAI 协议的 Kimi / Doubao。

通用回退：优先用 OPENAI_API_KEY 直连 OpenAI；若未设置该变量但设了 OPENROUTER_API_KEY，则自动改走 OpenRouter，并把模型名映射到其命名空间 （gpt-5.6-luna → openai/gpt-5.6-luna）。提示：gpt-5.6 系列直连 OpenAI 需组织验证， 只填 OPENROUTER_API_KEY（不填 OPENAI_API_KEY）即可强制走 OpenRouter，更省事。

演示说明了什么问题

一次真实运行（gpt-5.6-luna）会看到：

1. 需求澄清阶段：Agent 表现为「不断提问」——主动追问处理哪些文件类型、是否递归、是否保留原名、移动还是复制、目标目录怎么定，并逐条 save_requirement。它完全不写代码。
2. 代码实现阶段：同一个 Agent 换了提示词后表现为「写代码」——write_file 产出 Python 脚本，execute_code 自测，然后 submit_for_review。
3. 代码审查阶段：Agent 表现为「批判审查」——依次跑 run_linter / run_tests / analyze_complexity，发现真实问题（如缺少模块 docstring、冒烟测试 FileNotFoundError）后 request_revision 退回实现阶段。
4. 实现阶段据问题清单重写并修复，再次提交；审查通过后 approve_code，任务完成。

也就是说：提示词 + 工具集随阶段切换，行为模式随之明显不同，而任务状态（需求、代码、审查意见）在阶段间始终连续共享。运行结束时会打印每个角色的「行为分布」统计，直观对比三个阶段的行为差异。

预期输出示例

以下是一次真实运行（python demo.py，gpt-5.6-luna，经 OpenRouter 路由）的节选，完整展示三阶段的行为切换 （本次运行需求澄清阶段反复追问了大小写规则、触发了 1 次审查回退，最终撞到总步数上限 安全阀结束，也是真实运行中常见的一种结局，详见下方「局限」）：

模型：openai/gpt-5.6-luna  | base_url：https://openrouter.ai/api/v1

======================================================================
进入阶段：requirements  |  角色：需求分析师  |  可用工具：['ask_clarifying_question', 'save_requirement', 'complete_requirements_analysis']
======================================================================
[需求分析师] 提问: 你希望脚本整理哪些文件类型？例如按扩展名整理所有常见文件，还是只处理图片、文档、视频等指定类型？
[需求分析师] 模拟用户回答: 按文件类型分类：图片(jpg/png/gif)、文档(pdf/doc/txt)、音频(mp3/wav)、视频(mp4/mov)、压缩包(zip/rar)，其余归到 Others。
[需求分析师] 记录需求: file_categories = 按文件类型分类：图片 jpg/png/gif；文档 pdf/doc/txt；音频 mp3/wav；视频 mp4/mov；压缩包 zip/rar；其他扩展名归入 Others。
[需求分析师] 提问: 整理时是否需要递归处理下载文件夹的子目录，还是只处理下载文件夹根目录中的文件？
[需求分析师] 记录需求: recursive = 不递归处理子目录，只整理下载文件夹当前层级的文件，忽略已有子文件夹。
[需求分析师] 记录需求: operation = 移动文件（move），不保留原位置的文件。
[需求分析师] 记录需求: name_conflicts = 保留原文件名；目标目录存在同名文件时，在扩展名前添加 _1、_2 等序号，避免覆盖。
...（模拟用户对「扩展名是否区分大小写」只会给出兜底回答，Agent 因此反复追问同一点，共提问 19 次）
[需求分析师] 完成需求分析 -> 转交实现: 已确认：脚本按指定扩展名分类并将其他文件归入 Others；仅处理下载目录当前层级；默认路径为 ~/Downloads、也可通过命令行参数指定；移动文件而非复制；扩展名不区分大小写；保留原文件名，冲突时追加 _1、_2 等避免覆盖。

======================================================================
进入阶段：implementation  |  角色：软件工程师  |  可用工具：['write_file', 'read_file', 'execute_code', 'submit_for_review']
======================================================================
[软件工程师] 写文件: 已写入文件 organize_downloads.py（4312 字符，135 行）
[软件工程师] 执行代码自测: from pathlib import Path from tempfile import TemporaryDirectory import organize ...
[软件工程师] 提交审查 -> 转交审查: organize_downloads.py

======================================================================
进入阶段：review  |  角色：代码审查员  |  可用工具：['run_linter', 'run_tests', 'analyze_complexity', 'request_revision', 'approve_code']
======================================================================
[代码审查员] run_linter: [linter] 通过：未发现问题。
[代码审查员] run_tests: [tests] 冒烟测试结果：FAIL
[代码审查员] analyze_complexity: [complexity] 函数数量=6，分支/循环语句=16，最大嵌套深度=3
[代码审查员] 审查不通过 -> 回退实现: 第1次退回：['冒烟测试失败：`from __future__ import annotations` 不在文件开头触发 SyntaxError，请移除该 future import 或改用兼容写法。']

======================================================================
进入阶段：implementation  |  角色：软件工程师  |  可用工具：['write_file', 'read_file', 'execute_code', 'submit_for_review']
======================================================================
[软件工程师] 写文件: 已写入文件 organize_downloads.py（4218 字符，133 行）
[软件工程师] 提交审查 -> 转交审查: organize_downloads.py

...（审查阶段再次检查，如此循环，直到 approve_code 或达到步数/回退上限）

======================================================================
执行小结
======================================================================
[需求分析师] 行为分布：提问×19, 模拟用户回答×19, 记录需求×7, 完成需求分析 -> 转交实现×1
[软件工程师] 行为分布：写文件×2, 执行代码自测×4, 读文件×1, 提交审查 -> 转交审查×2
[代码审查员] 行为分布：run_linter×1, run_tests×1, analyze_complexity×1, 审查不通过 -> 回退实现×1

已确认需求条数：7
产出文件：['organize_downloads.py']
审查回退次数：1

三段「行为分布」清楚对照出同一个 Agent 在三种提示词下的不同行为模式：需求分析师只问不写， 软件工程师只写不审，代码审查员只查不写。

更强的模型会让这套「阶段脚手架」变得多余吗？

一个常见直觉是：脚手架（这里指「按阶段切换系统提示词 + 工具集」的状态机）只是给弱模型用的拐杖， 换上更强的模型，它自然会「先澄清、再实现、后审查」地自我组织，脚手架随之失效。 用同一套代码、同一个任务、同一个模拟用户，本地各真实跑一次 gpt-4o-mini 与 gpt-5.6-luna 对照， 结论是否定的：

观察项	gpt-4o-mini（较弱）	gpt-5.6-luna（较强推理模型）
需求澄清提问次数	5（一问一点，问完即走）	21（反复纠缠「大写扩展名 / 无扩展名文件如何归类」这一个边角情形）
是否跑完三阶段拿到 approve_code	是（1 次回退后审查通过、任务完成）	否（撞到 40 步总步数安全阀被强制结束）
审查回退次数	1	1

（运行命令：MODEL=gpt-4o-mini python demo.py --model gpt-4o-mini；python demo.py --model gpt-5.6-luna。 后者经 OpenRouter 路由为 openai/gpt-5.6-luna。）

要点有两个：

1. 这套脚手架不是「可以关掉的拐杖」，而是结构性约束。 每个阶段只把本阶段的工具暴露给模型 （需求阶段根本没有 write_file，实现阶段根本没有 approve_code），角色分离是被工具门控强制出来的， 对强弱模型一视同仁——没有哪个模型能「自我组织」跳过或合并阶段。也正因如此，本实验里并不存在 一个「关掉脚手架、让强模型自由发挥」的基线可供严格对照。
2. 换上更强的模型并没有让脚手架变多余，反而更依赖它的安全阀。 gpt-5.6-luna 更「较真」， 坚持把一个模拟用户答不上来的边角规则问到底，而且聪明到每次都换一种问法， 恰好绕开了 SimulatedUser「同一问题问两次就催它进入下一阶段」的防重复机制， 于是在需求阶段空转了二十多步、把 40 步预算烧光——最后是 max_total_steps 这个脚手架安全阀替它兜的底； 较弱的 gpt-4o-mini 反而因为「问几个大方向就收手」顺利跑完了全程。

诚实的边界：gpt-5.6-luna 这次没跑完，很大程度上是被预设答案的 SimulatedUser（见「局限」）拖累的—— 它答不上强模型追问的边角问题，才诱发了空转；换真人回答（--interactive）或更聪明的模拟用户， 强模型大概率能更快收敛。所以这组数据不能推出「强模型在这个任务上更差」， 只能支持一个更窄、但对读者更有用的结论：阶段化提示词 + 工具门控是一种结构性脚手架， 它带来的角色分离与安全阀对强弱模型同样生效，不会因为模型变强就自动失效或变得多余。

局限

• 依赖所选模型的能力：默认用便宜旗舰 gpt-5.6-luna 控制演示成本。注意「更强的模型 = 更快收敛」 并不总成立：越较真的推理模型越容易在需求澄清阶段追问预设 SimulatedUser 答不上的边角问题而空转 （见上一节的真实对照），此时更依赖 max_total_steps / max_revisions 这两个脚手架安全阀兜底。
• 单一固定任务：内置演示任务是「整理下载文件夹」，虽然新增了 --task 参数可覆盖， 但 simulated_user.py 的预设问答是围绕这个任务场景设计的，换成差异很大的任务时模拟用户可能答不上点子上。
• 模拟用户是预设答案：SimulatedUser 按关键词匹配预设回答，不是真正理解语义的用户， 遇到 Agent 提出预设脚本之外的问题时会退化为兜底回答或催促进入下一阶段。
• 真实 LLM 有随机性：即使 temperature=0.3，不同次运行的提问顺序、代码实现细节、 审查是否通过、回退次数都可能不同；也可能像上面这次示例一样撞到 max_revisions 安全阀 强制结束，而不是拿到 approve_code。

源代码

agent.py

"""
StagedAgent：根据“执行阶段”切换系统提示词与工具集的 Coding Agent。

设计要点（对应实验 10-1 的 6 项要求）：
1) 三个阶段各有明确角色的系统提示词（STAGE_PROMPTS）。
2) 每个阶段配套独立工具集（tools.STAGE*_TOOLS）。
3) 阶段转换由“特定工具调用”触发（complete_requirements_analysis /
   submit_for_review / request_revision / approve_code）。
4) 上下文跨阶段连续：self.history 一直累加，切阶段时只换掉 system 提示词，
   历史消息（含之前的需求、代码、审查意见）全部保留。
5) 审查发现问题时 request_revision 让流程回退到实现阶段。
6) 每一步都写入 self.logs，最后能看出不同提示词导致的不同行为。
"""

from __future__ import annotations

import json
from typing import Callable, Dict, List, Optional

from openai import OpenAI

from config import Config
from simulated_user import SimulatedUser
import tools as T


# ----------------------------------------------------------------------------
# 三个阶段的系统提示词——每个阶段一个明确的“角色”。
# ----------------------------------------------------------------------------
STAGE_PROMPTS: Dict[str, str] = {
    "requirements": (
        "你是一名严谨的【需求分析师】。当前处于【需求澄清阶段】。\n"
        "你的唯一职责是把用户模糊的需求问清楚，绝对不要写任何代码。\n"
        "工作方式：\n"
        "1. 针对不明确的地方，用 ask_clarifying_question 逐个提问（一次问一个）。\n"
        "2. 每当用户确认一个点，就用 save_requirement 把它记录下来。\n"
        "3. 把关键问题（处理哪些文件类型、是否递归子目录、是否保留原文件名、"
        "移动还是复制、目标目录如何指定等）都澄清并记录后，"
        "调用 complete_requirements_analysis 结束本阶段。\n"
        "记住：你不实现、不设计代码，只负责澄清和记录需求。"
    ),
    "implementation": (
        "你是一名资深【软件工程师】。当前处于【代码实现阶段】。\n"
        "上文已经有需求分析师确认好的需求，请严格按这些需求实现，不要自行增删功能。\n"
        "工作方式：\n"
        "1. 用 write_file 写出高质量、可读、带模块与函数 docstring 的 Python 代码，"
        "避免裸 except，注意异常处理。\n"
        "2. 可用 execute_code 做自测，确认逻辑正确、能运行。\n"
        "3. 代码完成并自测通过后，调用 submit_for_review 提交审查。\n"
        "如果是被审查阶段退回来的（历史里会有问题清单），"
        "请针对每一条问题逐个修复后再重新 submit_for_review。"
    ),
    "review": (
        "你是一名挑剔的【代码审查员】。当前处于【代码审查阶段】。\n"
        "你的职责是批判性地审查实现阶段提交的代码，把关质量。\n"
        "工作方式：\n"
        "1. 依次用 run_linter、run_tests、analyze_complexity 客观检查代码。\n"
        "2. 认真解读检查结果。只要 linter 报出问题、或测试失败，"
        "就必须调用 request_revision，把问题清单退回实现阶段修复。\n"
        "3. 只有当检查干净、测试通过、复杂度合理时，才调用 approve_code 批准。\n"
        "标准要严格，不要放过 linter 报出的问题。"
    ),
}

# 阶段名 -> 该阶段暴露的工具集
STAGE_TOOLS = {
    "requirements": T.STAGE1_TOOLS,
    "implementation": T.STAGE2_TOOLS,
    "review": T.STAGE3_TOOLS,
}

# 会触发阶段转换的“信号工具”集合
T_TRANSITION_TOOLS = {
    T.COMPLETE_REQUIREMENTS,
    T.SUBMIT_FOR_REVIEW,
    T.REQUEST_REVISION,
    T.APPROVE_CODE,
}

# 阶段名 -> 角色中文名（打印用）
STAGE_ROLE = {
    "requirements": "需求分析师",
    "implementation": "软件工程师",
    "review": "代码审查员",
}

# 阶段的线性顺序与打印标题
STAGE_ORDER = ["requirements", "implementation", "review"]
STAGE_TITLE = {
    "requirements": "阶段1 需求澄清",
    "implementation": "阶段2 代码实现",
    "review": "阶段3 代码审查",
}

# 从 requirements 之后的阶段起步时，用来预置的“已确认需求”。
# 取值与 simulated_user.py 中模拟用户会给出的答案一致，因此这是对
# 需求澄清阶段产物的忠实复现，而非凭空捏造，方便单独调试后两个阶段。
CANONICAL_REQUIREMENTS: Dict[str, str] = {
    "file_types": "图片(jpg/png/gif)、文档(pdf/doc/txt)、音频(mp3/wav)、"
                  "视频(mp4/mov)、压缩包(zip/rar)，其余归 Others",
    "recursive": "不递归，只整理下载文件夹当前这一层，忽略已有子文件夹",
    "naming": "保留原文件名；同名冲突时加 _1/_2 后缀避免覆盖",
    "move_or_copy": "移动（move），整理完原位置不再保留这些文件",
    "destination": "在下载文件夹内按类别建子目录"
                   "（Images/Documents/Audio/Video/Archives/Others）；"
                   "根路径用命令行参数传入，默认 ~/Downloads",
}


def stage_overview() -> str:
    """离线打印三阶段总览（角色 / 系统提示词 / 工具集 / 转换信号），无需 API Key。"""
    lines: List[str] = ["阶段化系统提示词 · 三阶段角色切换总览（离线，无需 API Key）"]
    for stage in STAGE_ORDER:
        tool_names = [t["function"]["name"] for t in STAGE_TOOLS[stage]]
        transitions = [n for n in tool_names if n in T_TRANSITION_TOOLS]
        normal = [n for n in tool_names if n not in T_TRANSITION_TOOLS]
        lines.append("")
        lines.append("=" * 70)
        lines.append(f"{STAGE_TITLE[stage]}  |  角色：{STAGE_ROLE[stage]}")
        lines.append("=" * 70)
        lines.append("系统提示词：")
        lines.append(STAGE_PROMPTS[stage])
        lines.append(f"工作工具：{normal}")
        lines.append(f"阶段转换信号工具：{transitions}")
    lines.append("")
    lines.append("阶段转换关系：")
    lines.append("  requirements  --complete_requirements_analysis-->  implementation")
    lines.append("  implementation  --submit_for_review-->  review")
    lines.append("  review  --request_revision-->  implementation  （审查不通过，回退重写）")
    lines.append("  review  --approve_code-->  完成")
    return "\n".join(lines)


class StagedAgent:
    def __init__(
        self,
        max_revisions: int = 2,
        verbose: bool = True,
        interactive: bool = False,
    ) -> None:
        Config.validate()
        self.client = OpenAI(api_key=Config.API_KEY, base_url=Config.BASE_URL)
        self.model = Config.MODEL

        self.workspace = T.Workspace()
        self.sim_user = SimulatedUser()

        # 跨阶段共享的对话历史（不含 system 提示词，system 每阶段单独拼）
        self.history: List[dict] = []
        # 结构化执行日志：每条 = {stage, role, action, detail}
        self.logs: List[dict] = []

        self.stage = "requirements"
        self.revision_count = 0
        self.max_revisions = max_revisions
        self.verbose = verbose
        # interactive=True 时，需求澄清阶段的问题改由真人从标准输入回答；
        # 默认 False 走 SimulatedUser 预设答案，可无人值守跑通全流程。
        self.interactive = interactive

    # --- 日志与打印 ------------------------------------------------------
    def _log(self, action: str, detail: str) -> None:
        entry = {
            "stage": self.stage,
            "role": STAGE_ROLE[self.stage],
            "action": action,
            "detail": detail,
        }
        self.logs.append(entry)
        if self.verbose:
            print(f"[{STAGE_ROLE[self.stage]}] {action}: {detail}")

    def _banner(self, text: str) -> None:
        if self.verbose:
            print("\n" + "=" * 70)
            print(text)
            print("=" * 70)

    # --- 工具分发 --------------------------------------------------------
    def _dispatch_tool(self, name: str, args: dict) -> str:
        """执行普通工具（非阶段转换工具），返回给模型的工具结果字符串。"""
        if name == "ask_clarifying_question":
            question = args.get("question", "")
            self._log("提问", question)
            if self.interactive:
                answer = input(f"  [请回答需求分析师的问题] {question}\n  > ").strip()
                answer = answer or "没有特别要求，按常识处理即可。"
                self._log("用户回答", answer)
            else:
                answer = self.sim_user.answer(question)
                self._log("模拟用户回答", answer)
            return answer
        if name == "save_requirement":
            res = self.workspace.save_requirement(args.get("key", ""), args.get("value", ""))
            self._log("记录需求", f"{args.get('key')} = {args.get('value')}")
            return res
        if name == "write_file":
            # 模型偶尔对必填字段显式传 null：.get(..., "") 兜不住 None，统一归一化为空串。
            res = self.workspace.write_file(args.get("path") or "", args.get("content") or "")
            self._log("写文件", res)
            return res
        if name == "read_file":
            self._log("读文件", args.get("path", ""))
            return self.workspace.read_file(args.get("path", ""))
        if name == "execute_code":
            code = args.get("code") or ""
            self._log("执行代码自测", code[:80].replace("\n", " ") + " ...")
            return self.workspace.execute_code(code)
        if name == "run_linter":
            res = self.workspace.run_linter(args.get("file", ""))
            self._log("run_linter", res.splitlines()[0])
            return res
        if name == "run_tests":
            res = self.workspace.run_tests(args.get("file", ""))
            self._log("run_tests", res.splitlines()[0])
            return res
        if name == "analyze_complexity":
            res = self.workspace.analyze_complexity(args.get("file", ""))
            self._log("analyze_complexity", res)
            return res
        return f"未知工具：{name}"

    # --- 主流程 ----------------------------------------------------------
    def run(self, user_task: str, start_stage: str = "requirements") -> None:
        # 用户的初始任务，作为共享上下文的第一条消息
        self.history.append({"role": "user", "content": user_task})
        self._banner(f"用户任务：{user_task}")

        # 阶段状态机：一直跑到 approve_code 或超过安全上限
        max_total_steps = 40
        steps = 0
        done = False

        if start_stage == "implementation":
            # 跳过需求澄清，直接从实现阶段起步：预置一份等价于需求澄清产物的
            # 已确认需求，方便单独调试实现/审查两个阶段而不必每次重跑澄清对话。
            self._seed_requirements()
            self._enter_stage("implementation")
        else:
            self._enter_stage("requirements")

        while not done and steps < max_total_steps:
            steps += 1
            done = self._run_one_model_turn()

        if not done:
            self._banner("达到步数上限，演示结束（未收到 approve_code）。")

    def _seed_requirements(self) -> None:
        """从 requirements 之后的阶段起步时，预置一份已确认需求并注入交接消息。"""
        self.workspace.requirements = dict(CANONICAL_REQUIREMENTS)
        reqs = "\n".join(f"- {k}: {v}" for k, v in self.workspace.requirements.items())
        self.history.append({
            "role": "user",
            "content": (
                "【阶段交接】（--start-stage 跳过了需求澄清）需求分析师已确认如下需求，"
                "请据此实现：\n" + reqs
            ),
        })
        self._log_seed(reqs)

    def _log_seed(self, reqs: str) -> None:
        if self.verbose:
            self._banner("已预置需求（跳过需求澄清阶段）")
            print(reqs)

    def _enter_stage(self, stage: str) -> None:
        self.stage = stage
        self._banner(
            f"进入阶段：{stage}  |  角色：{STAGE_ROLE[stage]}  |  "
            f"可用工具：{[t['function']['name'] for t in STAGE_TOOLS[stage]]}"
        )

    def _run_one_model_turn(self) -> bool:
        """
        调一次模型；执行它请求的所有工具调用。
        返回 True 表示整个任务结束（approve_code）。
        阶段转换工具会切换 self.stage 并立刻返回，让下一轮用新提示词+新工具。
        """
        messages = [{"role": "system", "content": STAGE_PROMPTS[self.stage]}] + self.history
        try:
            response = self.client.chat.completions.create(
                model=self.model,
                messages=messages,
                tools=STAGE_TOOLS[self.stage],
                temperature=Config.TEMPERATURE,
            )
        except Exception as exc:  # noqa: BLE001
            # 部分推理模型（如 gpt-5.x）只接受默认温度，显式传 temperature 会 400。
            # 识别到与温度相关的报错时，去掉 temperature 重试一次。
            if "temperature" in str(exc).lower():
                response = self.client.chat.completions.create(
                    model=self.model,
                    messages=messages,
                    tools=STAGE_TOOLS[self.stage],
                )
            else:
                raise
        msg = response.choices[0].message

        # 把助手消息（可能含 tool_calls）加入共享历史
        assistant_entry: dict = {"role": "assistant", "content": msg.content or ""}
        if msg.tool_calls:
            assistant_entry["tool_calls"] = [
                {
                    "id": tc.id,
                    "type": "function",
                    "function": {"name": tc.function.name, "arguments": tc.function.arguments},
                }
                for tc in msg.tool_calls
            ]
        self.history.append(assistant_entry)

        if msg.content and msg.content.strip():
            self._log("思考/发言", msg.content.strip())

        # 没有工具调用：模型只是说话。提示它继续使用工具推进。
        if not msg.tool_calls:
            self.history.append({
                "role": "user",
                "content": "请使用当前阶段提供的工具继续推进任务。",
            })
            return False

        # 逐个处理本轮所有工具调用。
        # 注意：即使遇到“阶段转换工具”，也必须先把这条 assistant 消息里的
        # 每一个 tool_call 都回一条 tool 消息（OpenAI 协议强制要求），
        # 因此把真正的阶段切换/上下文交接推迟到所有工具都响应完之后再做。
        pending_transition: Optional[dict] = None
        for tc in msg.tool_calls:
            name = tc.function.name
            try:
                args = json.loads(tc.function.arguments or "{}")
            except json.JSONDecodeError:
                args = {}

            if name in T_TRANSITION_TOOLS:
                tool_result, descriptor = self._transition_result(name, args)
                # 只认第一个转换工具，其余转换工具只作普通响应
                if pending_transition is None:
                    pending_transition = descriptor
            else:
                tool_result = self._dispatch_tool(name, args)

            self.history.append({
                "role": "tool",
                "tool_call_id": tc.id,
                "content": tool_result,
            })

        # 所有 tool 消息都补齐后，再执行阶段切换与上下文交接
        if pending_transition is not None:
            return self._apply_transition(pending_transition)
        return False

    def _transition_result(self, name: str, args: dict):
        """
        计算阶段转换工具要回给模型的 tool 结果字符串，并返回一个 descriptor
        描述“稍后要执行的切换动作”。此函数只记录日志，不改动 history / 阶段。
        返回 (tool_result_str, descriptor_dict)。
        """
        if name == T.COMPLETE_REQUIREMENTS:
            summary = args.get("summary", "")
            self._log("完成需求分析 -> 转交实现", summary)
            return (
                f"需求分析完成：{summary}。即将进入代码实现阶段。",
                {"kind": "to_implementation"},
            )

        if name == T.SUBMIT_FOR_REVIEW:
            file = args.get("file", "")
            self._log("提交审查 -> 转交审查", file)
            return (
                f"已提交 {file}，即将进入审查阶段。",
                {"kind": "to_review", "file": file},
            )

        if name == T.REQUEST_REVISION:
            issues = args.get("issues")
            if issues is None:
                issues = []
            if isinstance(issues, str):
                issues = [issues]
            self.workspace.review_issues = list(issues)
            self.revision_count += 1
            self._log("审查不通过 -> 回退实现", f"第{self.revision_count}次退回：{issues}")
            return (
                "已把问题清单退回实现阶段。",
                {"kind": "request_revision", "issues": list(issues)},
            )

        if name == T.APPROVE_CODE:
            comment = args.get("comment", "")
            self._log("审查通过 -> 任务完成", comment)
            return (f"代码已批准：{comment}", {"kind": "approve"})

        return ("未知的转换工具。", {"kind": "noop"})

    def _apply_transition(self, descriptor: dict) -> bool:
        """
        真正执行阶段切换：注入跨阶段交接的 user 消息 + 切换 self.stage。
        返回 True 表示整个任务结束。
        """
        kind = descriptor["kind"]

        if kind == "to_implementation":
            reqs = "\n".join(f"- {k}: {v}" for k, v in self.workspace.requirements.items())
            self.history.append({
                "role": "user",
                "content": (
                    "【阶段交接】需求分析已完成。已确认的需求如下，请据此实现：\n"
                    + (reqs or "（无显式记录）")
                ),
            })
            self._enter_stage("implementation")
            return False

        if kind == "to_review":
            file = descriptor.get("file", "")
            self.history.append({
                "role": "user",
                "content": (
                    f"【阶段交接】实现阶段已提交文件 `{file}` 供审查。"
                    f"当前工作区文件：{list(self.workspace.files)}。请开始严格审查。"
                ),
            })
            self._enter_stage("review")
            return False

        if kind == "request_revision":
            # 安全阀：回退次数过多则强制收尾，避免无限循环烧 token
            if self.revision_count > self.max_revisions:
                self._log("回退次数达上限", "强制结束演示")
                self.history.append({
                    "role": "user",
                    "content": "【系统】回退次数已达上限，演示到此结束。",
                })
                return True
            issue_text = "\n".join(f"- {x}" for x in (descriptor.get("issues") or []))
            self.history.append({
                "role": "user",
                "content": (
                    "【阶段交接·回退】审查未通过，需修复以下问题后重新提交：\n"
                    + issue_text
                ),
            })
            self._enter_stage("implementation")
            return False

        if kind == "approve":
            return True

        return False

    # --- 演示后打印小结 --------------------------------------------------
    def print_summary(self) -> None:
        self._banner("执行小结")
        # 统计每个阶段的动作，展示“不同提示词 -> 不同行为模式”
        by_stage: Dict[str, List[str]] = {}
        for e in self.logs:
            by_stage.setdefault(e["role"], []).append(e["action"])
        for role, actions in by_stage.items():
            from collections import Counter
            counts = Counter(actions)
            summary = ", ".join(f"{a}×{n}" for a, n in counts.items())
            print(f"[{role}] 行为分布：{summary}")

        print(f"\n已确认需求条数：{len(self.workspace.requirements)}")
        print(f"产出文件：{list(self.workspace.files)}")
        print(f"审查回退次数：{self.revision_count}")

config.py

"""
配置模块：集中读取环境变量。

实验 10-1 使用 OpenAI 官方 SDK，所有可调项都通过环境变量注入，
方便切换到兼容 OpenAI 协议的其他厂商（Kimi / Doubao 等）。
"""

import os

try:
    # 允许把配置写在 .env 里（可选依赖）
    from dotenv import load_dotenv

    load_dotenv()
except Exception:  # pragma: no cover - dotenv 不是硬性依赖
    pass


def _to_openrouter_model(model: str) -> str:
    """把模型名映射到 OpenRouter 命名空间（用于无 OPENAI_API_KEY 的回退路径）。"""
    if "/" in model:
        return model                      # 已是 OpenRouter 命名空间，原样使用
    if model.startswith("gpt-"):
        return "openai/" + model          # gpt-* -> openai/gpt-*
    if model.startswith("claude-"):
        return "anthropic/claude-opus-4.8"
    return "openai/gpt-5.6-luna"          # 兜底：当前便宜旗舰


class Config:
    """运行时配置。"""

    # 必填：OpenAI API Key（本实验默认用 OPENAI_API_KEY）
    API_KEY: str = os.environ.get("OPENAI_API_KEY", "")

    # 可选：兼容 OpenAI 协议的 base_url，默认官方地址
    BASE_URL: str = os.environ.get("OPENAI_BASE_URL", "https://api.openai.com/v1")

    # 可选：模型名，默认用当前便宜旗舰 gpt-5.6-luna 控制演示成本
    MODEL: str = os.environ.get("OPENAI_MODEL", "gpt-5.6-luna")

    # 采样温度，稍低一些让行为更稳定可复现
    TEMPERATURE: float = float(os.environ.get("OPENAI_TEMPERATURE", "0.3"))

    @classmethod
    def validate(cls) -> None:
        """校验并按需应用通用回退。

        0) gpt-5.x 系列直连 OpenAI 需组织验证（且这类推理模型只接受默认温度），
           因此只要设置了 OPENROUTER_API_KEY，即便同时有 OPENAI_API_KEY，也优先走
           OpenRouter，省去组织验证的麻烦；
        1) 否则有 OPENAI_API_KEY -> 直连 OpenAI（尊重 OPENAI_BASE_URL）；
        2) 否则有 OPENROUTER_API_KEY -> 改走 OpenRouter，并映射模型名；
        3) 都没有则报清晰错误。
        """
        or_key = os.environ.get("OPENROUTER_API_KEY", "")
        # gpt-5.x（含 gpt-5.6-luna 等推理旗舰）优先走 OpenRouter，规避组织验证。
        needs_openrouter = cls.MODEL.startswith("gpt-5") and "/" not in cls.MODEL
        if or_key and (needs_openrouter or not cls.API_KEY):
            cls.API_KEY = or_key
            cls.BASE_URL = "https://openrouter.ai/api/v1"
            cls.MODEL = _to_openrouter_model(cls.MODEL)
            return
        if cls.API_KEY:
            return
        raise SystemExit(
            "错误：未检测到 OPENAI_API_KEY 或 OPENROUTER_API_KEY 环境变量。\n"
            "请先 `export OPENAI_API_KEY=...`（或 OPENROUTER_API_KEY），"
            "或复制 env.example 为 .env 并填写。"
        )

demo.py

"""
实验 10-1 演示入口：一条命令跑通“需求澄清 -> 代码实现 -> 代码审查”三阶段。

    python demo.py                 # 默认任务、默认模型、最多 3 次审查回退
    python demo.py --list-stages   # 离线查看三阶段配置（无需 API Key）
    python demo.py --help          # 查看全部可选参数

演示任务：用户想要“写一个整理下载文件夹的 Python 脚本”。
需求本身模糊，因此需求澄清阶段的 Agent 会主动提问，由模拟用户自动回答；
之后进入实现阶段写代码、审查阶段严格把关（可能回退重写）。
"""

import argparse

from agent import StagedAgent, stage_overview
from config import Config


USER_TASK = "帮我写一个整理下载文件夹的 Python 脚本。"
DEFAULT_MAX_REVISIONS = 3


def parse_args() -> argparse.Namespace:
    """解析命令行参数。不传任何参数时，行为与原始固定脚本完全一致。"""
    parser = argparse.ArgumentParser(
        prog="demo.py",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        description=(
            "实验 10-1：阶段化系统提示词（需求澄清 -> 代码实现 -> 代码审查）演示。\n"
            "同一个 Agent 在三个阶段切换系统提示词与工具集，扮演不同角色，\n"
            "而对话历史与任务状态跨阶段连续共享。不加参数运行即为默认演示。"
        ),
        epilog=(
            "示例：\n"
            "  python demo.py                        默认任务，跑通三阶段\n"
            "  python demo.py --list-stages          离线查看三阶段配置（无需 API Key）\n"
            "  python demo.py --start-stage implementation   跳过需求澄清，从实现阶段起步\n"
            "  python demo.py --interactive          需求澄清阶段由你本人回答提问\n"
            "  python demo.py --model gpt-5.6-luna --task '写一个批量重命名图片的脚本'"
        ),
    )
    parser.add_argument(
        "--task",
        default=USER_TASK,
        help=f"交给 Agent 的用户任务（默认：{USER_TASK!r}）",
    )
    parser.add_argument(
        "--start-stage",
        choices=["requirements", "implementation"],
        default="requirements",
        help=(
            "从哪个阶段开始（默认：requirements）。选 implementation 会预置一份"
            "等价于需求澄清产物的已确认需求、直接从实现阶段起步，便于单独调试后两个阶段。"
            "（review 阶段依赖实现阶段产出的代码，无法作为起点。）"
        ),
    )
    parser.add_argument(
        "--interactive",
        action="store_true",
        help="需求澄清阶段改由真人从标准输入回答 Agent 的提问（默认用模拟用户自动回答）。",
    )
    parser.add_argument(
        "--max-revisions",
        type=int,
        default=DEFAULT_MAX_REVISIONS,
        help=f"审查阶段允许的最大回退次数，超过则强制结束演示（默认：{DEFAULT_MAX_REVISIONS}）",
    )
    parser.add_argument(
        "--model",
        default=None,
        help=f"覆盖 OPENAI_MODEL 环境变量指定的模型名（默认：使用环境变量，当前为 {Config.MODEL!r}）",
    )
    parser.add_argument(
        "--list-stages",
        action="store_true",
        help="离线打印三阶段（角色/系统提示词/工具集/转换信号）后退出，不调用任何 API。",
    )
    return parser.parse_args()


def main() -> None:
    args = parse_args()

    # 离线路径：只展示三阶段配置，不需要 API Key，也不发起任何请求。
    if args.list_stages:
        print(stage_overview())
        return

    if args.model:
        Config.MODEL = args.model

    # 先解析 provider（含 OpenRouter 回退），确保打印出的模型/端点是最终生效值。
    Config.validate()
    print("模型：%s  | base_url：%s" % (Config.MODEL, Config.BASE_URL))
    agent = StagedAgent(
        max_revisions=args.max_revisions,
        verbose=True,
        interactive=args.interactive,
    )
    agent.run(args.task, start_stage=args.start_stage)
    agent.print_summary()

    # 打印最终产出的主文件，方便肉眼确认实现阶段真的写了代码
    if agent.workspace.files:
        print("\n" + "=" * 70)
        print("最终产出文件内容：")
        print("=" * 70)
        for path, content in agent.workspace.files.items():
            print(f"\n--- {path} ---\n{content}")


if __name__ == "__main__":
    main()

simulated_user.py

"""
模拟用户：在需求澄清阶段自动回答 Agent 的提问，实现无人值守跑通三阶段。

真实产品里，ask_clarifying_question 会把问题抛给真人；这里用一组预设答案，
按关键词“打分匹配”Agent 的问题并给出回答（命中关键词最多者胜）。
另外内置一个防重复机制：如果同一个问题被反复追问，就明确告诉 Agent
“已经回答过、没别的要求了，请开始实现”，避免澄清阶段陷入死循环。
"""

from typing import Dict, List, Tuple


class SimulatedUser:
    def __init__(self) -> None:
        # (关键词列表, 预设答案)。命中关键词越多，越优先采用。
        self.playbook: List[Tuple[List[str], str]] = [
            (["类型", "格式", "扩展名", "分类", "type", "kind"],
             "按文件类型分类：图片(jpg/png/gif)、文档(pdf/doc/txt)、"
             "音频(mp3/wav)、视频(mp4/mov)、压缩包(zip/rar)，其余归到 Others。"),
            (["递归", "子目录", "子文件夹", "recursive", "subfolder"],
             "不需要递归，只整理下载文件夹当前这一层，忽略里面已有的子文件夹。"),
            (["原文件名", "重命名", "保留名", "同名", "冲突", "重复", "覆盖", "rename", "conflict"],
             "保留原文件名；如果同名文件已存在，就在文件名后加 _1、_2 避免覆盖。"),
            (["移动", "复制", "剪切", "move", "copy"],
             "用移动（move）而不是复制，整理完原位置就不再保留这些文件。"),
            (["目标", "目的地", "存到", "保存到", "路径", "位置", "哪个文件夹", "destination", "location", "path"],
             "不用单独指定目标目录：就在下载文件夹内部按类别创建子文件夹"
             "（Images/Documents/Audio/Video/Archives/Others），把文件移动进对应子文件夹即可。"
             "下载文件夹本身的路径用命令行参数传入，不传则默认 ~/Downloads。"),
            (["日期", "时间", "date", "time"],
             "不用按日期分，只按类型分类就行。"),
            (["确认", "开始", "还有", "其他", "别的", "补充", "proceed", "anything else"],
             "没有其他要求了，需求就这些，可以开始实现。"),
        ]
        self.default_answer = "按常识处理即可，不用太复杂，保持脚本简单可读。"
        self.qa_log: List[Tuple[str, str]] = []
        self._asked_count: Dict[str, int] = {}

    def _match(self, q: str) -> str:
        best_reply, best_score = self.default_answer, 0
        for keywords, reply in self.playbook:
            score = sum(1 for kw in keywords if kw.lower() in q)
            if score > best_score:
                best_reply, best_score = reply, score
        return best_reply

    def answer(self, question: str) -> str:
        norm = "".join(question.lower().split())
        self._asked_count[norm] = self._asked_count.get(norm, 0) + 1

        # 防重复：同一问题问到第 2 次，就催促 Agent 结束澄清、进入实现
        if self._asked_count[norm] >= 2:
            reply = (
                "这个问题我刚才已经回答过了，没有别的要求了。"
                "需求已经足够清楚，请直接调用 complete_requirements_analysis 进入实现阶段。"
            )
        else:
            reply = self._match(question)

        self.qa_log.append((question, reply))
        return reply

test_null_issues.py

"""Null issues on request_revision must behave like empty list."""
from unittest.mock import MagicMock

from agent import StagedAgent
import tools as T


def test_null_issues_like_empty():
    agent = StagedAgent.__new__(StagedAgent)
    agent.workspace = T.Workspace()
    agent.revision_count = 0
    agent.logs = []
    agent._log = lambda *a, **k: None
    msg, desc = agent._transition_result(T.REQUEST_REVISION, {"issues": None})
    assert agent.workspace.review_issues == []
    assert desc["issues"] == []
    assert "退回" in msg or "问题" in msg

test_null_tool_args.py

"""回归测试：模型对必填工具参数显式传 null 时，工具分发不应崩溃。

此前 write_file 的 content 为 null 时 len(None) -> TypeError，
execute_code 的 code 为 null 时在日志切片处就 None[:80] -> TypeError；
现在分发处统一把 None 归一化为空串。不实例化 StagedAgent.__init__
（避免 Config 校验 API Key），只测 _dispatch_tool 本身。
"""

import sys
from pathlib import Path
from types import ModuleType

sys.path.insert(0, str(Path(__file__).parent))

try:
    import openai  # noqa: F401
except ImportError:
    sys.modules["openai"] = ModuleType("openai")
    sys.modules["openai"].OpenAI = object

import tools as T
from agent import StagedAgent
from simulated_user import SimulatedUser


def _bare_agent():
    ag = object.__new__(StagedAgent)
    ag.workspace = T.Workspace()
    ag.logs = []
    ag.verbose = False
    ag.stage = "implementation"
    ag.interactive = False
    ag.sim_user = SimulatedUser()
    return ag


def test_write_file_null_content_coerced_to_empty_string():
    ag = _bare_agent()
    res = ag._dispatch_tool("write_file", {"path": "a.py", "content": None})
    assert "已写入文件 a.py" in res
    assert ag.workspace.files["a.py"] == ""


def test_execute_code_null_code_coerced_to_empty_string():
    ag = _bare_agent()
    res = ag._dispatch_tool("execute_code", {"code": None})
    assert isinstance(res, str)
    assert "退出码: 0" in res


def test_write_file_normal_content_unchanged():
    ag = _bare_agent()
    res = ag._dispatch_tool(
        "write_file", {"path": "a.py", "content": "print('hi')\n"})
    assert "已写入文件 a.py" in res
    assert ag.workspace.files["a.py"] == "print('hi')\n"

tools.py

"""
工具实现与工具集定义。

本文件包含两部分：
1. Workspace：一个进程内的“虚拟工作区”，负责保存需求、文件内容，
   并提供真实的代码执行 / 语法检查 / 复杂度分析能力。
2. 三个阶段各自的工具 JSON Schema（供 OpenAI function calling 使用）。

关键点：不同阶段暴露给模型的工具集是不同的，这是“阶段化系统提示词”实验
的核心之一——提示词换了角色，工具也随之切换。
"""

from __future__ import annotations

import ast
import os
import subprocess
import sys
import tempfile
from typing import Dict, List


# ----------------------------------------------------------------------------
# 触发阶段转换的“信号工具”名字。Agent 主循环看到这些工具被调用就切换阶段。
# ----------------------------------------------------------------------------
COMPLETE_REQUIREMENTS = "complete_requirements_analysis"  # 阶段1 -> 阶段2
SUBMIT_FOR_REVIEW = "submit_for_review"                   # 阶段2 -> 阶段3
REQUEST_REVISION = "request_revision"                     # 阶段3 -> 阶段2（回退）
APPROVE_CODE = "approve_code"                             # 阶段3 -> 完成


class Workspace:
    """跨阶段共享的任务状态（需求、文件、审查意见）。"""

    def __init__(self) -> None:
        # 阶段1 收集到的、已确认的需求（key -> value）
        self.requirements: Dict[str, str] = {}
        # 阶段2 写出的“文件系统”（path -> content）
        self.files: Dict[str, str] = {}
        # 阶段3 退回时记录的问题清单，供阶段2 修复时参考
        self.review_issues: List[str] = []

    # --- 阶段1：需求分析师的工具实现 -------------------------------------
    def save_requirement(self, key: str, value: str) -> str:
        self.requirements[key] = value
        return f"已记录需求 [{key}] = {value}"

    # --- 阶段2：软件工程师的工具实现 -------------------------------------
    def write_file(self, path: str, content: str) -> str:
        self.files[path] = content
        return f"已写入文件 {path}（{len(content)} 字符，{content.count(chr(10)) + 1} 行）"

    def read_file(self, path: str) -> str:
        if path not in self.files:
            return f"错误：文件 {path} 不存在。当前文件列表：{list(self.files) or '空'}"
        return self.files[path]

    def execute_code(self, code: str) -> str:
        """在临时目录里真实执行一段 Python，返回 stdout/stderr（带超时）。"""
        return _run_python_source(code)

    # --- 阶段3：代码审查员的工具实现 -------------------------------------
    def run_linter(self, path: str) -> str:
        """轻量静态检查：语法编译 + 常见坏味道，不引入额外依赖。"""
        if path not in self.files:
            return f"错误：文件 {path} 不存在。"
        source = self.files[path]
        problems: List[str] = []

        # 1) 语法能否编译
        try:
            tree = ast.parse(source)
        except SyntaxError as exc:
            return f"[linter] 语法错误：第 {exc.lineno} 行 {exc.msg}"

        # 2) 逐行的风格问题（阈值定得“严格但可达标”，方便演示先退回再通过）
        for i, line in enumerate(source.splitlines(), start=1):
            if len(line) > 120:
                problems.append(f"L{i}: 行超过 120 字符（{len(line)}），请折行或精简")
            if line.rstrip() != line:
                problems.append(f"L{i}: 行尾有多余空白")
            if "\t" in line:
                problems.append(f"L{i}: 使用了 Tab 缩进，建议用空格")

        # 3) 基于 AST 的问题：缺少模块 docstring、裸 except
        if not ast.get_docstring(tree):
            problems.append("模块缺少文件级 docstring（请在文件开头加一段三引号说明）")
        for node in ast.walk(tree):
            if isinstance(node, ast.ExceptHandler) and node.type is None:
                problems.append(f"L{node.lineno}: 使用了裸 except，建议捕获具体异常")

        if not problems:
            return "[linter] 通过：未发现问题。"
        return "[linter] 发现 %d 个问题：\n- %s" % (len(problems), "\n- ".join(problems))

    def run_tests(self, path: str) -> str:
        """冒烟测试：把文件跑起来，验证 import / 主流程不崩溃。"""
        if path not in self.files:
            return f"错误：文件 {path} 不存在。"
        # 造一个假的“下载文件夹”，让整理脚本有东西可整理
        harness = (
            "import os, tempfile, runpy, sys\n"
            "d = tempfile.mkdtemp()\n"
            "for name in ['a.jpg','b.pdf','c.txt','d.mp3','readme']:\n"
            "    open(os.path.join(d, name), 'w').close()\n"
            "sys.argv = ['script', d]\n"
            "print('SMOKE_TEST target dir:', d)\n"
            + self.files[path]
        )
        result = _run_python_source(harness)
        ok = "Traceback" not in result and "Error" not in result
        verdict = "PASS" if ok else "FAIL"
        return f"[tests] 冒烟测试结果：{verdict}\n{result}"

    def analyze_complexity(self, path: str) -> str:
        """用 AST 估算复杂度：函数数量、最大分支数、最大嵌套深度。"""
        if path not in self.files:
            return f"错误：文件 {path} 不存在。"
        try:
            tree = ast.parse(self.files[path])
        except SyntaxError as exc:
            return f"[complexity] 无法解析：{exc.msg}"

        funcs = [n for n in ast.walk(tree)
                 if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef))]
        branch_types = (ast.If, ast.For, ast.While, ast.Try, ast.With)
        total_branches = sum(1 for n in ast.walk(tree) if isinstance(n, branch_types))

        def depth(node: ast.AST, level: int = 0) -> int:
            best = level
            for child in ast.iter_child_nodes(node):
                inc = 1 if isinstance(child, branch_types) else 0
                best = max(best, depth(child, level + inc))
            return best

        return (
            "[complexity] 函数数量=%d，分支/循环语句=%d，最大嵌套深度=%d"
            % (len(funcs), total_branches, depth(tree))
        )


def _run_python_source(source: str, timeout: int = 10) -> str:
    """把源码写到临时文件并用子进程执行，返回合并后的输出。"""
    with tempfile.TemporaryDirectory() as tmp:
        script = os.path.join(tmp, "snippet.py")
        with open(script, "w", encoding="utf-8") as fh:
            fh.write(source)
        try:
            proc = subprocess.run(
                [sys.executable, script],
                capture_output=True,
                text=True,
                timeout=timeout,
                cwd=tmp,
            )
        except subprocess.TimeoutExpired:
            return f"执行超时（>{timeout}s）"
        out = (proc.stdout or "").strip()
        err = (proc.stderr or "").strip()
        parts = []
        if out:
            parts.append("stdout:\n" + out)
        if err:
            parts.append("stderr:\n" + err)
        parts.append(f"退出码: {proc.returncode}")
        return "\n".join(parts)


# ----------------------------------------------------------------------------
# 各阶段的工具 Schema（OpenAI tools 格式）。每个阶段只暴露自己那套工具。
# ----------------------------------------------------------------------------

def _tool(name: str, description: str, properties: dict, required: list) -> dict:
    return {
        "type": "function",
        "function": {
            "name": name,
            "description": description,
            "parameters": {
                "type": "object",
                "properties": properties,
                "required": required,
            },
        },
    }


STAGE1_TOOLS = [
    _tool(
        "ask_clarifying_question",
        "向用户提出一个澄清需求的问题，用户会回答。需求不明确时必须先问清楚。",
        {"question": {"type": "string", "description": "要问用户的问题"}},
        ["question"],
    ),
    _tool(
        "save_requirement",
        "把一条已经确认的需求记录到需求文档中，供后续实现阶段使用。",
        {
            "key": {"type": "string", "description": "需求项名称，如 file_types"},
            "value": {"type": "string", "description": "需求项取值/描述"},
        },
        ["key", "value"],
    ),
    _tool(
        COMPLETE_REQUIREMENTS,
        "当所有关键需求都已澄清并记录后调用，结束需求分析阶段，进入代码实现阶段。",
        {"summary": {"type": "string", "description": "对已确认需求的一句话总结"}},
        ["summary"],
    ),
]

STAGE2_TOOLS = [
    _tool(
        "write_file",
        "写入（或覆盖）一个文件的完整内容。",
        {
            "path": {"type": "string", "description": "文件路径，如 organize_downloads.py"},
            "content": {"type": "string", "description": "文件的完整内容"},
        },
        ["path", "content"],
    ),
    _tool(
        "read_file",
        "读取一个已写入文件的内容。",
        {"path": {"type": "string", "description": "文件路径"}},
        ["path"],
    ),
    _tool(
        "execute_code",
        "执行一段 Python 代码用于自测/验证，返回标准输出与错误。",
        {"code": {"type": "string", "description": "要执行的 Python 代码"}},
        ["code"],
    ),
    _tool(
        SUBMIT_FOR_REVIEW,
        "当代码实现完成且自测通过后调用，提交给代码审查员，进入审查阶段。",
        {"file": {"type": "string", "description": "要提交审查的主文件路径"}},
        ["file"],
    ),
]

STAGE3_TOOLS = [
    _tool(
        "run_linter",
        "对文件运行静态检查，返回代码风格/规范问题。",
        {"file": {"type": "string", "description": "文件路径"}},
        ["file"],
    ),
    _tool(
        "run_tests",
        "对文件运行冒烟测试，验证能否正常运行。",
        {"file": {"type": "string", "description": "文件路径"}},
        ["file"],
    ),
    _tool(
        "analyze_complexity",
        "分析文件的代码复杂度（函数数、分支数、嵌套深度）。",
        {"file": {"type": "string", "description": "文件路径"}},
        ["file"],
    ),
    _tool(
        REQUEST_REVISION,
        "当审查发现必须修复的问题时调用，把代码退回实现阶段并附上问题清单。",
        {
            "issues": {
                "type": "array",
                "items": {"type": "string"},
                "description": "需要修复的问题列表",
            }
        },
        ["issues"],
    ),
    _tool(
        APPROVE_CODE,
        "当代码通过所有审查、质量达标时调用，批准代码，任务完成。",
        {"comment": {"type": "string", "description": "审查通过的简短评语"}},
        ["comment"],
    ),
]