📅 2026-07-18　|　🏷️ Python · AI Agent 方向　|　⏱️ 建议 5h　|　🎯 Agent 面试的灵魂——ReAct 手写 + Function Calling 设计

---

📌 今日知识地图

Agent 面试全景
│
├── 模块一：Prompt Engineering
│   ├── Zero-shot / Few-shot / CoT / ToT / Self-Consistency
│   └── 为什么 ReAct 是 Agent 的基石？
│
├── 模块二：Function Calling 深度拆解
│   ├── OpenAI Function Calling 协议
│   ├── JSON Schema 工具定义（用 Pydantic 生成）
│   ├── 并行调用、错误处理、强制调用
│   └── Tool Choice 策略
│
├── 模块三：Agent 核心架构（四要素）
│   ├── Planning（规划）：任务分解 / 反思 / 自适应
│   ├── Memory（记忆）：短期 + 长期 + 工作记忆
│   ├── Tools（工具）：注册 / 校验 / 重试 / Fallback
│   └── Action（行动）：执行 / 解析 / 循环终止
│
├── 模块四：ReAct 框架完整手写实现
│   ├── Thought → Action → Observation 循环
│   ├── 状态管理 & 循环终止条件
│   └── 完整可运行代码
│
└── 模块五：Agent 决策策略对比
    └── ReAct vs Plan-and-Execute vs Reflexion vs LATS

---

模块一：Prompt Engineering — 从入门到 Agent 基石

1.1 一张图看懂提示工程的进化

Zero-shot              "翻译成英文：你好"
  ↓ 效果不够好？给示例
Few-shot               "Q: 苹果 → A: Apple   Q: 你好 → A:"
  ↓ 推理题还是错？让模型展示推理过程
Chain-of-Thought (CoT) "一步一步思考：小明有 5 个苹果..."
  ↓ 一条思路可能不对？多条思路投票
Self-Consistency       "用 5 种不同的思路解这道题，取多数答案"
  ↓ 复杂问题需要回溯？用树形结构
Tree-of-Thought (ToT)  生成多个思路 → 评估 → 选最优分支 → 继续
  ↓ 需要跟外部环境交互？
ReAct                  思考(Thought) → 行动(Action) → 观察(Observation) → 循环

1.2 各策略详解

Zero-shot：不给示例，直接问。

Prompt: 判断这条评论的情感是正面还是负面："等了两个小时才上菜，服务员态度也很差。"
Output: 负面

Few-shot：给几个示例，让模型学会格式。本质是 In-Context Learning——不更新模型参数，通过示例激活模型的模式匹配能力。

评论："环境很好，菜品精致" → 情感：正面
评论："价格太贵了，不值这个价" → 情感：负面
评论："无功无过，随便吃吃还行" → 情感：中性
评论："等了两小时，态度还差" → 情感：
Output: 负面

面试话术：“Few-shot 的关键技巧：示例覆盖目标场景多样性（正/负/中都要有），格式高度一致（模型对格式极其敏感），示例顺序有影响（近因效应，重要示例放最后）。”

Chain-of-Thought (CoT)：让模型"出声思考"。对于复杂推理任务，仅靠"请一步一步思考"就能大幅提升准确率。原理是自回归生成中前面的 token 影响后面——推理过程本身充当了"工作记忆"，把隐含推理变成了显式文字。

Prompt: 小明有 5 个苹果，小红给了他 3 个，他又吃了 2 个，现在有几个？请一步一步推理。

Output:
第一步：小明初始有 5 个苹果
第二步：小红给了他 3 个 → 5 + 3 = 8 个
第三步：他吃了 2 个 → 8 - 2 = 6 个
答案：6 个

Self-Consistency：一个思路不够，多个思路投票。对同一问题采样多条 CoT 推理路径（调高 temperature），多数投票选答案。核心洞见：模型不是一直错的——正确答案在不同采样中出现最频繁。代价是计算量翻 N 倍（N 通常 5-10）。

Tree-of-Thought (ToT)：把推理建模为树搜索。每步生成多个候选思路 → 评估每个分支 → 选最有希望的分支深入 → 可以回溯。比 CoT 更系统，但延迟也更高。

1.3 ReAct：为什么它是所有 Agent 的基石？

ReAct = Reasoning + Acting = 将推理（CoT）和行动（工具调用）交织在一起。

Q: 北京今天的气温比上海高多少？

传统 CoT（只会想，不会查）：
  A: 我需要知道北京和上海的气温...但我不知道，让我猜一下...

ReAct（边想边查）：
  Thought: 我需要查北京和上海当前的气温
  Action: search_weather("北京")
  Observation: 北京当前 35°C
  Thought: 北京 35°C，还需要上海的气温
  Action: search_weather("上海")
  Observation: 上海当前 32°C
  Thought: 北京 35°C, 上海 32°C, 北京比上海高 3°C
  Final Answer: 北京比上海高 3°C

ReAct 论文（Yao et al., 2022）的核心发现：单独用 CoT 会胡编乱造（因为不知道事实），单独用 Action 会盲目试错（因为没有推理指引方向）。两者结合产生 1+1>2 的效果。 消融实验证明，去掉 Reasoning 后退化为盲目试错，去掉 Acting 后退化为胡编乱造。

面试加分点：“这正是所有现代 Agent 框架——LangChain AgentExecutor、OpenAI Assistants、Anthropic Tool Use、字节 Coze——的底层范式。”

---

模块二：Function Calling — LLM 长出双手

2.1 最关键的认知纠正

Function Calling ≠ LLM 在调用函数

LLM 只负责输出结构化 JSON（选哪个工具、传什么参数）
→ 你的代码解析 JSON
→ 你的代码执行真正的函数
→ 结果以 tool role 消息喂回 LLM
→ LLM 继续生成

LLM 的角色 = 路由器 + 参数提取器（不是执行器！）

2.2 完整调用流程（五步）

第一步：发送请求，tools 参数告诉 LLM 有哪些工具可用。

POST /chat/completions
{
  "model": "gpt-4",
  "messages": [{"role": "user", "content": "北京天气怎么样？"}],
  "tools": [{
    "type": "function",
    "function": {
      "name": "get_weather",
      "description": "查询指定城市的天气",
      "parameters": {
        "type": "object",
        "properties": {
          "city": {"type": "string", "description": "城市名"},
          "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}
        },
        "required": ["city"]
      }
    }
  }],
  "tool_choice": "auto"
}

第二步：LLM 返回 tool_calls（JSON 格式，finish_reason=“tool_calls”）。

{
  "choices": [{
    "finish_reason": "tool_calls",
    "message": {
      "role": "assistant",
      "tool_calls": [{
        "id": "call_abc123",
        "type": "function",
        "function": {
          "name": "get_weather",
          "arguments": "{\"city\": \"北京\", \"unit\": \"celsius\"}"
        }
      }]
    }
  }]
}

第三步：你的代码执行 get_weather("北京", "celsius")。

第四步：把执行结果以 tool role 消息喂回 LLM。

messages += [
  assistant_message_with_tool_calls,
  {"role": "tool", "tool_call_id": "call_abc123",
   "content": "北京今天晴，35°C，湿度 60%"}
]

第五步：LLM 基于结果生成最终回复：“北京今天晴天，气温 35°C，湿度 60%，注意防暑降温。”

2.3 用 Pydantic 自动生成 JSON Schema（Agent 最佳实践）

from pydantic import BaseModel, Field
from typing import Literal

class WeatherParams(BaseModel):
    """天气查询参数"""
    city: str = Field(..., description="城市名", min_length=1, max_length=100)
    unit: Literal["celsius", "fahrenheit"] = Field(default="celsius")

class SearchParams(BaseModel):
    """搜索参数"""
    query: str = Field(..., description="搜索关键词", min_length=1, max_length=500)
    max_results: int = Field(default=5, ge=1, le=20)

class ToolRegistry:
    """工具注册器 —— Agent 开发的核心组件"""

    def __init__(self):
        self._tools = {}

    def register(self, name: str, description: str,
                 params_model: type[BaseModel], handler: callable):
        self._tools[name] = {
            "openai_spec": {
                "type": "function",
                "function": {
                    "name": name,
                    "description": description,
                    "parameters": params_model.model_json_schema(),
                }
            },
            "handler": handler,
            "params_model": params_model,
        }

    def get_openai_tools(self) -> list[dict]:
        """返回 OpenAI API 格式的 tools 列表"""
        return [t["openai_spec"] for t in self._tools.values()]

    def execute(self, name: str, arguments: dict):
        """执行工具：Pydantic 校验 → 调用 handler → 返回结果"""
        if name not in self._tools:
            return f"错误：未知工具 {name}"
        tool = self._tools[name]
        params = tool["params_model"](**arguments)  # Pydantic 自动校验！
        return tool["handler"](**params.model_dump())

# 使用示例
registry = ToolRegistry()
registry.register("get_weather", "查询指定城市的天气", WeatherParams,
                  lambda city, unit: f"{city}天气：晴，35°C")
registry.register("search", "搜索互联网", SearchParams,
                  lambda query, max_results: f"搜索'{query}'返回{max_results}条")

# 一行生成所有工具的 OpenAI 格式定义
print(registry.get_openai_tools())

2.4 Tool Choice 四种策略

tool_choice	行为	适用场景
"auto"（默认）	LLM 自己决定是否调工具	大多数场景
"none"	强制不调工具，只生成文本	不需要工具的闲聊
"required"	必须调一个工具	Agent 中间步骤（必须获取外部信息）
{"type":"function","function":{"name":"xxx"}}	强制调指定工具	已知需要用哪个工具

2.5 并行工具调用 & 错误处理

import asyncio, json

async def agent_loop(messages, tools, max_iterations=10):
    """Agent 主循环 —— 支持并行工具调用"""
    for i in range(max_iterations):
        response = await llm_call(messages, tools)
        msg = response.choices[0].message

        # LLM 给出最终回复（不再调工具）
        if not msg.tool_calls:
            return msg.content

        messages.append(msg)

        # 并行执行所有工具调用（这是 Agent 提速的关键！）
        async def exec_one(tc):
            func_name = tc.function.name
            arguments = json.loads(tc.function.arguments)
            try:
                return await registry.execute(func_name, arguments)
            except Exception as e:
                return f"工具执行失败: {e}"

        results = await asyncio.gather(
            *[exec_one(tc) for tc in msg.tool_calls],
            return_exceptions=True  # 某个失败不拖累其他
        )

        # 把所有结果喂回 LLM
        for tc, result in zip(msg.tool_calls, results):
            messages.append({
                "role": "tool",
                "tool_call_id": tc.id,
                "content": str(result),
            })

    return "达到最大迭代次数"

---

模块三：Agent 核心架构 — 四要素深度拆解

3.1 架构总览

                    ┌──────────────┐
  用户输入 ────────→│   Planning   │←──────┐
                    │  任务分解    │       │
                    └──────┬───────┘       │
                           │               │
              ┌────────────┼───────────────┤
              ↓            ↓               │
        ┌──────────┐ ┌──────────┐   ┌─────┴─────┐
        │  Memory  │ │  Tools   │   │  Action   │
        │ 短期+长期 │ │ 工具集   │   │ 执行+解析  │
        └──────────┘ └──────────┘   └───────────┘
              │            │               │
              └────────────┴───────────────┘
                           │
                    回到 Planning (反思调整)

3.2 Planning（规划）— Agent 的大脑皮层

层次	做什么	代表方法	例子
任务分解	把大任务拆成子任务	CoT, ReAct, Plan-and-Solve	“写市场报告” → 搜索数据 → 分析竞品 → 撰写
反思	评估之前行动，修正错误	Self-Refine, Reflexion	“第一步搜索结果不相关，换关键词重搜”
自适应	根据进度调整后续计划	Re-prompting, Re-planning	“已花3步在搜索上，先基于现有信息写初稿”

3.3 Memory（记忆）— Agent 不健忘的关键

三层记忆架构：

┌─────────────────────────────────────┐
│         短期记忆（上下文窗口）        │
│   当前对话的最近 N 条消息             │
│   容量：~128K tokens（GPT-4o）       │
│   特点：精确，但容量有限              │
├─────────────────────────────────────┤
│         工作记忆（Scratchpad）        │
│   当前任务的中间状态（Redis）         │
│   特点：任务隔离，快速读写             │
├─────────────────────────────────────┤
│         长期记忆（向量数据库）         │
│   历史对话摘要、用户偏好、知识库       │
│   容量：近乎无限                      │
│   特点：语义检索，可能有噪声           │
└─────────────────────────────────────┘

class AgentMemory:
    def __init__(self, vector_db, redis_client):
        self.vector_db = vector_db       # ChromaDB / Milvus
        self.redis = redis_client        # 工作记忆
        self.short_term: list[dict] = [] # 短期记忆

    def add_message(self, role: str, content: str):
        self.short_term.append({"role": role, "content": content})
        # 短期记忆超过阈值 → 压缩到长期记忆
        if len(self.short_term) > 40:
            self._compress_to_long_term()

    def _compress_to_long_term(self):
        """将旧消息压缩为摘要，存入长期记忆"""
        old_messages = self.short_term[:25]
        summary = llm_summarize(old_messages)  # LLM 生成摘要
        self.vector_db.insert(
            text=summary,
            metadata={"timestamp": time.time(), "type": "conversation_summary"}
        )
        self.short_term = self.short_term[25:]  # 保留最近的

    def retrieve_context(self, query: str, top_k: int = 5) -> str:
        """从长期记忆中检索与当前查询相关的内容"""
        results = self.vector_db.search(query, top_k=top_k)
        return "\n".join(r["text"] for r in results)

面试话术：“记忆系统的三个关键设计决策：① 压缩策略——何时触发压缩、摘要保留哪些信息；② 检索时机——每次对话前检索还是按需检索；③ 记忆衰减——旧信息权重降低（时间衰减或重要性加权）。”

3.4 Tools（工具）— 设计原则

原则	说明	反面案例
单一职责	一个工具只做一件事	“万能助手” — LLM 不知道该不该用它
幂等性	同参数多次调用结果一致	"发送邮件"调两次发两封
明确 Schema	JSON Schema 描述精确到每个字段	参数: data: string — 太模糊
有意义错误	失败时返回 LLM 能理解的信息	Error: 500 vs 搜索API配额耗尽，请改用本地搜索
超时保护	长时间无响应要超时	一直等 → Agent 卡死
人工确认	敏感操作（付费、删除）需确认	直接删库不确认

3.5 Action（行动）— 闭环的最后一公里

Action 模块的职责链：

接收 tool_calls → 解析函数名+参数 → ToolManager.execute()
→ 解析结果（成功/失败/部分成功）
→ 判断循环终止条件：
    ① finish_reason="stop"（LLM 给出最终文本回复）
    ② 达到最大迭代次数（兜底）
    ③ 人工介入终止（Human-in-the-loop）
    ④ 连续 N 次相同 Action+Input（死循环检测）
→ 格式化为 Observation → 喂回 Planning

---

面试题精选（8 道）

Q1. ReAct 框架的原理？跟纯 CoT 有什么区别？（字节/阿里/腾讯 高频）

标准回答：

ReAct = Reasoning + Acting，将推理和行动在同一个循环中交替进行。Thought 分析状况并决定行动，Action 调用工具获取信息，Observation 接收结果进入下一轮思考。

跟 CoT 的核心区别：CoT 只推理不行动，信息来源于模型训练时的知识，无法获取实时信息，容易幻觉。ReAct 通过 Action 与外部交互获取真实信息，Reasoning 让行动有方向（不乱试），Acting 让推理有依据（不乱编）。两者互补产生 1+1>2 的效果。

所有现代 Agent 框架（LangChain AgentExecutor、OpenAI Function Calling、Anthropic Tool Use）本质上都是 ReAct 范式的工程化实现。

Q2. Function Calling 中 LLM 真的调用了函数吗？（字节/腾讯）

标准回答：

不。LLM 只负责输出结构化 JSON（选哪个工具、传什么参数），真正的函数执行由开发者的代码完成。流程：开发者把工具定义放在 tools 参数里 → LLM 返回 tool_calls（JSON 格式的函数名+参数） → 代码解析并执行 → 结果以 tool role 消息喂回 LLM → LLM 继续生成。

LLM 在流程中的角色是"路由器 + 参数提取器"，不是"执行器"。这是面试中最重要的认知纠正。

Q3. Agent 的四个核心组件？（阿里/百度）

标准回答：

Planning（规划）：任务分解、反思和自适应调整。没有规划 = 布朗运动。
Memory（记忆）：短期（上下文窗口）+ 工作（Redis）+ 长期（向量数据库）三层。没有记忆 = 每次都是"初次见面"。
Tools（工具）：扩展 LLM 能力边界。没有工具 = “只会聊天的脑子”。
Action（行动）：执行工具、解析结果、判断终止条件。没有 Action = “光想不做”。

四者形成闭环：Planning 决定做什么 → Tools 提供能力 → Action 执行 → Memory 保存结果 → Planning 基于记忆反思调整。

Q4. 如何设计 Agent 的记忆系统？（腾讯/字节）

标准回答：

三层架构：① 短期记忆 = 上下文窗口，保留最近 N 轮对话，精确但容量有限（~128K tokens）；② 工作记忆 = Redis，存储当前任务中间状态（“已搜索哪些关键词”），任务间隔离；③ 长期记忆 = 向量数据库（ChromaDB/Milvus），存储历史对话摘要和用户偏好，语义检索按需召回。

关键设计决策：压缩策略（窗口达80%触发LLM摘要→存入长期记忆）、检索时机（按需 vs 每轮检索）、记忆衰减（时间衰减或重要性加权，旧信息权重降低）。

Q5. 如何处理 Agent 工具调用的错误？（字节/快手）

标准回答：

三层容错：① 参数校验（Pydantic 调用前校验类型和范围，非法参数直接拒绝并返回明确错误信息）；② 重试机制（指数退避，最多3次，超时和临时错误可重试，业务错误不重试）；③ Fallback 降级（主工具失败 → 备选工具 → 告知 LLM 当前能力受限）。

最关键的原则：错误信息必须对 LLM 友好。返回 “搜索 API 配额耗尽，建议使用本地搜索或告知用户稍后重试” 而不是 “Error: 500”。LLM 看到明确、可操作的错误信息后才能做出合理的下一步决策。

Q6. Plan-and-Execute 和 ReAct 怎么选？（阿里/腾讯）

标准回答：

任务明确、环境稳定、步骤间依赖清晰 → Plan-and-Execute（如"写一份市场分析报告"，先规划子任务再逐步执行）。环境不确定、需根据中间结果调整策略 → ReAct（如"帮我 Debug 这段代码"，调试过程充满意外）。实际产品常用混合方案：高层用 Plan-and-Execute 做任务分解，底层子任务用 ReAct 灵活执行。

Q7. 如何防止 Agent 陷入死循环？（字节）

标准回答：

三种检测机制：① 最大迭代次数（简单粗暴，兜底保护）；② 重复动作检测（连续3次相同 Action + 相同 Input，说明卡在同一个操作里）；③ 无进展检测（连续N步后没有向目标靠近，如搜索仍无果）。

检测到死循环后：终止当前任务 → 告知用户当前进度 → 请求人工介入或重新规划。生产环境通常 max_iterations=15，重复检测阈值=3。

Q8. 提高 Function Calling 准确率的方法？（字节/腾讯）

标准回答：

五个方法：① JSON Schema 越精确越好（字段类型、枚举值、取值范围、自然语言描述缺一不可）；② 工具 description 写清楚"什么时候用"而非"做什么"；③ Few-shot 示例（在 system prompt 中给1-2个正确的工具调用示例）；④ 简单工具优于复杂工具（一个工具只做一件事，LLM 更容易选对）；⑤ 错误反馈闭环（工具调用失败时让 LLM 看到具体错误信息并尝试修正）。

---

今日代码实战（3 道）

# ═══════════════════════════════════════════════
# 练习 1：精简版 ReAct Agent（可直接运行理解核心循环）
# ═══════════════════════════════════════════════
import re, json

class SimpleReAct:
    """最精简的 ReAct Agent —— 帮助理解 Thought → Action → Observation 循环"""

    def __init__(self, tools: dict, max_steps: int = 5):
        self.tools = tools          # {"tool_name": handler_function}
        self.max_steps = max_steps
        self.history = []           # 记录完整执行轨迹

    def _build_prompt(self, query: str) -> str:
        tool_desc = "\n".join(
            f"- {name}: {fn.__doc__}" for name, fn in self.tools.items()
        )
        return f"""你是 ReAct Agent。可用工具：
{tool_desc}

回复格式（严格遵守）：
Thought: 你的推理
Action: 工具名
Action Input: JSON参数

或者：

Thought: 你的推理
Final Answer: 最终回答

用户问题：{query}"""

    def _parse(self, text: str) -> dict:
        """解析 LLM 的 ReAct 格式输出"""
        result = {"thought": "", "action": "", "input": {}, "is_final": False}

        # 提取 Thought
        tm = re.search(r"Thought:\s*(.+?)(?=\n(?:Action|Final)|$)", text, re.DOTALL)
        if tm:
            result["thought"] = tm.group(1).strip()

        # 检查是否为最终回答
        fm = re.search(r"Final Answer:\s*(.+)", text, re.DOTALL)
        if fm:
            result["is_final"] = True
            result["final_answer"] = fm.group(1).strip()
            return result

        # 提取 Action 和 Action Input
        am = re.search(r"Action:\s*(.+?)$", text, re.MULTILINE)
        im = re.search(r"Action Input:\s*(\{.+?\})", text, re.DOTALL)
        if am:
            result["action"] = am.group(1).strip()
        if im:
            try:
                result["input"] = json.loads(im.group(1))
            except json.JSONDecodeError:
                result["input"] = {}

        return result

    def run(self, query: str, llm) -> str:
        """Agent 主循环 —— llm 参数是调用方传入的 LLM 函数"""
        prompt = self._build_prompt(query)
        msgs = [{"role": "user", "content": prompt}]

        for step in range(self.max_steps):
            response = llm(msgs)  # 调用 LLM（OpenAI API / 本地模型等）
            parsed = self._parse(response)

            if parsed["is_final"]:
                return parsed["final_answer"]

            action = parsed["action"]
            if action not in self.tools:
                msgs.append({"role": "user",
                    "content": f"错误：未知工具 {action}，请使用可用工具重试。"})
                continue

            # 执行工具
            try:
                result = self.tools[action](**parsed["input"])
            except Exception as e:
                result = f"工具执行失败: {e}"

            # 记录历史
            self.history.append({
                "step": step + 1, "thought": parsed["thought"],
                "action": action, "input": parsed["input"],
                "result": str(result),
            })

            # 喂回结果
            msgs.append({"role": "user", "content": f"Observation: {result}"})

        return f"超过最大步数 {self.max_steps}"


# ═══════════════════════════════════════════════
# 练习 2：Pydantic 工具注册器 —— Agent 工具管理的标准方案
# ═══════════════════════════════════════════════
from pydantic import BaseModel, Field
from typing import Callable, Any

class FunctionTool:
    """单个工具的完整封装 —— 定义 + 校验 + 执行"""
    def __init__(self, name: str, description: str,
                 params_model: type[BaseModel], handler: Callable):
        self.name = name
        self.handler = handler
        self.params_model = params_model
        # 自动生成 OpenAI Function Calling 格式
        self.openai_spec = {
            "type": "function",
            "function": {
                "name": name,
                "description": description,
                "parameters": params_model.model_json_schema(),
            }
        }

    def execute(self, arguments: dict) -> Any:
        """先 Pydantic 校验参数 → 再调用 handler"""
        params = self.params_model(**arguments)
        return self.handler(**params.model_dump())


# ═══════════════════════════════════════════════
# 练习 3：记忆管理器 —— Agent 不健忘的秘密
# ═══════════════════════════════════════════════
class MemoryManager:
    """Agent 记忆管理器精简版"""
    def __init__(self, max_short_term: int = 40, summary_trigger: int = 30):
        self.short_term: list[dict] = []
        self.long_term: list[str] = []     # 生产环境用向量数据库替代 list
        self.max_short = max_short_term
        self.trigger = summary_trigger

    def add(self, msg: dict):
        """添加消息，超过阈值自动压缩到长期记忆"""
        self.short_term.append(msg)
        if len(self.short_term) >= self.trigger:
            self._compress()

    def _compress(self):
        """压缩旧消息为摘要"""
        old = self.short_term[:20]
        # 生产环境：调用 LLM 生成摘要而非简单拼接
        summary = f"对话摘要（共{len(old)}条）: " + \
                  "; ".join([m.get("content", "")[:80] for m in old])
        self.long_term.append(summary)
        self.short_term = self.short_term[20:]

    def get_context(self, query: str) -> list[dict]:
        """组装给 LLM 的完整上下文"""
        # 从长期记忆中取相关摘要（生产环境用向量搜索）
        relevant = [{"role": "system", "content": f"[历史记忆] {m}"}
                    for m in self.long_term[-3:]]
        return relevant + self.short_term

---

模块四：ReAct 框架完整手写实现（面试高分手写题！）

import json, asyncio, re
from dataclasses import dataclass, field

@dataclass
class AgentStep:
    """单次 ReAct 循环的完整记录"""
    thought: str
    action: str
    action_input: dict
    observation: str

@dataclass
class ReActAgent:
    """ReAct Agent 完整实现 —— 建议手写三遍形成肌肉记忆"""

    llm: callable                       # LLM 调用函数
    tool_manager: 'ToolManager'         # 工具管理器
    system_prompt: str = ""             # 用户自定义系统提示词
    max_iterations: int = 10            # 最大迭代次数
    verbose: bool = True                # 是否打印详细日志

    # 运行时状态
    messages: list[dict] = field(default_factory=list)
    steps: list[AgentStep] = field(default_factory=list)

    def _build_react_prompt(self) -> str:
        """构建 ReAct 格式的系统提示词 —— 这是整个 Agent 的灵魂"""
        return f"""你是一个 ReAct Agent，通过"思考-行动-观察"的循环来解决问题。

## 回复格式（严格遵守！）

需要调用工具时：
Thought: <你的推理过程——分析当前状况，决定下一步做什么>
Action: <工具名称>
Action Input: <JSON 格式的工具参数>

已经可以回答用户时：
Thought: 我现在已经掌握了所有必要信息
Final Answer: <你的最终回答>

## 重要规则
1. 每次只能调用一个工具
2. 工具参数必须是合法的 JSON
3. 如果工具返回错误，分析错误原因并尝试修正参数后重试
4. 不要编造信息——所有事实信息必须来自工具调用结果
5. 不确定用哪个工具时，在 Thought 中分析后再做决定

{self.system_prompt}"""

    async def run(self, user_query: str) -> str:
        """Agent 主循环 —— ReAct 的完整实现"""
        self.messages = [
            {"role": "system", "content": self._build_react_prompt()},
            {"role": "user", "content": user_query},
        ]

        for i in range(self.max_iterations):
            if self.verbose:
                print(f"\n{'='*50}")
                print(f"第 {i+1}/{self.max_iterations} 轮")
                print(f"{'='*50}")

            # ① 调用 LLM
            response = await self.llm(self.messages)

            if self.verbose:
                print(f"\nLLM 回复:\n{response[:300]}...")

            # ② 解析 LLM 输出
            thought, action, action_input = self._parse_react_output(response)

            if self.verbose:
                print(f"\nThought: {thought}")

            # ③ 检查是否是最终回答
            if action == "FINAL_ANSWER":
                if self.verbose:
                    print(f"Final Answer: {action_input}")
                return action_input

            # ④ 执行工具
            if self.verbose:
                print(f"Action: {action}({action_input})")

            result = await self.tool_manager.execute(action, action_input)
            observation = result.data if result.success else f"错误: {result.error}"

            if self.verbose:
                print(f"Observation: {str(observation)[:200]}")

            # ⑤ 记录步骤
            step = AgentStep(
                thought=thought, action=action,
                action_input=action_input, observation=observation,
            )
            self.steps.append(step)

            # ⑥ 将 Assistant 回复和工具结果都加入对话历史
            self.messages.append({"role": "assistant", "content": response})
            self.messages.append({"role": "user", "content": f"Observation: {observation}"})

            # ⑦ 死循环检测
            if self._detect_loop():
                return "检测到死循环，Agent 停止。请检查工具定义和提示词。"

        return (f"Agent 在 {self.max_iterations} 轮迭代后仍未完成任务。"
                f"已执行步骤：{len(self.steps)} 步")

    def _parse_react_output(self, text: str) -> tuple:
        """解析 LLM 的 ReAct 格式输出 —— 核心解析器"""
        # 提取 Thought
        thought_m = re.search(r"Thought:\s*(.+?)(?=\n(?:Action|Final)|$)", text, re.DOTALL)
        thought = thought_m.group(1).strip() if thought_m else ""

        # 检查是否为最终回答
        final_m = re.search(r"Final Answer:\s*(.+)", text, re.DOTALL)
        if final_m:
            return thought, "FINAL_ANSWER", final_m.group(1).strip()

        # 提取 Action
        action_m = re.search(r"Action:\s*(.+?)(?=\n|$)", text)
        action = action_m.group(1).strip() if action_m else ""

        # 提取 Action Input（JSON）
        input_m = re.search(r"Action Input:\s*(\{.+?\})", text, re.DOTALL)
        action_input = {}
        if input_m:
            try:
                action_input = json.loads(input_m.group(1))
            except json.JSONDecodeError:
                action_input = {"raw": input_m.group(1)}

        return thought, action, action_input

    def _detect_loop(self) -> bool:
        """检测死循环：连续 3 次相同 Action + 相同 Input"""
        if len(self.steps) < 3:
            return False
        last_three = self.steps[-3:]
        return all(
            s.action == last_three[0].action and
            s.action_input == last_three[0].action_input
            for s in last_three
        )

---

模块五：Agent 决策策略对比

策略	流程	优点	缺点	适用场景
ReAct	Thought→Action→Observation 串行循环	简单可靠，社区最成熟	串行效率低（一次一个工具）	大多数通用 Agent
Plan-and-Execute	先完整规划→再逐步执行	全局最优，可并行无依赖步骤	计划跟不上变化	任务明确、环境稳定
Reflexion	ReAct + 长期失败记忆	能从错误中学习、持续改进	需额外存储和检索系统	需要持续优化的 Agent
LATS	ToT + ReAct + MCTS	理论上最优（搜索+推理+行动）	极其昂贵（每次多条路径）	需要极高可靠性的关键任务
LLMCompiler	分析依赖→并行执行无依赖工具	速度极快	依赖分析可能出错	多工具并行查询场景

面试话术：“实际项目中我倾向于混合策略：高层用 Plan-and-Execute 做任务分解和依赖分析，底层每个子任务用 ReAct 灵活执行。这样既有全局规划的效率，又有单步执行的灵活性。LangGraph 的 StateGraph + 条件边就是这种混合策略的最佳实现。”

---

📊 今日知识图谱

Agent 核心速通 DAY 4
│
├── Prompt Engineering
│   ├── Zero-shot → Few-shot(ICL) → CoT(出声思考)
│   ├── Self-Consistency(多路径投票) → ToT(树搜索)
│   └── ReAct = Reasoning + Acting（所有 Agent 的根基）
│
├── Function Calling
│   ├── 原理：LLM输出JSON → 代码执行 → 结果喂回（LLM 不是执行器！）
│   ├── Pydantic → JSON Schema（一行生成工具定义）
│   ├── tool_choice 四种策略（auto/none/required/指定）
│   └── 并行调用 + 错误处理 + 重试机制
│
├── Agent 四要素
│   ├── Planning：分解 → 反思 → 自适应
│   ├── Memory：短期(上下文) + 工作(Redis) + 长期(向量库)
│   ├── Tools：单一职责/明确Schema/有意义错误/超时/人工确认
│   └── Action：执行/解析/终止条件/死循环检测
│
├── ReAct 手写实现(重点！)
│   ├── Thought → Action → Observation 完整循环
│   ├── _parse_react_output 正则解析器
│   ├── _detect_loop 死循环检测
│   └── AgentStep 完整轨迹记录
│
└── 决策策略
    ├── ReAct: 灵活但串行（通用首选）
    ├── Plan-and-Execute: 高效但僵硬（明确任务用）
    ├── Reflexion: 从错误中学习（持续优化用）
    └── LATS: 最优但昂贵（关键任务用）

---

🔜 明日预告

Day 5 — RAG 检索增强生成全体系

• RAG 原理 & vs 微调/Long Context 全面对比
• Chunking 五策略：固定/递归/语义/父子文档/Small-to-Big
• Embedding 模型选型（BGE/GTE/Jina/OpenAI）& 向量数据库对比
• 混合检索（Dense + Sparse + BM25）+ Rerank + HyDE
• Lost in the Middle & 上下文压缩（LongLLMLingua）
• Agentic RAG：Self-RAG / Corrective RAG / 多跳检索
• 从零搭建完整 RAG 系统的完整代码
• 8 道 RAG 面试真题（含公司标签）

---

💡 速通心法：Day 4 的核心是把"Agent 到底怎么工作"刻进肌肉记忆。面试官问"你用过 Agent 吗"，不是在问你有没有调过 LangChain 的 agent.run()，而是在问你能不能讲清楚：LLM 怎么选工具？参数怎么校验？失败怎么兜底？记忆怎么跨对话保留？死循环怎么检测？把今天 ReAct 手写代码敲一遍，对着代码把 Thought→Action→Observation 循环讲三遍，这五个问题你就能一边画架构图一边回答了。