高级 Agent 架构

⚙️ AgentExecutor 是如何管理 Agent 的思考-行动循环的？内部有哪些关键逻辑？

AgentExecutor 是 LangChain 中运行 Agent 的“执行引擎”，它封装了 ReAct（Reasoning + Acting）循环，负责调度 Agent 的思考、行动和观察，直到任务完成或达到终止条件。

内部执行流程

AgentExecutor 的循环可以用以下流程图表示：

┌─────────────────────┐
│  用户输入 + 历史记忆  │
└─────────┬───────────┘
          ▼
┌──────────────────────────────┐
│ 1. 构建 Prompt                │
│ (系统指令 + 可用工具 + scratchpad) │
└─────────┬────────────────────┘
          ▼
┌──────────────────────────────┐
│ 2. 调用 LLM，获取生成文本      │
└─────────┬────────────────────┘
          ▼
┌──────────────────────────────┐
│ 3. 输出解析 (Output Parser)   │
│ 提取 Thought, Action, Action Input │
└─────────┬────────────────────┘
          ▼
      ┌───┴───┐
      │是最终答案?│
      └───┬───┘
          │ 是 → 返回最终答案
          │ 否
          ▼
┌──────────────────────────────┐
│ 4. 执行工具 (Tool Execution)   │
│ 得到 Observation               │
└─────────┬────────────────────┘
          ▼
┌──────────────────────────────┐
│ 5. 追加到 scratchpad          │
│ 回到步骤 1，继续循环           │
└──────────────────────────────┘

关键内部逻辑

• _take_next_step：核心方法，负责执行一步：调用 LLM、解析输出、执行工具（如果有 Action），返回 AgentFinish 或 AgentAction。

• Scratchpad 管理：所有中间步骤（Thought/Action/Observation）被拼接成 intermediate_steps 列表，在下一轮作为上下文注入 Prompt，让 Agent “记住”自己做了什么。

• 错误处理：如果输出解析失败，AgentExecutor 会捕获异常，将错误信息追加到 Prompt，让 LLM 修正输出（可通过 handle_parsing_errors 控制）。

• 终止条件：

• LLM 主动输出 Final Answer:。
• 达到 max_iterations（最大步数）。
• 自定义 stop 词触发。

• 工具返回 AgentFinish 信号（某些特殊工具）。

• 早停机制：达到最大步数后，若未完成，根据 early_stopping_method 决定是报错 (force) 还是让 LLM 强制总结 (generate)。

✅ 实践感悟：AgentExecutor 其实就是一个“while 循环”，但它的价值在于把错误处理、状态管理和终止逻辑都优雅地封装了。如果自己写循环，很容易遗漏边界情况。

---

🧠 在 AgentExecutor 中，如果 token 即将超出上下文窗口，它会做什么？

答案：默认情况下，LangChain 的 AgentExecutor 不会主动监控或动态截断 token 数量。 它没有内建的“token 预算管理器”。这意味着如果 Prompt（系统指令 + 工具描述 + 历史对话 + scratchpad）的总 token 数超出模型的最大上下文窗口，API 会直接返回错误（如 OpenAI 的 context_length_exceeded），Agent 执行失败。

规避策略

1. 使用摘要记忆或窗口记忆：最直接的方法。通过 ConversationSummaryBufferMemory 或 ConversationBufferWindowMemory 限制注入的历史长度，为 scratchpad 留出足够空间。

2. 自定义 Token 计数器与裁剪：

3. 可以在 AgentExecutor 的循环中，利用回调或包装器，在每次调用 LLM 前计算当前 Prompt 的 token 数（使用 tiktoken 或 get_num_tokens）。
4. 如果超过阈值，自动裁剪 scratchpad 中最旧的几轮交互，或对早期步骤进行摘要。
5. 示例思路：

if token_count > max_tokens * 0.9:
    # 保留最近 N 步的 scratchpad，更早的丢弃或用一句话概括
    truncated_scratchpad = summarize_old_steps(scratchpad)

1. 选择更大的上下文模型：如 gpt-4-1106-preview 支持 128K，可以解决大部分问题。

2. 优化 Prompt 设计：精简工具描述，移除不必要的示例，减少静态 Prompt 消耗。

✅ 经验之谈：在早期的 Agent 项目中，我经常遇到因为历史太长导致 API 报错的情况。后来直接用了 ConversationSummaryBufferMemory，并为 scratchpad 设置了最大保留步数（例如只保留最近 5 步），问题就基本消失了。如果你的 Agent 需要非常长的推理链，务必自己实现一个 token 监控器。

---

🔄 如何让 Agent 具备“自我反思”能力？在 LangChain 中可以实现 Self-Reflection 吗？

自我反思（Self-Reflection）是指 Agent 在执行过程中，能够评估自己的行为、识别错误并调整策略。LangChain 本身没有内建一个“Self-Reflection Agent”类，但你可以通过以下方式实现。

实现方法

1. 在 Prompt 中注入反思指令
2. 修改 Agent 的 Prompt，加入类似“每执行完一个工具，请思考：结果是否符合预期？如果不，你应该调整什么？”的指令。

3. 还可以加入一个专用的反思步骤，例如要求 Agent 每隔 N 步输出一次 Reflection:，评估进展。

4. 使用自定义的 AgentExecutor 循环

5. 继承 AgentExecutor，在每次工具执行后，额外调用一次 LLM 进行反思。这次反思的结果不作为 Action，而是追加到 scratchpad 中作为 Thought，指导后续推理。

6. 结合 Self-Refine 链

7. LangChain 提供了 SelfRefineChain，它是在生成答案后，让 LLM 对自己的答案进行反馈和优化。你可以将 Agent 的最终输出作为 Self-Refine 的输入，实现“事后反思”。

8. 实现类似 Reflexion 的模式（Self-RAG 风格）

9. 在 Agent 的输出中增加特殊的标记，如 [Relevant]、[Irrelevant]、[Correct]，让 LLM 在生成 Action 的同时评价工具结果。这需要修改 Prompt 和输出解析器。

10. 利用工具反馈进行被动反思

11. 工具返回的错误信息如果足够详细，Agent 可以根据它调整行为，这是一种被动的反思。

✅ 实践：我在一个数据分析 Agent 中，自己在 Prompt 里加了一句：“如果连续两次工具调用返回相似的结果，请反思你是否陷入了死循环，并尝试改变查询方式。”效果立竿见影，Agent 的思维链变得更加清晰，不容易卡死。

---

📋 什么是 Plan and Execute Agent？它的 Plan 阶段和 Execute 阶段分别做什么？

Plan and Execute Agent 是一种将复杂任务先规划、后执行的 Agent 模式。它不象 ReAct Agent 那样一边想一边做，而是像一个项目经理：先制定详细计划，然后逐步执行计划中的每个步骤。

两个阶段

• Plan 阶段（规划阶段）
• 输入：用户的目标。
• 输出：一个结构化的“计划”（Plan），通常是一个步骤列表，每个步骤描述一个子任务和需要调用的工具/参数。

• 这个阶段完全由 LLM 完成，不调用任何外部工具。

• Execute 阶段（执行阶段）

• 接收计划，然后按照顺序（或根据依赖）执行每个步骤。
• 执行每个步骤时，可以调用相应的工具，获取结果。
• 所有步骤执行完毕后，汇总结果，生成最终答案。

LangChain 中的实现

LangChain 提供了 PlanAndExecute 系列组件，核心类包括 Planner 和 Executor。例如：

from langchain.experimental.plan_and_execute import PlanAndExecute, load_agent_executor, load_chat_planner
planner = load_chat_planner(llm)
executor = load_agent_executor(llm, tools)
agent = PlanAndExecute(planner=planner, executor=executor)

Planner 会生成类似：

Step 1: 搜索北京天气
Step 2: 搜索北京室内活动
Step 3: 根据天气推荐活动

• Executor 会针对每个 Step 调用 Agent 来完成。

✅ 适用场景：任务步骤清晰、目标明确，且执行过程中不需要根据中间结果大幅调整计划的情况。对于探索性强、需要频繁调整策略的任务，ReAct 更灵活。

---

⚖️ 对比 Plan-and-Execute Agent 与 ReAct Agent，各自的优缺点。

查看内嵌表格

简单记忆：

• Plan-and-Execute：先画好地图，再按图索骥。适合已知路线。

• ReAct：走一步看一步，随时调整方向。适合未知环境。

✅ 选型建议：如果你的任务可以分解为多个相对独立的子任务（比如“收集数据→分析→生成图表”），Plan-and-Execute 能显著降低成本。但如果任务中间可能会遇到未知情况，ReAct 更稳妥。

---

🧩 你如何实现一个分层 Agent？即一个主 Agent 将任务拆解后派给多个子 Agent。

分层 Agent（Hierarchical Agent）的核心是一个“主管” Agent + 多个“专家” Agent。主管 Agent 负责理解用户意图，将任务拆解，分配给最合适的子 Agent，并汇总它们的结果。

实现方式

1. 将子 Agent 封装为 Tool 这是最简单、最直接的方式。每个专家 Agent 被包装成一个 Tool，主 Agent 像调用普通工具一样调用它们。

from langchain.tools import Tool

# 假设已有两个专家 Agent 的 executor
legal_expert = create_legal_agent()
medical_expert = create_medical_agent()

def call_legal_expert(query: str) -> str:
    return legal_expert.run(query)

def call_medical_expert(query: str) -> str:
    return medical_expert.run(query)

tools = [
    Tool(name="LegalExpert", func=call_legal_expert, description="处理法律问题"),
    Tool(name="MedicalExpert", func=call_medical_expert, description="处理医疗问题"),
]
# 主 Agent 使用这些工具
manager_agent = create_openai_functions_agent(llm, tools, prompt)

1. 自定义主管 Agent 的逻辑 如果子 Agent 的调用需要更复杂的上下文（如共享记忆），可以自定义 AgentExecutor，在其中实现任务分配逻辑。

2. 使用 Multi-Agent 框架（如 AutoGen）与 LangChain 集成

对于更复杂的协作，可以引入专门的多 Agent 框架，让 LangChain 的工具和链作为底层能力。

关键设计点

• 接口标准化：子 Agent 的输入输出应简单明了，推荐使用 JSON 格式。

• 错误隔离：子 Agent 失败时，应返回明确的错误信息，供主 Agent 决策。

• 记忆共享：如果需要，所有 Agent 可以共用同一个 RedisChatMessageHistory 或向量库。

✅ 实践：我做过一个企业信息助手，主 Agent 负责理解问题，如果是IT支持就转给 IT 专家 Agent，如果是HR政策就转给 HR 专家 Agent。用 Tool 包装子 Agent 非常有效，代码也简洁。

---

📡 在 LangChain 中，如何实现 Agent 之间的通信？你会用什么方式传递消息？

LangChain 没有内建的 Agent 间通信协议，但你可以通过共享 Memory 或消息队列来模拟通信。

常用方法

1. 通过共享 Memory 传递 所有 Agent 使用同一个 ChatMessageHistory 实例（或同一个 Redis key）。一个 Agent 写入消息，另一个 Agent 读取。这是最直接的“黑板模式”。

2. 通过 Tool 的输入输出传递 如分层 Agent 中，主管 Agent 调用子 Agent 时，通过工具的参数传递信息，子 Agent 的结果通过返回值传回。这是点对点通信。

3. 使用外部消息队列或数据库 在复杂的多 Agent 系统中，可以使用 Redis Pub/Sub、Kafka 等消息中间件。每个 Agent 订阅特定频道，接收任务并发布结果。这种方式解耦性最好。

4. 利用 LangChain 的 Callbacks 在 Agent 的关键节点（如工具调用前后），通过回调将消息写入共享存储，其他 Agent 可以监听这些变化。

传递消息的格式：建议使用 JSON，包含 from, to, task, payload, timestamp 等字段，便于扩展和追踪。

✅ 经验：对于简单的链式协作，用 Tool 传参就够了。只有需要多个 Agent 并行、异步通信时，才考虑引入消息队列。不要过度设计。

---

📊 什么是“结构化工具输出”？如何让 Agent 的工具输出 JSON 并供后续步骤解析？

结构化工具输出是指工具的返回值不是自然语言字符串，而是结构化的数据（如 JSON、YAML、XML），这样 Agent 或后续的链可以精确解析其中的字段，进行进一步处理。

实现方式

1. 工具直接返回 JSON 字符串 在工具函数内部，构造 JSON 并返回。然后在 Agent 的 Prompt 中说明这个工具返回的是 JSON，以便 Agent 知道可以从中提取字段。

import json
@tool
def get_user_profile(user_id: str) -> str:
    profile = {"name": "张三", "age": 30, "city": "北京"}
    return json.dumps(profile, ensure_ascii=False)

1. 使用 Pydantic 模型作为工具输出 LangChain 支持用 Pydantic 定义工具的 args_schema（输入），但输出目前仍以字符串为主。你可以通过自定义输出解析器，在 Agent 外部将返回的 JSON 解析为 Pydantic 对象。

2. 在 Agent 的 Prompt 中要求解析 告诉 LLM：“工具的返回值为 JSON 格式，你可以通过键名访问其中的数据”。LLM 能够理解并在 Thought 中引用这些键。

3. 结合 StructuredTool 和自定义解析 如果你希望工具的输出能够被后续链直接使用，可以将工具包装在一个 Tool 中，并在 func 返回时做结构化标记。或者使用 LangChain 的 JsonOutputToolsParser 等解析 Agent 的 Action。

✅ 应用：在一个金融分析 Agent 中，我让“数据查询”工具返回 JSON 数组，然后 Agent 在 Thought 中会写：“从结果中看，股价为 XX，涨跌幅为 YY”，这样后续的自然语言生成就非常精准。

---

🤝 当多个 Agent 协作时，如何避免它们产生冲突或重复工作？

多 Agent 协作常见的冲突包括：重复执行相同操作、结果不一致、资源竞争。避免冲突需要从架构设计和通信机制两方面入手。

预防措施

1. 明确职责边界：每个 Agent 的工具集和功能描述要有清晰界限，避免功能重叠。例如，一个 Agent 只能“读取”，另一个 Agent 只能“写入”。

2. 中央调度与状态管理：引入一个“协调者 Agent”或一个共享状态管理器（如 Redis），记录哪些任务已经被执行、谁在执行、进度如何。Agent 在执行前先检查状态，避免重复。

3. 任务分解与去重：在任务分解阶段，协调者就去除重复的子任务，并将任务唯一分配给特定 Agent。

4. 基于锁的并发控制：如果多个 Agent 可能修改同一资源，使用分布式锁（如 Redis Lock）确保同一时间只有一个 Agent 操作。

5. 日志与审计追踪：每个 Agent 的操作都记录到日志中，便于事后发现和解决冲突。

6. 让 Agent 之间“沟通”：Agent 可以向协调者报告自己正在处理的任务，协调者广播给其他 Agent，实现情报共享。

✅ 现实类比：就像项目团队，每个成员有明确的职责描述，任务分配由项目经理（协调者）统一调度，避免两个人做同一件事。多 Agent 系统也需要这样一个“项目经理”。

---

🤖 你怎样用 LangChain 搭建一个类似 AutoGPT 的自主智能体？核心组件有哪些？

AutoGPT 是一个能够自主设定目标、规划、执行、自我反思的智能体。使用 LangChain 搭建类似系统，核心在于循环控制、长期记忆、工具使用和自我反思。

核心组件

1. 一个强大的 LLM：作为大脑，负责推理、规划和决策。推荐 gpt-4 或同等性能模型。

2. AgentExecutor + ReAct/OpenAI Functions Agent：作为执行引擎，支持多步推理和工具调用。

3. 长期记忆系统：

4. VectorStoreRetrieverMemory：存储和检索历史对话、任务状态、知识片段。

5. ConversationSummaryBufferMemory：压缩对话历史，保持上下文连贯。

6. 丰富的工具集：

7. 网络搜索工具（Google/Bing API）
8. 文件读写工具
9. 代码执行工具（Python REPL）

10. 外部 API 工具

11. 自我反思机制：

12. 修改 Prompt，加入定期反思指令。
13. 使用自定义 AgentExecutor，在特定步骤后插入反思调用。

14. 集成 Self-Refine 链。

15. 任务管理器：

16. 一个数据结构（如 JSON）存储任务列表、优先级、完成状态。

17. 一个工具让 Agent 能够读取和更新任务列表。

18. 安全与终止机制：

19. 严格限制 Agent 的步数和操作权限。
20. 人工审批工具（如删除文件前需要确认）。

简易架构流程

用户目标 → Planner Agent (生成任务列表)
          → Executor Agent (循环执行任务)
              ←→ Memory (检索相关记忆)
              ←→ Tools (调用外部工具)
              ←→ Self-Reflection (评估进展)
          → 输出最终结果

✅ 实践感受：搭建 AutoGPT 类系统并不难，难的是如何控制成本和保证稳定性。我通常会给 Agent 设定一个“预算上限”（步数和 token 数），并且在执行危险操作时强制人工确认。另外，长期记忆用向量库非常关键，它能让 Agent 记住之前的推理和经验。

---