Reflection：Agent 如何审视自己的输出并纠正错误

任务：分析某产品的竞争态势，给出市场定位建议。

Step 1 - Thought: 先搜索主要竞品信息
Action: search_web("竞品A 产品特性")
Observation: 返回了竞品A的官方宣传内容

Step 2 - Thought: 再搜索竞品B
Action: search_web("竞品B 产品特性")
Observation: 返回了竞品B的官方宣传内容

Step 3 - Thought: 信息足够了，开始分析
Answer: 基于以上信息，建议定位为...

import openai
import json

client = openai.OpenAI()

async def reflect_on_output(
    task: str,
    execution_trace: list[dict],
    output: str,
    criteria: list[str] | None = None
) -> dict:
    """
    对 Agent 输出做批判性审查。
    返回：问题列表、严重程度、是否需要修正、修正建议。
    """

    default_criteria = [
        "输出是否完整回答了原始任务",
        "结论是否有充分的信息支撑（没有无依据的推断）",
        "信息来源是否可靠、是否存在明显偏差",
        "是否遗漏了任务中明确要求的内容",
        "输出格式是否符合任务要求",
    ]
    active_criteria = criteria or default_criteria

    trace_summary = "\n".join([
        f"[{s['type']}] {str(s['content'])[:200]}"
        for s in execution_trace
    ])

    response = await client.chat.completions.create(
        model="gpt-4o",
        messages=[{
            "role": "user",
            "content": f"""
你是一个严格的质量审查员。请对以下 Agent 的执行结果做批判性评估。

原始任务：
{task}

执行过程摘要：
{trace_summary}

Agent 的输出：
{output}

请根据以下标准逐一评估：
{chr(10).join(f"{i+1}. {c}" for i, c in enumerate(active_criteria))}

对每个发现的问题，说明：
- 问题描述
- 严重程度（致命 / 重要 / 轻微）
- 修正建议

输出 JSON 格式：
{{
  "issues": [
    {{
      "criterion": "对应的评估标准",
      "description": "问题描述",
      "severity": "致命|重要|轻微",
      "suggestion": "修正建议"
    }}
  ],
  "overall_quality": "高|中|低",
  "needs_revision": true/false,
  "revision_scope": "完全重做|局部修正|无需修正"
}}
"""
        }],
        response_format={"type": "json_object"}
    )

    return json.loads(response.choices[0].message.content)


async def execute_with_reflection(
    task: str,
    agent_fn,
    tools: list,
    max_reflection_rounds: int = 2
) -> str:
    """
    执行任务 + 输出后审查的完整流程。
    """
    output, trace = await agent_fn(task, tools)

    for round_idx in range(max_reflection_rounds):
        review = await reflect_on_output(task, trace, output)

        # 没有问题或只有轻微问题，直接返回
        fatal_issues = [i for i in review["issues"] if i["severity"] == "致命"]
        important_issues = [i for i in review["issues"] if i["severity"] == "重要"]

        if not fatal_issues and not important_issues:
            break

        # 有致命问题，重新执行
        if fatal_issues:
            revision_prompt = build_revision_prompt(task, output, fatal_issues)
            output, trace = await agent_fn(revision_prompt, tools)

        # 只有重要问题，局部修正
        elif important_issues:
            output = await revise_output(task, output, important_issues)
            break

    return output


def build_revision_prompt(task: str, output: str, issues: list[dict]) -> str:
    issue_text = "\n".join([
        f"- {i['description']}（建议：{i['suggestion']}）"
        for i in issues
    ])
    return f"""
原始任务：{task}

上一次执行的输出存在以下严重问题：
{issue_text}

请重新执行任务，特别注意避免上述问题。
"""

async def reflective_checkpoint(
    plan: "ExecutionPlan",
    completed_step: "PlanStep",
    recent_trace: list[dict]
) -> dict:
    """
    在每个计划步骤完成后触发反思，检查是否走在正确的轨道上。
    """
    response = await client.chat.completions.create(
        model="gpt-4o",
        messages=[{
            "role": "user",
            "content": f"""
任务目标：{plan.goal}
完成标准：{plan.definition_of_done}

刚完成的步骤：{completed_step.description}
步骤结果：{completed_step.result_summary}

最近的执行轨迹：
{format_trace(recent_trace)}

请从以下角度进行反思：

1. 方向检查：当前的执行方向是否仍然指向原始目标？
   有没有被某个工具结果或中间发现带偏？

2. 质量检查：刚完成的步骤，结果的质量如何？
   有没有信息偏差、遗漏或不可靠的来源？

3. 假设检查：执行过程中有没有隐含的假设？
   这些假设是否经过了验证？

4. 风险预判：基于当前进展，后续步骤有什么潜在的风险？

输出 JSON：
{{
  "on_track": true/false,
  "direction_issues": "方向问题描述，无则为 null",
  "quality_issues": "质量问题描述，无则为 null",
  "unverified_assumptions": ["假设1", "假设2"],
  "upcoming_risks": ["风险1", "风险2"],
  "recommended_action": "继续|调整方向|暂停确认|回退重做",
  "adjustment_notes": "如果需要调整，具体说明"
}}
"""
        }],
        response_format={"type": "json_object"}
    )

    return json.loads(response.choices[0].message.content)

async def multi_agent_critique(
    task: str,
    output: str,
    execution_trace: list[dict],
    num_critics: int = 2
) -> dict:
    """
    多个批评者 Agent 从不同角度审查输出，综合意见后给出裁决。
    """
    critic_perspectives = [
        {
            "role": "事实核查员",
            "focus": "检查输出中的每一个事实性陈述是否有来源支撑，是否存在无依据的推断",
        },
        {
            "role": "逻辑审查员",
            "focus": "检查推理链是否有跳跃，结论是否真的从前提中得出，有无自相矛盾",
        },
        {
            "role": "完整性审查员",
            "focus": "检查任务要求的所有方面是否都得到了回答，有无遗漏",
        },
    ]

    # 并行调用多个批评者
    import asyncio
    critic_tasks = [
        call_critic(task, output, perspective)
        for perspective in critic_perspectives[:num_critics]
    ]
    critiques = await asyncio.gather(*critic_tasks)

    # 综合裁决
    combined = await synthesize_critiques(task, output, critiques)
    return combined


async def call_critic(task: str, output: str, perspective: dict) -> dict:
    response = await client.chat.completions.create(
        model="gpt-4o",
        messages=[{
            "role": "system",
            "content": f"你是一个{perspective['role']}。{perspective['focus']}"
        }, {
            "role": "user",
            "content": f"任务：{task}\n\n待审查的输出：\n{output}"
        }],
        response_format={"type": "json_object"}
    )
    return json.loads(response.choices[0].message.content)

# 这种写法几乎没有用
"请检查你的输出是否有问题，如果有，请修正。"

RETRIEVAL_CRITERIA = [
    "所有事实性陈述是否都能在工具返回的内容中找到来源",
    "是否混入了工具未返回、来自训练知识的内容",
    "搜索结果是否存在明显的时效性问题（内容已过期）",
    "是否只引用了某一方的观点，忽略了其他视角",
]

CODE_CRITERIA = [
    "代码是否在沙箱里实际执行过并验证了输出",
    "有没有假设了不存在的变量、函数或导入",
    "错误处理是否完整（边界情况、空值、异常）",
    "代码逻辑是否真的解决了原始问题，而不只是看起来像",
]

ANALYSIS_CRITERIA = [
    "结论是否真的从数据中得出，还是先有结论再找数据支撑",
    "是否存在混淆相关性和因果性的推断",
    "分析是否覆盖了任务要求的所有维度",
    "建议是否具体可执行，还是停留在泛泛而谈",
]

# 防止反思循环的安全边界
MAX_REFLECTION_ROUNDS = 2  # 最多审查 2 轮

# 只对严重问题触发重新执行
# 轻微问题直接接受，不触发修正
SEVERITY_THRESHOLD = "重要"  # 低于此严重程度的问题忽略