Agent_02_ReAct模式

# 反例：没有 Thought，直接调工具
for step in steps:
    tool_name = llm.choose_tool(step)  # LLM 直接选工具
    result = tools.execute(tool_name)  # 执行

Thought: 用户想分析销售数据，但首先需要读取文件。我应该用 read_csv 工具。
Action: read_csv(path="sales.csv")
Observation: 读取成功，共 1000 行，5 列...
Thought: 数据已经读取完毕，接下来需要分析趋势。用 analyze 工具。
Action: analyze(data=...)
Observation: Q4 增长 15%...
Thought: 分析完成了，可以生成最终报告了。
Final Answer: 根据分析，Q4 销售增长 15%...

"""
ReAct Agent —— Thought → Action → Observation 循环
"""

from dataclasses import dataclass
from typing import Any


@dataclass
class ReActStep:
    """ReAct 单步记录"""
    thought: str = ""
    action_name: str = ""
    action_args: dict = None
    observation: str = ""
    is_final: bool = False


class ReActAgent:
    """ReAct Agent 主循环"""

    def __init__(self, llm, tool_registry, max_steps: int = 10, verbose: bool = True):
        self.llm = llm
        self.tools = tool_registry
        self.max_steps = max_steps
        self.verbose = verbose

    def run(self, goal: str) -> str:
        system_prompt = self._build_system_prompt()
        context = self._build_initial_context(goal)

        steps: list[ReActStep] = []

        for step_num in range(1, self.max_steps + 1):
            if self.verbose:
                print(f"\n{'='*50}")
                print(f"  第 {step_num} 步")
                print(f"{'='*50}")

            # 1. LLM 输出 Thought + Action
            response = self.llm.generate_with_react_format(
                context=context,
                system=system_prompt,
                steps=steps,
            )

            step = ReActStep(
                thought=response.get("thought", ""),
                action_name=response.get("action", ""),
                action_args=response.get("action_args", {}),
                is_final=response.get("is_final", False),
            )

            if self.verbose:
                print(f"  Thought: {step.thought}")

            # 2. 判断是否是最终答案
            if step.is_final:
                step.observation = step.action_name
                steps.append(step)
                if self.verbose:
                    print(f"  Final Answer: {step.observation}")
                return step.observation

            # 3. 执行工具
            if self.verbose:
                print(f"  Action: {step.action_name}({step.action_args})")

            result = self.tools.execute(step.action_name, **step.action_args)

            step.observation = result.output if result.success else f"错误: {result.error}"
            steps.append(step)

            if self.verbose:
                status = "成功" if result.success else f"失败: {result.error}"
                print(f"  Observation: {status}")

            # 4. 追加到上下文
            context = self._append_step(context, step)

        return "达到最大步数限制，未能完成。最后的结果:\n" + self._format_steps(steps)

    def _build_system_prompt(self) -> str:
        tool_list = "\n".join(
            f"- {t['name']}: {t['description']}"
            for t in self.tools.get_schemas()
        )
        return f"""你是一个 ReAct Agent。你必须严格遵守以下格式输出：

Thought: <你的思考过程>
Action: <工具名称>
Action Input: <工具的 JSON 参数>

或者当你有了最终答案时：

Thought: 我已经有了答案
Final Answer: <你的最终回复>

可用工具：
{tool_list}

注意：
1. 每一步必须先 Thought，再 Action
2. 不要编造 Observation，Observation 由我提供
3. 一旦有了最终答案，输出 Final Answer 并结束"""

    def _build_initial_context(self, goal: str) -> str:
        return f"目标: {goal}\n\n开始执行。\n"

    def _append_step(self, context: str, step: ReActStep) -> str:
        context += f"Thought: {step.thought}\n"
        if step.is_final:
            context += f"Final Answer: {step.observation}\n"
        else:
            context += f"Action: {step.action_name}\n"
            context += f"Action Input: {step.action_args}\n"
            context += f"Observation: {step.observation}\n"
        return context

    def _format_steps(self, steps: list[ReActStep]) -> str:
        lines = []
        for i, s in enumerate(steps, 1):
            lines.append(f"Step {i}:")
            lines.append(f"  Thought: {s.thought}")
            if not s.is_final:
                lines.append(f"  Action: {s.action_name}({s.action_args})")
            lines.append(f"  Observation: {s.observation}")
        return "\n".join(lines)

import re
import json


class ReActOutputParser:
    """ReAct 输出解析器 —— 把 LLM 的文本输出解析为结构化数据"""

    def parse(self, text: str) -> dict:
        """
        解析 LLM 输出

        返回格式：
        {
            "thought": "思考内容",
            "action": "工具名" 或 "",
            "action_args": {"key": "value"},
            "is_final": False,
            "final_answer": "",
        }
        """

        # 策略一：精确匹配 Thought/Action/Observation 格式
        thought_match = re.search(r"Thought:\s*(.+?)(?=\nAction:|\nFinal Answer:|$)", text, re.DOTALL)
        final_match = re.search(r"Final Answer:\s*(.+)$", text, re.DOTALL)
        action_match = re.search(r"Action:\s*(.+?)(?=\n|$)", text)
        args_match = re.search(r"Action Input:\s*(.+?)(?=\n|$)", text, re.DOTALL)

        thought = thought_match.group(1).strip() if thought_match else ""

        if final_match:
            return {
                "thought": thought,
                "action": "",
                "action_args": {},
                "is_final": True,
                "final_answer": final_match.group(1).strip(),
            }

        if action_match:
            action_name = action_match.group(1).strip()

            action_args = {}
            if args_match:
                args_text = args_match.group(1).strip()
                action_args = self._safe_parse_json(args_text)

            return {
                "thought": thought,
                "action": action_name,
                "action_args": action_args,
                "is_final": False,
                "final_answer": "",
            }

        # 策略二：模糊匹配
        return self._fuzzy_parse(text)

    def _safe_parse_json(self, text: str) -> dict:
        """安全解析 JSON，失败返回空字典"""
        try:
            return json.loads(text)
        except json.JSONDecodeError:
            try:
                fixed = re.sub(r'(\w+):', r'"\1":', text)
                return json.loads(fixed)
            except json.JSONDecodeError:
                brace_match = re.search(r'\{(.+?)\}', text, re.DOTALL)
                if brace_match:
                    try:
                        fixed = re.sub(r'(\w+):', r'"\1":', brace_match.group(1))
                        return json.loads("{" + fixed + "}")
                    except json.JSONDecodeError:
                        pass
                return {}

    def _fuzzy_parse(self, text: str) -> dict:
        """模糊解析——LLM 不按格式输出时的兜底"""
        conclusion_keywords = ["结论", "总结", "最终", "因此", "所以", "综上所述"]
        if any(kw in text for kw in conclusion_keywords):
            return {
                "thought": "",
                "action": "",
                "action_args": {},
                "is_final": True,
                "final_answer": text,
            }

        return {
            "thought": "",
            "action": "",
            "action_args": {},
            "is_final": True,
            "final_answer": text,
        }

from dataclasses import dataclass


@dataclass
class ReactResponse:
    """ReAct 响应"""
    thought: str = ""
    action: str = ""
    args: dict = None
    is_final: bool = False
    answer: str = ""
    needs_replan: bool = False


class ReActStrategy:
    """ReAct 循环的三种策略"""

    @staticmethod
    def standard(llm, tools, goal, max_steps=10):
        """
        标准 ReAct：每步 Thought → Action → Observation

        适用场景：工具不多、步骤不长的任务。
        优点：简单直接。
        缺点：容易陷入长循环，每步都要 LLM 输出。
        """
        context = f"目标: {goal}\n"
        for _ in range(max_steps):
            response = llm.generate_react(context)
            if response.is_final:
                return response.answer
            result = tools.execute(response.action, **response.args)
            context += f"\nThought: {response.thought}\nAction: {response.action}\nObservation: {result}"
        return "达到最大步数"

    @staticmethod
    def plan_then_act(llm, tools, goal, max_steps=10):
        """
        先规划再执行：先让 LLM 生成完整计划，然后按 ReAct 循环执行

        适用场景：步骤较多、逻辑复杂的任务。
        优点：LLM 先想好全局计划，不容易跑偏。
        缺点：规划可能不准确，执行中需要动态调整。
        """
        # Phase 1：规划
        plan = llm.generate_plan(goal)
        context = f"目标: {goal}\n计划: {plan}\n"

        # Phase 2：按 ReAct 执行，但可以偏离计划
        for step in range(max_steps):
            response = llm.generate_react_with_plan(context, plan)
            if response.is_final:
                return response.answer
            result = tools.execute(response.action, **response.args)
            context += f"\nObservation: {result}"
            if response.needs_replan:
                plan = llm.update_plan(context, plan)
        return "达到最大步数"

    @staticmethod
    def reflexion(llm, tools, goal, max_steps=10, max_retries=2):
        """
        Reflexion ReAct：执行后自我反思，结果不好就重试

        适用场景：对输出质量要求高、允许重试的任务。
        优点：自我纠错，结果质量高。
        缺点：成本高，每轮重试都要重新调 LLM。
        """
        context = f"目标: {goal}\n"
        for attempt in range(max_retries):
            result, steps = ReActStrategy._run_react_loop(
                llm, tools, context, max_steps
            )

            critique = llm.critique(result, goal)
            if critique.is_good:
                return result

            context += f"\n\n上次尝试结果: {result}\n反思: {critique.feedback}\n请重新执行，注意避免以上问题。"

        return result

    @staticmethod
    def _run_react_loop(llm, tools, context, max_steps):
        """ReAct 循环的内部实现"""
        steps = []
        for _ in range(max_steps):
            response = llm.generate_react(context)
            if response.is_final:
                return ReactResponse(is_final=True, answer=response.answer), steps
            result = tools.execute(response.action, **response.args)
            context += f"\nObservation: {result}"
            steps.append(response)
        return ReactResponse(is_final=True, answer="达到最大步数"), steps

"""
Few-shot ReAct Prompt 设计
"""


class ReActPromptBuilder:
    """ReAct 系统 prompt 构建器"""

    @staticmethod
    def bad(tools: list[dict]) -> str:
        """坏 prompt —— 只有规则描述，没有示例"""
        tool_desc = "\n".join(f"- {t['name']}: {t['description']}" for t in tools)
        return f"""你是一个智能助手。请使用以下格式回答：

Thought: 你的思考
Action: 工具名
Action Input: 参数
Observation: 结果

可用工具：
{tool_desc}

请先思考再行动。"""

    @staticmethod
    def good(tools: list[dict]) -> str:
        """好 prompt —— 一个完整的 few-shot 示例"""
        tool_desc = "\n".join(f"- {t['name']}: {t['description']}" for t in tools)
        return f"""你是一个数据分析助手。请严格遵守以下格式：

Thought: <思考为什么需要这个步骤>
Action: <工具名称>
Action Input: <JSON 格式的参数>
Observation: <工具执行结果（由我提供）>

或者当你有最终答案时：

Thought: <思考为什么可以结束了>
Final Answer: <你的最终回复>

可用工具：
{tool_desc}

---
示例：

用户：北京今天天气怎么样？

Thought: 用户想知道北京的天气，我需要调用天气查询工具。
Action: get_weather
Action Input: {{"city": "北京", "unit": "celsius"}}
Observation: 北京今天晴，气温 15-22°C，微风。

Thought: 已经获取了天气信息，可以回答用户了。
Final Answer: 北京今天晴天，气温 15-22°C，风力较小。
---

现在开始执行。遵守上面的格式，不要跳过 Thought。"""

    @staticmethod
    def production(tools: list[dict], preferences: dict = None) -> str:
        """生产级 prompt —— 多个示例 + 边界情况处理 + 偏好设置"""
        tool_desc = "\n".join(f"- {t['name']}: {t['description']}" for t in tools)
        tool_names = ", ".join(t["name"] for t in tools)

        prefs = ""
        if preferences:
            prefs = "\n用户偏好：\n" + "\n".join(f"- {k}: {v}" for k, v in preferences.items())

        return f"""你是一个 ReAct Agent。你必须严格遵守以下格式输出。

## 输出格式

每一步必须按以下顺序输出（不要跳过任何一步）：

Thought: <解释你为什么要做这个动作，1-2 句话>
Action: <工具名称，必须是以下之一：{tool_names}>
Action Input: <严格的 JSON 格式参数，key 用双引号>

Observation: <我会在下轮提供执行结果>

当你有了足够的信息来回答用户时：

Thought: <解释为什么可以结束了>
Final Answer: <你的最终回复>

## 禁止行为

1. 不要编造 Observation——Observation 由我提供
2. 不要使用不在工具列表中的工具
3. 不要在 Action Input 中使用单引号（必须用双引号）
4. 不要连续两次调用同一个工具（除非参数不同）
5. 如果已经尝试了 5 步还没有进展，请总结当前信息并输出 Final Answer

## 示例

示例一：工具调用

用户：比较北京和上海的人口

Thought: 用户需要比较两个城市的人口，我需要分别查询北京和上海的人口数据。先查北京。
Action: get_city_population
Action Input: {{"city": "北京"}}
Observation: 北京人口约 2189 万。

Thought: 已获取北京的数据，接下来查询上海的人口。
Action: get_city_population
Action Input: {{"city": "上海"}}
Observation: 上海人口约 2487 万。

Thought: 两个城市的数据都获取到了，可以比较并回答用户。
Final Answer: 北京人口约 2189 万，上海人口约 2487 万，上海比北京多约 298 万人。

示例二：信息不足时直接说明

用户：火星上有水吗？

Thought: 这个问题涉及天文学知识，我不需要调用任何工具，可以直接回答。但需要注意我掌握的信息可能不是最新的。
Final Answer: 根据目前的研究，火星上存在水冰形式的，主要分布在两极地区和地下。

{prefs}

现在开始执行。"""

"""
ReAct 上下文管理
"""


class ContextStrategy:
    """上下文管理策略"""

    @staticmethod
    def sliding_window(messages: list[dict], keep: int = 8, system: str = "") -> list[dict]:
        """
        策略一：滑动窗口

        只保留最近的 N 条消息。
        """
        system_msgs = [m for m in messages if m.get("role") == "system"]
        non_system = [m for m in messages if m.get("role") != "system"]

        return system_msgs + non_system[-keep:]

    @staticmethod
    def summary_compression(llm, messages: list[dict], keep: int = 5, system: str = "") -> list[dict]:
        """
        策略二：摘要压缩

        用 LLM 把早期步骤压缩成一条摘要消息。
        """
        system_msgs = [m for m in messages if m.get("role") == "system"]
        non_system = [m for m in messages if m.get("role") != "system"]

        if len(non_system) <= keep:
            return messages

        early = non_system[:-keep]
        early_text = "\n".join(f"{m['role']}: {m['content']}" for m in early)

        prompt = f"""请将以下对话历史压缩为一段摘要，只保留：
1. 用户的原始目标
2. 已完成的工具调用和关键结果
3. 当前的进度和发现
不要保留 Thought 和 Action 的细节。

{early_text}
"""
        summary = llm.generate(prompt, system="你是对话摘要专家。输出要简洁。")

        compressed = system_msgs + [
            {"role": "system", "content": f"[早期对话摘要]\n{summary}"},
        ] + non_system[-keep:]

        return compressed

    @staticmethod
    def tool_output_truncation(messages: list[dict], max_observation_len: int = 300) -> list[dict]:
        """
        策略三：工具输出截断

        从源头控制 Observation 的长度。
        """
        result = []
        for msg in messages:
            if msg.get("role") == "tool" and len(msg["content"]) > max_observation_len:
                truncated = msg["content"][:max_observation_len] + "...（已截断）"
                result.append({**msg, "content": truncated})
            else:
                result.append(msg)
        return result

    @staticmethod
    def hybrid(llm, messages: list[dict], token_threshold: float = 0.7) -> list[dict]:
        """
        混合策略：截断 + 滑动窗口 + 摘要压缩
        """
        messages = ContextStrategy.tool_output_truncation(messages, max_observation_len=500)
        messages = ContextStrategy.sliding_window(messages, keep=12)

        estimated_tokens = sum(len(m.get("content", "")) // 3 for m in messages)
        if estimated_tokens > token_threshold * 4096:
            messages = ContextStrategy.summary_compression(llm, messages, keep=5)

        return messages

"""
幻觉传播链分析
"""


class HallucinationTracker:
    """
    幻觉传播追踪器

    标记哪些信息是工具返回的，哪些是 LLM 推断的。
    如果最终答案引用了未被工具验证的信息，发出警告。
    """

    def __init__(self):
        self.steps: list[dict] = []
        self.facts: dict[str, str] = {}
        self.unverified_claims: list[str] = []

    def record_step(self, thought: str, action: str, observation: str) -> None:
        """记录一步，区分事实和推断"""
        self.steps.append({
            "thought": thought,
            "action": action,
            "observation": observation,
        })

        if observation:
            self._extract_facts(observation, source="observation")

        if thought:
            self._extract_claims(thought)

    def _extract_facts(self, text: str, source: str) -> None:
        """从 Observation 中提取事实"""
        import re
        patterns = [
            r"(\S+)\s+(?:是|有|为|约)\s+(.+?)[\.\,，。]",
            r"(\S+):\s*(.+?)[\.\,，。]",
        ]
        for pattern in patterns:
            for match in re.finditer(pattern, text):
                key = match.group(1).strip()
                value = match.group(2).strip()
                self.facts[f"{key}: {value}"] = source

    def _extract_claims(self, thought: str) -> None:
        """从 Thought 中提取声明"""
        import re
        patterns = [
            r"(?:因此|所以|说明|意味着|看来).+?[\.\,，。]",
            r"(?:应该|可能|大概|似乎).+?[\.\,，。]",
        ]
        for pattern in patterns:
            for match in re.finditer(pattern, thought):
                self.unverified_claims.append(match.group().strip())

    def verify_final_answer(self, answer: str) -> list[str]:
        """验证最终答案——检查是否引用了未被验证的信息"""
        import re
        unverified = []

        for claim in self.unverified_claims:
            if any(word in answer for word in re.findall(r'\w{2,}', claim)):
                supported = any(
                    fact_key in claim or fact_val in claim
                    for fact_key, fact_val in (f.split(": ") for f in self.facts)
                )
                if not supported:
                    unverified.append(claim)

        return unverified

    def report(self) -> str:
        """生成追踪报告"""
        lines = [
            f"共记录 {len(self.steps)} 步",
            f"已验证事实: {len(self.facts)} 条",
            f"未验证声明: {len(self.unverified_claims)} 条",
        ]
        if self.unverified_claims:
            lines.append("\n未验证声明:")
            for claim in self.unverified_claims:
                lines.append(f"  - {claim}")
        return "\n".join(lines)


# 使用示例
tracker = HallucinationTracker()

tracker.record_step(
    thought="数据已读取，接下来分析趋势",
    action="analyze(data)",
    observation="Q4 增长 15%，总销售额 500 万",
)
tracker.record_step(
    thought="看来 Q3 表现不佳，可能是季节性因素",
    action="search('Q3 销售下降原因')",
    observation="未找到相关信息",
)
tracker.record_step(
    thought="因此 Q3 下降主要是因为市场竞争加剧",
    action="",
    observation="",
)

report = tracker.report()
print(report)

# 低质量 Thought（纯复述）：
Thought: 我需要读取文件。
Action: read_csv

# 高质量 Thought（有推理）：
Thought: 用户提到"销售数据"，但没有指定格式。根据文件扩展名 .csv，
        应该用 read_csv 工具而不是 read_excel。如果文件不存在，
        需要提示用户确认路径。
Action: read_csv(path="sales.csv")

Thought: 我需要读取文件。
Action: read_csv
Observation: 成功

Thought: 文件读完了，我需要分析。
Action: analyze
Observation: 分析完成

Thought: 分析完了，我输出结果。
Action: output
Observation: 输出完成

Thought: 我需要分析销售数据的季度趋势
Action: read_image("photo.jpg")

data = read_csv("sales.csv")
result = analyze(data)

tool_name = llm.choose_tool(user_input)
result = tools.execute(tool_name)

plan = llm.generate_plan(user_input)
for tool in plan:
    result = tools.execute(tool, result)

for _ in range(max_steps):
    thought, action = llm.think_and_act(context)
    observation = tools.execute(action)
    context += f"\nThought: {thought}\nObservation: {observation}"