规划与反思：Self-Reflection 模式

from typing import Optional
from dataclasses import dataclass

@dataclass
class ReflectionResult:
    """反思结果的结构化表示"""
    is_correct: bool
    confidence: float  # 0.0 ~ 1.0
    error_type: Optional[str]  # "hallucination" | "logic" | "format" | None
    correction: Optional[str]  # 修正建议
    reasoning: str  # 为什么这样判断

import re
from openai import OpenAI

client = OpenAI()

def reflect_on_output(text: str, criteria: str) -> ReflectionResult:
    """让 LLM 对自身输出进行反思评估"""
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[
            {"role": "system", "content": f"你是严格的代码审查员。评估标准：{criteria}"},
            {"role": "user", "content": f"审查以下输出是否存在问题：\n\n{text}"}
        ],
        response_format={"type": "json_object"}
    )
    import json
    data = json.loads(response.choices[0].message.content)
    return ReflectionResult(**data)

class ReflexionAgent:
    """Reflexion 核心循环：Act → Evaluate → Reflect"""

    def __init__(self, llm, max_reflections=3):
        self.llm = llm
        self.max_reflections = max_reflections
        self.reflection_history = []

    def run(self, task: str) -> str:
        trajectory = []
        reflections = []

        for attempt in range(self.max_reflections + 1):
            # Act: 根据当前知识和反思历史生成答案
            answer = self.act(task, trajectory, reflections)
            trajectory.append(answer)

            # Evaluate: 评估答案是否正确
            eval_result = self.evaluate(task, answer)
            if eval_result.is_correct:
                return answer

            # Reflect: 分析错误，生成改进反馈
            reflection = self.reflect(task, answer, eval_result)
            reflections.append(reflection)

        return answer  # 返回最后一次尝试

    def act(self, task, trajectory, reflections):
        prompt = self._build_act_prompt(task, reflections)
        return self.llm.generate(prompt)

    def evaluate(self, task, answer):
        prompt = self._build_eval_prompt(task, answer)
        return self.llm.generate_json(prompt)

def reflect(task: str, answer: str, eval_result: EvalResult) -> str:
    """生成结构化的反思反馈"""
    prompt = f"""任务：{task}
你的回答：{answer}
评估结果：{'正确' if eval_result.is_correct else '错误'}
{'错误详情：' + eval_result.details if eval_result.details else ''}

请反思：
1. 你在哪一步出错了？
2. 错误的根本原因是什么？
3. 下次应该采取什么不同的策略？

请用简洁的语言回答，不超过 200 字。"""
    return llm.generate(prompt)

# 使用示例
agent = ReflexionAgent(llm=gpt4_client, max_reflections=3)
result = agent.run("找出 2024 年诺贝尔物理学奖得主的研究贡献")

class PlanAndSolveAgent:
    """Plan-and-Solve Agent：先规划，再逐步执行"""

    def __init__(self, llm):
        self.llm = llm

    def run(self, task: str) -> str:
        # Phase 1: Plan — 生成执行计划
        plan = self.generate_plan(task)
        print(f"执行计划：\n{plan}")

        # Phase 2: Solve — 按步骤执行
        results = []
        for i, step in enumerate(plan.steps, 1):
            step_result = self.execute_step(task, step, results)
            results.append(step_result)
            print(f"步骤 {i}/{len(plan.steps)}: {step_result}")

        # Phase 3: Synthesize — 综合所有步骤结果
        final = self.synthesize(task, results)
        return final

    def generate_plan(self, task: str) -> Plan:
        prompt = f"""任务：{task}

请制定一个详细的执行计划，包含：
1. 需要完成的所有子任务
2. 子任务之间的依赖关系
3. 每个步骤需要的输入和预期输出"""
        raw = self.llm.generate(prompt)
        return Plan.from_text(raw)

class Plan:
    """结构化的执行计划"""

    @dataclass
    class Step:
        id: int
        description: str
        depends_on: List[int]  # 依赖的步骤 ID
        expected_output: str

    steps: List[Step]

    @classmethod
    def from_text(cls, text: str) -> 'Plan':
        """从 LLM 生成的文本中解析出结构化计划"""
        import re
        steps = []
        for i, match in enumerate(
            re.finditer(r'步骤[：:](.+?)(?=步骤|$)', text, re.DOTALL)
        ):
            desc = match.group(1).strip()
            steps.append(
                cls.Step(
                    id=i+1,
                    description=desc,
                    depends_on=[],  # 默认无依赖
                    expected_output=""
                )
            )
        return cls(steps=steps)

    def execution_order(self) -> List[Step]:
        """按依赖关系拓扑排序"""
        from graphlib import TopologicalSorter
        ts = TopologicalSorter()
        for step in self.steps:
            ts.add(step.id, *step.depends_on)
        ordered_ids = list(ts.static_order())
        return [s for s in self.steps if s.id in ordered_ids]

class TreeNode:
    """ToT 搜索树节点"""

    def __init__(self, state: str, depth: int = 0):
        self.state = state       # 当前思维状态
        self.depth = depth       # 当前深度
        self.children = []       # 子节点（候选分支）
        self.value = 0.0         # 评估分数
        self.parent = None

    def add_child(self, child: 'TreeNode'):
        child.parent = self
        self.children.append(child)

class TreeOfThoughts:
    """Tree of Thoughts 搜索框架"""

    def __init__(self, llm, branching=3, max_depth=5, beam_width=5):
        self.llm = llm
        self.branching = branching    # 每个节点生成几个候选
        self.max_depth = max_depth    # 最大搜索深度
        self.beam_width = beam_width  # 束搜索宽度

    def search(self, initial_state: str) -> str:
        root = TreeNode(initial_state)
        frontier = [root]  # 当前层的候选节点

        for depth in range(self.max_depth):
            candidates = []

            # 生成：扩展每个 frontier 节点
            for node in frontier:
                children = self.generate_thoughts(node)
                for child in children:
                    node.add_child(child)
                    candidates.append(child)

            # 评估：为每个候选打分
            for candidate in candidates:
                candidate.value = self.evaluate(candidate)

            # 选择：保留得分最高的 beam_width 个
            candidates.sort(key=lambda n: n.value, reverse=True)
            frontier = candidates[:self.beam_width]

            # 检查是否找到满意解
            if frontier[0].value > 0.9:
                return self.extract_solution(frontier[0])

        return self.extract_solution(frontier[0])

class TreeOfThoughts:
    # ... 接上文 ...

    def generate_thoughts(self, node: TreeNode) -> List[TreeNode]:
        """从当前节点生成多个候选思维步骤"""
        prompt = f"""当前状态：{node.state}
深度：{node.depth}/{self.max_depth}

请生成 {self.branching} 个不同的下一步思维，
每个思维之间应该有实质性差异。

格式：
思维 1: ...
思维 2: ...
思维 3: ..."""
        response = self.llm.generate(prompt)
        thoughts = self._parse_thoughts(response)
        return [
            TreeNode(t, depth=node.depth + 1)
            for t in thoughts
        ]

    def evaluate(self, node: TreeNode) -> float:
        """评估节点状态的潜在价值"""
        prompt = f"""评估以下思维步骤的质量（0.0-1.0）：
{node.state}

评分标准：
- 逻辑连贯性
- 与问题的相关性
- 解决路径的可行性

只返回一个 0.0-1.0 的数字。"""
        score = self.llm.generate(prompt)
        return float(score.strip())

from collections import defaultdict

class ThoughtNode:
    """GoT 图中的思维节点"""

    def __init__(self, id: str, content: str, thought_type: str = "generate"):
        self.id = id
        self.content = content
        self.type = thought_type  # generate / aggregate / refine / loop
        self.parents = []
        self.children = []
        self.score = 0.0

class GraphOfThoughts:
    """Graph of Thoughts 框架"""

    def __init__(self, llm):
        self.llm = llm
        self.nodes = {}
        self.edges = defaultdict(list)

    def add_node(self, node: ThoughtNode):
        self.nodes[node.id] = node

    def add_edge(self, from_id: str, to_id: str):
        self.edges[from_id].append(to_id)
        self.nodes[to_id].parents.append(from_id)
        self.nodes[from_id].children.append(to_id)

    def aggregate(self, parent_ids: List[str], prompt: str) -> ThoughtNode:
        """聚合多个思维节点的结果"""
        parent_contents = [
            self.nodes[pid].content for pid in parent_ids
            if pid in self.nodes
        ]
        agg_prompt = f"""综合以下多个思维的结果：
{''.join(f'--- 思维 {i+1} ---\n{c}\n' for i, c in enumerate(parent_contents))}
\n{prompt}"""
        result = self.llm.generate(agg_prompt)
        node_id = f"agg_{len(self.nodes)}"
        node = ThoughtNode(node_id, result, "aggregate")
        self.add_node(node)
        for pid in parent_ids:
            self.add_edge(pid, node_id)
        return node

class GraphOfThoughts:
    # ... 接上文 ...

    def refine(self, node_id: str, iterations: int = 3) -> ThoughtNode:
        """对指定思维节点进行多轮精炼"""
        current = self.nodes[node_id]

        for i in range(iterations):
            prompt = f"""当前版本：
{current.content}

请改进上述内容，要求：
1. 修正任何逻辑漏洞
2. 补充缺失的关键信息
3. 提高表达的清晰度和精确性"""
            refined = self.llm.generate(prompt)
            new_id = f"{node_id}_refine_{i}"
            new_node = ThoughtNode(new_id, refined, "refine")
            self.add_node(new_node)
            self.add_edge(current.id, new_id)
            current = new_node

        return current

    def find_solution(self) -> Optional[ThoughtNode]:
        """找到得分最高的终端节点作为最终答案"""
        terminal_nodes = [
            n for n in self.nodes.values()
            if not n.children  # 没有子节点 = 终端节点
        ]
        if not terminal_nodes:
            return None
        return max(terminal_nodes, key=lambda n: n.score)

REFLECTION_PROMPT = """你是一个严格的自我审查系统。请按以下步骤审查你的上一个回答：

第一步：事实核查
- 回答中提到的每个事实陈述是否可以被验证？
- 是否存在无法证实的断言？
- 如有不确定的地方，请明确指出。

第二步：逻辑检查
- 推理链条是否完整？是否存在跳跃？
- 前提和结论之间是否有充分的逻辑关系？
- 是否存在自相矛盾的地方？

第三步：完整性评估
- 是否回答了问题的所有部分？
- 是否遗漏了重要的角度或信息？
- 是否有需要补充的上下文？

第四步：格式验证
- 输出是否符合要求的格式？
- 是否包含不必要的内容？

请以 JSON 格式返回审查结果：
{{
    "is_correct": true/false,
    "issues": ["问题1", "问题2"],
    "confidence": 0.0-1.0,
    "suggested_fix": "修正建议"
}}"""

def self_consistency_check(task: str, n_samples: int = 5) -> dict:
    """Self-Consistency 检查：生成多个答案并比较一致性"""
    answers = []
    for _ in range(n_samples):
        ans = llm.generate(
            task,
            temperature=0.7  # 增加多样性
        )
        answers.append(ans)

    # 提取关键结论并比较
    key_points = []
    for ans in answers:
        prompt = f"""从以下回答中提取核心结论（最多 3 条）：
{ans}
只列出结论，每条一行。"""
        points = llm.generate(prompt).strip().split('\n')
        key_points.append(set(points))

    # 计算一致性得分
    consensus = set.intersection(*key_points)
    consistency_score = len(consensus) / max(len(set.union(*key_points)), 1)

    return {
        "consensus_points": list(consensus),
        "consistency_score": round(consistency_score, 2),
        "divergent_points": [
            p for p in set.union(*key_points) if p not in consensus
        ]
    }

import subprocess
import tempfile
import os

class SelfCorrectingCoder:
    """自纠错代码生成 Agent"""

    def __init__(self, llm, max_attempts=5):
        self.llm = llm
        self.max_attempts = max_attempts

    def generate_and_test(self, problem: str, test_cases: str) -> dict:
        """生成代码并通过测试，失败则自我修正"""
        # Step 1: Plan
        plan = self._plan(problem)
        print(f"编码计划：\n{plan}\n")

        last_error = None
        for attempt in range(self.max_attempts):
            # Step 2: Generate
            if attempt == 0:
                code = self._generate(problem, plan)
            else:
                code = self._generate_with_feedback(
                    problem, plan, last_error
                )

            # Step 3: Test
            test_result = self._run_tests(code, test_cases)

            if test_result["passed"]:
                print(f"✅ 第 {attempt+1} 次尝试通过所有测试！")
                return {
                    "code": code,
                    "attempts": attempt + 1,
                    "success": True
                }

            last_error = test_result["error"]
            print(f"❌ 第 {attempt+1} 次尝试失败：{last_error}")

        return {
            "code": code,
            "attempts": self.max_attempts,
            "success": False,
            "last_error": last_error
        }

class SelfCorrectingCoder:
    # ... 接上文 ...

    def _plan(self, problem: str) -> str:
        """生成编码计划"""
        return self.llm.generate(f"""问题：{problem}

请制定编码计划：
1. 函数签名和参数设计
2. 核心算法思路
3. 边界情况处理
4. 时间/空间复杂度分析""")

    def _generate(self, problem: str, plan: str) -> str:
        """根据计划生成代码"""
        return self.llm.generate(f"""{plan}

请编写完整可运行的 Python 代码。
只返回代码，不要解释。""")

    def _generate_with_feedback(self, problem, plan, error: str) -> str:
        """基于错误反馈重新生成代码"""
        return self.llm.generate(f"""{plan}

上次生成的代码有以下错误：
{error}

请分析错误原因并修正代码。
只返回修正后的完整代码，不要解释。""")

    def _run_tests(self, code: str, tests: str) -> dict:
        """运行测试用例"""
        full_code = f"{code}\n\n{tests}"
        with tempfile.NamedTemporaryFile(
            mode='w', suffix='.py', delete=False
        ) as f:
            f.write(full_code)
            tmp_path = f.name

        try:
            result = subprocess.run(
                ['python3', tmp_path],
                capture_output=True, text=True, timeout=10
            )
            return {
                "passed": result.returncode == 0,
                "error": result.stderr[:500] if result.stderr else None
            }
        finally:
            os.unlink(tmp_path)