集成、元控制、思维树、反思性、PEV等模式详解

当你构建一个大型AI系统时，实际上是在将不同的智能体设计模式 (agentic design patterns) 组合在一起。每种模式都有其自己的阶段、构建方法、输出和评估方式。如果我们退后一步对这些模式进行归类，它们可以被分解为17种高级架构，这些架构捕捉了智能体系统可以采取的主要形式…

其中一些模式包括：

• • 多智能体系统 (Multi-Agent System)，多个工具和智能体协同工作以解决一个问题。
• • 集成决策系统 (Ensemble Decision System)，多个智能体各自提出一个答案，然后投票选出最佳答案。
• • 思维树 (Tree-of-Thoughts)，智能体在选择最有希望的方向之前，会探索多条不同的推理路径。
• • 反思性方法 (Reflexive approach)，智能体能够识别并承认其未知之处。
• • ReAct循环 (ReAct loop)，智能体在思考、采取行动、然后再次思考之间交替进行，以优化其流程。

在这篇博客中，我们将逐一解析这些不同类型的智能体架构，并展示每种架构在完整的AI系统中扮演的独特角色。

我们将直观地理解每种架构的重要性，编写其工作流的代码，并对其进行评估，以检验它是否真正比基线有所改进.

所所有内容都可以在如下 GitHub 仓库中找到：

https://github.com/FareedKhan-dev/all-agentic-architectures

代码库的组织结构如下：

all-agentic-architectures/    ├── 01_reflection.ipynb    ├── 02_tool_use.ipynb    ├── 03_ReAct.ipynb    ...    ├── 06_PEV.ipynb    ├── 07_blackboard.ipynb    ├── 08_episodic_with_semantic.ipynb    ├── 09_tree_of_thoughts.ipynb    ...    ├── 14_dry_run.ipynb    ├── 15_RLHF.ipynb    ├── 16_cellular_automata.ipynb    └── 17_reflexive_metacognitive.ipynb

环境设置

在开始构建每种架构之前，我们需要进行基础设置，并明确我们将使用什么、为什么使用这些模块、模型，以及它们如何组合在一起。

我们知道，LangChain、LangGraph 和 LangSmith 基本上是构建任何严肃的 RAG 或智能体系统的行业标准模块。它们为我们提供了构建、编排所需的一切，最重要的是，当事情变得复杂时，它们能帮助我们弄清楚智能体内部发生了什么。因此，我们选择使用这三者。

第一步是导入我们的核心库。这样可以避免后续重复操作，并保持设置整洁。

import osfrom typing importList, Dict, Any, Optional, Annotated, TypedDictfrom dotenv import load_dotenv  # 从 .env 文件加载环境变量# 用于数据建模/验证的 Pydanticfrom pydantic import BaseModel, Field# LangChain 和 LangGraph 组件from langchain_nebius import ChatNebius  # Nebius LLM 包装器from langchain_tavily import TavilySearch  # Tavily 搜索工具集成from langchain_core.prompts import ChatPromptTemplate  # 用于构建提示from langgraph.graph import StateGraph, END  # 构建状态机图from langgraph.prebuilt import ToolNode, tools_condition  # 预构建的节点和条件# 用于美化输出from rich.console import Console  # 控制台样式from rich.markdown import Markdown  # 在终端中渲染 markdown

让我们快速分析一下我们使用这三者的原因。

• • LangChain 是我们的工具箱，为我们提供了核心构建块，如提示、工具定义和 LLM 包装器。
• • LangGraph 是我们的编排引擎，将所有组件连接成具有循环和分支的复杂工作流。
• • LangSmith 是我们的调试器，它以可视化的方式追踪智能体所采取的每一步，使我们能够快速发现并修复问题。

我们将通过 Nebius AI 或 Together AI 等提供商使用开源 LLM。好处是它们的使用方式与标准的 OpenAI 模块类似，所以我们不需要做太多改动就能开始。如果我们想在本地运行，只需换成 Ollama 这样的工具即可。

为了确保我们的智能体不受限于静态数据，我们为它们提供了访问 Tavily API 的权限，以进行实时网络搜索（我认为每月1000个积分足够测试）。这样，它们就可以……

主动地去寻找信息，从而将重点放在真正的推理和工具使用上。

接下来，我们需要设置环境变量。这里我们将存放敏感信息，如 API 密钥。为此，在同一目录下创建一个名为 .env 的文件，并将你的密钥放入其中，例如：

# Nebius LLM 的 API 密钥（与 ChatNebius 一起使用）NEBIUS_API_KEY="your_nebius_api_key_here"# LangSmith 的 API 密钥（LangChain 的可观测性/遥测平台）LANGCHAIN_API_KEY="your_langsmith_api_key_here"# Tavily 搜索工具的 API 密钥（与 TavilySearch 集成一起使用）TAVILY_API_KEY="your_tavily_api_key_here"

一旦 .env 文件设置好，我们就可以使用之前导入的 dotenv 模块轻松地将这些密钥加载到代码中。

load_dotenv()  # 从 .env 文件加载环境变量# 启用 LangSmith 追踪以进行监控/调试os.environ["LANGCHAIN_TRACING_V2"] = "true"os.environ["LANGCHAIN_PROJECT"] = "Implementing 17 Agentic Architectures"  # 用于分组追踪的项目名称# 验证所有必需的 API 密钥是否可用for key in ["NEBIUS_API_KEY", "LANGCHAIN_API_KEY", "TAVILY_API_KEY"]:    if not os.environ.get(key):  # 如果在环境变量中未找到密钥        print(f"未找到 {key}。请创建一个 .env 文件并进行设置。")

现在我们的基本设置已经准备就绪，我们可以开始逐一构建每种架构，看看它们的表现如何，以及它们在大型AI系统中最适合的位置。

反思 (Reflection)

我们要研究的第一个架构是反思 (Reflection)。这可能是你在智能体工作流中看到的最常见和最基础的模式。

它的核心是赋予智能体一种能力，让它能够退后一步，审视自己的工作，并加以改进。

在大型AI系统中，这种模式非常适合任何对生成输出质量要求至关重要的阶段。例如，生成复杂的代码、撰写详细的技术报告等任务——在这些场景下，一个简单的初稿答案是远远不够的，并可能导致现实世界的问题。

让我们来理解一下这个流程。

反思智能体工作流 (由Fareed Khan创建)

1. 1. 生成 (Generate)： 智能体接收用户提示，并生成一个初步的草稿或解决方案。这是它的第一次、未经筛选的尝试。
2. 2. 批判 (Critique)： 接着，智能体转换角色，成为自己的批评者。它分析草稿中的缺陷，提出诸如“这是否正确？”、“这是否高效？”或“我遗漏了什么？”等问题。
3. 3. 优化 (Refine)： 最后，智能体根据自己的批判，生成一个最终的、改进版的输出，直接修正它所发现的缺陷。

在创建智能体的逻辑之前，我们需要定义它将使用的数据结构。使用 Pydantic 模型是一个很好的方法，可以强制 LLM 给出干净、结构化的 JSON 输出，这对于一个多步骤流程至关重要，因为上一步的输出会成为下一步的输入。

class DraftCode(BaseModel):    """用于智能体生成的初始代码草稿的模式。"""    code: str = Field(description="为解决用户请求而生成的Python代码。")  # 原始草稿    explanation: str = Field(description="关于代码工作原理的简要说明。")  # 推理classCritique(BaseModel):    """用于对生成代码进行自我批判的模式。"""    has_errors: bool = Field(description="代码是否存在潜在的错误或逻辑缺陷？")  # 错误检查    is_efficient: bool = Field(description="代码的编写方式是否高效和优化？")  # 性能检查    suggested_improvements: List[str] = Field(description="具体的、可操作的代码改进建议。")  # 具体修复建议    critique_summary: str = Field(description="批判的摘要。")  # 审查概述classRefinedCode(BaseModel):    """用于在采纳批判后最终优化代码的模式。"""    refined_code: str = Field(description="最终改进的Python代码。")  # 优化版本    refinement_summary: str = Field(description="基于批判所做更改的摘要。")  # 更改说明

有了这些模式，我们与 LLM 之间就有了一个明确的“契约”。Critique 模型尤其有用，因为它强制进行结构化审查，要求对错误和效率进行具体检查，而不仅仅是发出一个模糊的“审查代码”指令。

首先是我们的 generator_node。它唯一的工作就是接收用户请求并生成初稿。

def generator_node(state):    """生成代码的初始草稿。"""    console.print("--- 1. 正在生成初始草稿 ---")    # 初始化 LLM 以返回一个结构化的 DraftCode 对象    generator_llm = llm.with_structured_output(DraftCode)        prompt = f"""你是一位专业的Python程序员。请编写一个Python函数来解决以下请求。    提供一个简单、清晰的实现和一个解释。        请求: {state['user_request']}    """        draft = generator_llm.invoke(prompt)    return {"draft": draft.model_dump()}

这个节点将为我们提供初始的 draft 和它的 explanation，然后我们会将其传递给批判者进行审查。

现在是反思过程的核心——critic_node。在这里，智能体扮演一位高级开发人员的角色，对自己的工作进行严格的代码审查。

def critic_node(state):    """批判生成的代码，检查错误和低效之处。"""    console.print("--- 2. 正在批判草稿 ---")    # 初始化 LLM 以返回一个结构化的 Critique 对象    critic_llm = llm.with_structured_output(Critique)        code_to_critique = state['draft']['code']        prompt = f"""你是一位专业的代码审查员和资深Python开发人员。你的任务是对以下代码进行彻底的批判。        分析代码的以下方面：    1.  **错误与缺陷：** 是否存在任何潜在的运行时错误、逻辑缺陷或未处理的边界情况？    2.  **效率与最佳实践：** 这是解决问题的最有效方法吗？它是否遵循标准的Python约定（PEP 8）？        提供一个结构化的批判，附带具体的、可操作的建议。        待审查代码：    ```python    {code_to_critique}
``````plaintext
"""critique = critic_llm.invoke(prompt)return {"critique": critique.model_dump()}
``````plaintext
这里的输出是一个结构化的 `Critique` 对象。这比一个模糊的**“这个可以改进”**的消息要好得多，因为它为我们的下一步提供了具体的、可操作的反馈。我们逻辑的最后一部分是 `refiner_node`。它接收原始草稿和编辑的反馈，并创建最终的改进版本。```pythondef refiner_node(state):    """根据批判优化代码。"""    console.print("--- 3. 正在优化代码 ---")    # 初始化 LLM 以返回一个结构化的 RefinedCode 对象    refiner_llm = llm.with_structured_output(RefinedCode)        draft_code = state['draft']['code']    critique_suggestions = json.dumps(state['critique'], indent=2)        prompt = f"""你是一位专业的Python程序员，任务是根据一份批判来优化一段代码。        你的目标是重写原始代码，实施批判中所有建议的改进。        **原始代码：**    ```python    {draft_code}
``````plaintext
**批判与建议：**{critique_suggestions}请提供最终的、优化后的代码以及你所做更改的摘要。"""refined_code = refiner_llm.invoke(prompt)return {"refined_code": refined_code.model_dump()}
``````plaintext
好了，我们已经有了三个逻辑部分。现在，我们需要将它们连接成一个工作流。这就是 LangGraph 发挥作用的地方。我们将定义在节点之间传递的 `state`，然后构建图本身。```pythonclass ReflectionState(TypedDict):    """表示我们反思图的状态。"""    user_request: str    draft: Optional[dict]    critique: Optional[dict]    refined_code: Optional[dict]# 初始化一个新的状态图graph_builder = StateGraph(ReflectionState)# 将节点添加到图中graph_builder.add_node("generator", generator_node)graph_builder.add_node("critic", critic_node)graph_builder.add_node("refiner", refiner_node)# 将工作流的边定义为一个简单的线性序列graph_builder.set_entry_point("generator")graph_builder.add_edge("generator", "critic")graph_builder.add_edge("critic", "refiner")graph_builder.add_edge("refiner", END)# 将图编译成一个可运行的应用程序reflection_app = graph_builder.compile()

流程是一条简单的直线：generator -> critic -> refiner。这是经典的反思模式，现在我们准备好测试它了。

反思智能体 (由Fareed Khan创建)

为了测试这个工作流，我们将给它一个经典的编程问题，其中一个简单的初次尝试通常是低效的——找到第n个斐波那契数。这是一个完美的测试用例，因为一个简单的递归解决方案很容易编写，但计算成本高昂，留下了很大的改进空间。

user_request = "编写一个Python函数来找到第n个斐波那契数。"initial_input = {"user_request": user_request}console.print(f"[bold cyan]🚀 正在为请求启动反思工作流：[/bold cyan] '{user_request}'\n")# 流式传输结果并捕获最终状态final_state = Nonefor state_update in reflection_app.stream(initial_input, stream_mode="values"):    final_state = state_updateconsole.print("\n[bold green]✅ 反思工作流完成！[/bold green]")

让我们看看前后对比，了解我们的智能体做了什么。

--- ### 初始草稿 ---解释：此函数使用递归方法计算第n个斐波那契数... 对于较大的n值，由于重复计算，此方法效率不高...1deffibonacci(n):2     if n <= 0:3         return04     elif n == 1:5         return16     else:7         return fibonacci(n-1) + fibonacci(n-2)--- ### 批判 ---摘要：该函数存在潜在的错误和低效问题。应进行修订以处理负数输入并提高其时间复杂度。建议改进：- 该函数不能正确处理负数。- 由于重复计算，该函数的时间复杂度很高。考虑使用动态规划或记忆化。- 该函数不遵循PEP 8约定...--- ### 最终优化代码 ---优化摘要：原始代码已修订，以处理负数输入，提高其时间复杂度，并遵循PEP 8约定。1deffibonacci(n):2     """计算第n个斐波那契数。"""3     if n < 0:4         raise ValueError("n必须是一个非负整数")5     elif n == 0:6         return07     elif n == 1:8         return19     else:10         fib = [0, 1]11         for i inrange(2, n + 1):12             fib.append(fib[i-1] + fib[i-2])13         return fib[n]

初始草稿是一个简单的递归函数——正确，但效率极低。批判者指出了指数级复杂度和其它问题。而优化后的代码则是一个更智能、更健壮的迭代解决方案。这是该模式按预期工作的完美示例。

为了让这一点更具体，我们引入另一个LLM作为公正的“裁判”，对初始草稿和最终代码进行评分。这将为我们提供改进的量化衡量标准。

class CodeEvaluation(BaseModel):    """用于评估一段代码的模式。"""    correctness_score: int = Field(description="代码逻辑正确性的评分（1-10分）。")    efficiency_score: int = Field(description="代码算法效率的评分（1-10分）。")    style_score: int = Field(description="代码风格和可读性的评分（PEP 8）（1-10分）。")    justification: str = Field(description="对评分的简要说明。")def evaluate_code(code_to_evaluate: str):    prompt = f"""你是一位专业的Python代码评审员。请从正确性、效率和风格三个方面，以1-10分制评估以下函数，并提供简要说明。        代码：    ```python    {code_to_evaluate}
``````plaintext
"""return judge_llm.invoke(prompt)

if final_state and ‘draft’ in final_state and ‘refined_code’ in final_state:
console.print(“— 正在评估初始草稿 —”)
initial_draft_evaluation = evaluate_code(final_state[‘draft’][‘code’])
console.print(initial_draft_evaluation.model_dump()) # 已更正：使用 .model_dump()

console.print("\n--- 正在评估优化后的代码 ---")refined_code_evaluation = evaluate_code(final_state['refined_code']['refined_code'])console.print(refined_code_evaluation.model_dump()) # 已更正：使用 .model_dump()

else:
console.print(“[bold red]错误：无法进行评估，因为final_state不完整。[/bold red]”)

当我们让裁判对两个版本进行评判时，分数说明了一切。```python--- 正在评估初始草稿 ---{    'correctness_score': 2,    'efficiency_score': 4,    'style_score': 2,    'justification': '该函数的时间复杂度为O(2^n)...'}--- 正在评估优化后的代码 ---{    'correctness_score': 8,    'efficiency_score': 6,    'style_score': 9,    'justification': '代码是正确的... 它的时间复杂度为O(n)...'}

初始草稿得分很差，尤其是在效率方面。优化后的代码在各个方面都有了巨大的提升。

这为我们提供了确凿的证据，证明反思过程不仅仅是改变了代码——它使其变得更好。

一直在更新，更多的大模型学习和面试资料已经上传带到CSDN的官方了，有需要的朋友可以扫描下方二维码免费领取【保证100%免费】👇👇

工具使用 (Tool Using)

我们刚刚构建的反思模式非常适合提升智能体的内部推理能力。

但是，当智能体需要它尚不知道的信息时会发生什么呢？

如果没有访问外部工具的权限，LLM 就只能局限于其预训练的参数。它能生成和推理，但无法查询新数据或与外部世界互动。这就是我们的第二种架构——工具使用 (Tool Use)——发挥作用的地方。

在任何大型AI系统中，工具使用不是可选项，而是重要且必需的组件。它充当了智能体推理与现实世界数据之间的桥梁。无论是支持机器人查询订单状态，还是金融智能体获取实时股价，都离不开它。

让我们来理解一下这个流程。

工具使用工作流 (由Fareed Khan创建)

1. 1. 接收查询 (Receive Query)： 智能体从用户那里接收到一个请求。
2. 2. 决策 (Decision)： 智能体分析查询并查看其可用的工具。然后它决定是否需要使用工具来准确回答问题。
3. 3. 行动 (Action)： 如果需要工具，智能体将格式化对该工具的调用，例如，一个带有正确参数的特定函数。
4. 4. 观察 (Observation)： 系统执行工具调用，结果（即“观察”）被返回给智能体。
5. 5. 综合 (Synthesis)： 智能体接收工具的输出，结合自身的推理，生成一个最终的、有根据的答案给用户。

要构建这个，我们需要给我们的智能体一个工具。为此，我们将使用 TavilySearchResults 工具，它让我们的智能体能够搜索网络。这里最重要的部分是描述。LLM 通过阅读这个自然语言描述来了解工具的功能以及何时应该使用它，所以清晰准确的描述是关键。

# 初始化工具。我们可以设置最大结果数以保持上下文简洁。search_tool = TavilySearchResults(max_results=2)# 为工具提供清晰的名称和描述对于智能体至关重要search_tool.name = "web_search"search_tool.description = "一个可用于在互联网上搜索任何主题的最新信息的工具，包括新闻、事件和时事。"tools = [search_tool]

现在我们有了一个功能性工具，可以开始构建将学习如何使用它的智能体了。一个使用工具的智能体的状态非常简单，它只是一个消息列表，用于跟踪整个对话历史。

class AgentState(TypedDict):    messages: Annotated[list[AnyMessage], add_messages]

接下来，我们必须让 LLM **“意识到”**我们给它的工具。这是一个关键步骤。我们使用 .bind_tools() 方法，它实质上是将工具的名称和描述注入到 LLM 的系统提示中，使其能够决定何时调用该工具。

llm = ChatNebius(model="meta-llama/Meta-Llama-3.1-8B-Instruct", temperature=0)# 将工具绑定到LLM，使其具备工具意识llm_with_tools = llm.bind_tools(tools)

现在我们可以使用 LangGraph 定义我们的智能体工作流。我们需要两个主要节点：agent_node（“大脑”），它调用 LLM 来决定下一步做什么；以及 tool_node（“手”），它实际执行工具。

def agent_node(state: AgentState):    """调用LLM以决定下一步行动的主节点。"""    console.print("--- 智能体：思考中... ---")    response = llm_with_tools.invoke(state["messages"])    return {"messages": [response]}# ToolNode 是 LangGraph 中一个预构建的节点，用于执行工具tool_node = ToolNode(tools)

在 agent_node 运行之后，我们需要一个路由器来决定下一步去哪里。如果智能体的最后一条消息包含 tool_calls 属性，这意味着它想使用一个工具，所以我们路由到 tool_node。如果没有，则意味着智能体已经有了最终答案，我们可以结束工作流。

def router_function(state: AgentState) -> str:    """检查智能体的最后一条消息以决定下一步。"""    last_message = state["messages"][-1]    if last_message.tool_calls:        # 智能体请求调用工具        console.print("--- 路由器：决定调用工具。---")        return "call_tool"    else:        # 智能体已提供最终答案        console.print("--- 路由器：决定结束。---")        return "__end__"

好了，我们有了所有的部分。让我们把它们连接成一个图。这里的关键是使用我们的 router_function 的条件边，它创建了智能体的主要推理循环：agent -> router -> tool -> agent。

graph_builder = StateGraph(AgentState)# 添加节点graph_builder.add_node("agent", agent_node)graph_builder.add_node("call_tool", tool_node)# 设置入口点graph_builder.set_entry_point("agent")# 添加条件路由器graph_builder.add_conditional_edges(    "agent",    router_function,)# 添加从工具节点返回到智能体的边以完成循环graph_builder.add_edge("call_tool", "agent")# 编译图tool_agent_app = graph_builder.compile()

工具调用架构 (由Fareed Khan创建)

现在我们来测试一下。我们将给它一个它不可能从训练数据中知道的问题，迫使它使用网络搜索工具来寻找实时答案。

user_query = "苹果最新的WWDC活动有哪些主要发布？"initial_input = {"messages": [("user", user_query)]}console.print(f"[bold cyan]🚀 正在为请求启动工具使用工作流：[/bold cyan] '{user_query}'\n")for chunk in tool_agent_app.stream(initial_input, stream_mode="values"):    chunk["messages"][-1].pretty_print()    console.print("\n---\n")console.print("\n[bold green]✅ 工具使用工作流完成！[/bold green]")

让我们看一下输出来了解智能体的思考过程。

================================= Human Message =================================苹果最新的WWDC活动有哪些主要发布？------ 智能体：思考中... ------ 路由器：决定调用工具。---================================== Ai Message ==================================工具调用：  web_search (call_abc123)  参数：    query: Apple WWDC latest announcements---================================= Tool Message =================================名称：web_search[{"title": "WWDC 2025: 我们所知的一切...", "url": "...", "content": "苹果的活动持续了一个小时... 我们回顾了所有的发布... iOS 26, iPadOS 26, macOS Tahoe..."}]------ 智能体：思考中... ------ 路由器：决定结束。---================================== Ai Message ==================================苹果最新WWDC活动的主要发布包括一个将指导未来十年iOS、iPadOS和macOS发展的新设计，iPhone的新功能... 以及每个平台的更新，包括iOS 26、iPadOS 26、CarPlay、macOS Tahoe...

追踪记录清楚地显示了智能体的逻辑：

1. 1. 首先，agent_node 思考并决定需要搜索网络，输出一个 tool_calls 请求。
2. 2. 接着，tool_node 执行该搜索并返回一个带有原始网络结果的 ToolMessage。
3. 3. 最后，agent_node 再次运行，这次将搜索结果作为上下文来综合一个最终的、有帮助的答案给用户。

为了使这个过程规范化，我们将再次引入我们的“LLM即裁判”，但这次的标准专门用于评估工具的使用。

class ToolUseEvaluation(BaseModel):    """用于评估智能体工具使用和最终答案的模式。"""    tool_selection_score: int = Field(description="智能体是否为任务选择了正确工具的评分（1-5分）。")    tool_input_score: int = Field(description="工具输入的格式和相关性评分（1-5分）。")    synthesis_quality_score: int = Field(description="智能体将工具输出整合到最终答案中的质量评分（1-5分）。")    justification: str = Field(description="对评分的简要说明。")

当我们对完整的对话追踪进行评判时，我们会得到一个结构化的评估。

--- 正在评估工具使用性能 ---{    'tool_selection_score': 5,    'tool_input_score': 5,    'synthesis_quality_score': 4,    'justification': "AI智能体正确地使用了网络搜索工具来寻找相关信息... 工具输出的格式良好且相关... AI智能体在综合信息方面本可以做得更好..."}

高分证明了我们的智能体不仅仅是在调用工具，而是在有效地使用它。

它正确地识别了何时搜索、搜索什么以及如何使用结果。这种架构是几乎所有实用AI助手的基础构建块。

ReAct (推理 + 行动)

我们上一个智能体是一大进步。它可以使用工具获取实时数据，这非常重要。但问题是，它有点像一次性操作：它决定需要一个工具，调用一次，然后尝试回答。

但是，当问题更复杂，需要多个相互依赖的步骤来解决时，会发生什么呢？

ReAct (推理 + 行动) (Reason + Act) 的核心是创建一个循环。它使智能体能够动态地推理下一步该做什么，采取一个行动（比如调用工具），观察结果，然后利用这些新信息再次推理。这是从一个静态的工具调用者到一个自适应的问题解决者的转变。

在任何AI系统中，ReAct都是处理需要多跳推理任务的首选模式。

让我们来理解一下这个流程。

ReAct工作流 (由Fareed Khan创建)

1. 1. 接收目标 (Receive Goal)： 智能体被赋予一个无法一步解决的复杂任务。
2. 2. 思考 (推理) (Think (Reason))： 智能体产生一个想法，比如：“要回答这个问题，我首先需要找到信息X”。
3. 3. 行动 (Act)： 基于这个想法，它执行一个行动，比如调用搜索工具查找“X”。
4. 4. 观察 (Observe)： 智能体从工具那里得到“X”的结果。
5. 5. 重复 (Repeat)： 它利用这些新信息回到第2步，思考：“好了，现在我有了X，我需要用它来找到Y”。这个循环一直持续到最终目标达成。

好消息是，我们已经构建了大部分组件。我们将重用 AgentState、web_search 和 tool_node。我们图逻辑中唯一的变化是：在 tool_node 运行后，我们将输出返回给 agent_node 而不是结束。这就创建了推理循环，让智能体能够审查结果并选择下一步。

def react_agent_node(state: AgentState):    """思考并决定下一步的智能体节点。"""    console.print("--- REACT 智能体：思考中... ---")    response = llm_with_tools.invoke(state["messages"])    return {"messages": [response]}# 具有其标志性循环的ReAct图react_graph_builder = StateGraph(AgentState)react_graph_builder.add_node("agent", react_agent_node)react_graph_builder.add_node("tools", tool_node) # 我们重用之前的tool_nodereact_graph_builder.set_entry_point("agent")react_graph_builder.add_conditional_edges(    "agent",    # 我们可以重用相同的路由器函数    router_function,    # 映射现在定义了路由逻辑    {"call_tool": "tools", "__end__": "__end__"})# 这是关键的区别：边从工具节点返回到智能体react_graph_builder.add_edge("tools", "agent")react_agent_app = react_graph_builder.compile()

ReAct架构 (由Fareed Khan创建)

就是这样。唯一真正的改变是 react_graph_builder.add_edge("tools", "agent")。这一行代码创建了循环，将我们简单的工具使用者变成了一个动态的 ReAct 智能体。

为了看看这个循环为什么如此强大，让我们给它一个不可能一次性解决的任务——一个经典的多跳问题。一个简单的使用工具的智能体会失败，因为它无法将步骤链接起来。

multi_step_query = "制作科幻电影《沙丘》的公司的现任CEO是谁，以及该公司最近一部电影的预算是多少？"console.print(f"[bold yellow]正在对ReAct智能体进行多步查询测试：[/bold yellow] '{multi_step_query}'\n")final_react_output = None# 流式输出以查看逐步推理过程for chunk in react_agent_app.stream({"messages": [("user", multi_step_query)]}, stream_mode="values"):    final_react_output = chunk    console.print(f"--- [bold purple]当前状态更新[/bold purple] ---")    chunk['messages'][-1].pretty_print()    console.print("\n")console.print("\n--- [bold green]ReAct智能体的最终输出[/bold green] ---")console.print(Markdown(final_react_output['messages'][-1].content))
``````plaintext
--- Human Message ---  制作《沙丘》的公司的CEO是谁...  --- REACT 智能体：思考中... ---  --- 路由器：调用工具...  --- Ai Message ---  工具调用：web_search...  参数：query: current CEO of company that made Dune...  --- Tool Message ---  [{"title": "沙丘：第三部 - 维基百科", "content": "传奇影业CEO Joshua Grode..."}]  --- REACT 智能体：思考中... ---  --- 路由器：结束...  --- 最终输出 ---  制作《沙丘》的公司的CEO是Joshua Grode...  最近一部电影的预算未找到...

当我们运行这个时，智能体的执行追踪显示了一个远比之前智能的过程。它不只是进行一次搜索，而是通过推理来解决问题：

1. 1. 想法1： “首先，我需要找出是哪家公司制作了电影《沙丘》。”
2. 2. 行动1： 调用 web_search('production company for Dune movie')。
3. 3. 观察1： 得到“传奇影业 (Legendary Entertainment)”。
4. 4. 想法2： “好的，现在我需要传奇影业的CEO。”
5. 5. 行动2： 调用 web_search('CEO of Legendary Entertainment')，依此类推，直到它集齐所有信息。

为了使改进规范化，我们可以再次使用我们的“LLM即裁判”，这次的重点是任务完成度。

class TaskEvaluation(BaseModel):    """用于评估智能体完成任务能力的模式。"""    task_completion_score: int = Field(description="智能体是否成功完成了用户请求的所有部分的评分（1-10分）。")    reasoning_quality_score: int = Field(description="智能体展示的逻辑流程和推理过程的质量评分（1-10分）。")    justification: str = Field(description="对评分的简要说明。")def evaluate_agent_output(query: str, agent_output: dict):    """运行一个“LLM即裁判”来评估智能体的最终表现。"""    trace = "\n".join([f"{m.type}: {m.content}" for m in agent_output['messages']])        prompt = f"""你是一位专业的AI智能体评审员。请根据给定任务，以1-10分制评估以下智能体的表现。10分表示任务完美完成，1分表示完全失败。        **用户任务：**    {query}        **完整的智能体对话追踪：**
``````plaintext
{trace}```"""judge_llm = llm.with_structured_output(TaskEvaluation)return judge_llm.invoke(prompt)
``````plaintext
让我们看看分数。一个尝试此任务的基础智能体会得到很低的分数，因为它无法收集所有必需的信息。然而，我们的 ReAct 智能体表现得好得多。```python--- 正在评估ReAct智能体的输出 ---{    'task_completion_score': 8,    'reasoning_quality_score': 9,    'justification': "该智能体正确地将问题分解为多个步骤... 它成功地识别了公司，然后是CEO。虽然它在寻找最近一部电影的预算时遇到了困难，但其推理过程是合理的，并且完成了大部分任务。"}

我们可以看到，reasoning_quality_score 确认了它的逐步过程是合乎逻辑的，并得到了我们的裁判（LLM）的验证。

ReAct模式赋予了智能体处理这类需要动态思维的复杂、多跳问题的能力。

规划 (Planning)

ReAct 模式非常适合在处理问题时边做边学。但对于步骤可预测的任务，它可能有点低效。这就像一个人每次只问一个转弯的路线，而不是先看一眼完整的地图。这就是规划 (Planning) 架构发挥作用的地方。

这种模式引入了一个至关重要的前瞻层。

与其一步步地做出反应，一个规划智能体会在采取任何行动之前，首先创建一个完整的“作战计划”。

在AI系统中，对于任何结构化的、多步骤的过程，规划都是你的主力。比如数据处理管道、报告生成，或者任何你知道操作顺序的工作流。它带来了可预测性和效率，使智能体的行为更易于追踪和调试。

让我们来理解一下这个流程。

规划方法 (由Fareed Khan创建)

1. 1. 接收目标 (Receive Goal)： 智能体被赋予一个复杂的任务。
2. 2. 规划 (Plan)： 一个专门的“规划器”组件分析目标，并生成一个实现该目标所需子任务的有序列表。例如：["查找事实A", "查找事实B", "使用A和B计算C"]。
3. 3. 执行 (Execute)： 一个“执行器”组件接收该计划，并按顺序执行每个子任务，根据需要使用工具。
4. 4. 综合 (Synthesize)： 一旦计划中的所有步骤都完成，一个最终组件会将已执行步骤的结果组合成一个连贯的最终答案。

让我们开始构建它。

我们将创建三个核心组件：一个 planner_node 来创建策略，一个 executor_node 来执行它，以及一个 synthesizer_node 来组装最终报告。

首先，我们需要一个专门的 planner_node。这里的关键是一个非常明确的提示，告诉 LLM 它的工作是创建一个简单、可执行的步骤列表。

class Plan(BaseModel):    """一个为回答用户查询而执行的工具调用计划。"""    steps: List[str] = Field(description="一个工具调用列表，执行后将回答查询。")classPlanningState(TypedDict):    user_request: str    plan: Optional[List[str]]    intermediate_steps: List[str] # 将存储工具输出    final_answer: Optional[str]defplanner_node(state: PlanningState):    """生成一个行动计划以回答用户的请求。"""    console.print("--- 规划器：正在分解任务... ---")    planner_llm = llm.with_structured_output(Plan)        prompt = f"""你是一位专家规划师。你的工作是创建一个分步计划来回答用户的请求。        计划中的每一步都必须是对 `web_search` 工具的一次调用。        **用户请求：**        {state['user_request']}    """    plan_result = planner_llm.invoke(prompt)    console.print(f"--- 规划器：生成的计划：{plan_result.steps} ---")    return {"plan": plan_result.steps}

接下来是 executor_node。这是一个简单直接的工作者，它只是从计划中取出下一步，运行工具，并将结果添加到我们的状态中。

def executor_node(state: PlanningState):    """执行计划中的下一步。"""    console.print("--- 执行器：正在运行下一步... ---")    next_step = state["plan"][0]        # 在真实应用中，你会解析工具名称和参数。这里我们假设是'web_search'。    query = next_step.replace("web_search('", "").replace("')", "")    result = search_tool.invoke({"query": query})        return {        "plan": state["plan"][1:], # 弹出已执行的步骤        "intermediate_steps": state["intermediate_steps"] + [result]    }

现在我们只需要将它们连接成一个图。一个路由器将检查计划中是否还有剩余的步骤。如果有，它会循环回到执行器。如果没有，它会进入一个最终的 synthesizer_node（我们可以重用之前模式中的）来生成答案。

def planning_router(state: PlanningState):    """根据计划路由到执行器或综合器。"""    ifnot state["plan"]:        console.print("--- 路由器：计划完成。转向综合器。---")        return"synthesize"    else:        console.print("--- 路由器：计划还有更多步骤。继续执行。---")        return"execute"planning_graph_builder = StateGraph(PlanningState)planning_graph_builder.add_node("plan", planner_node)planning_graph_builder.add_node("execute", executor_node)planning_graph_builder.add_node("synthesize", synthesizer_node)planning_graph_builder.set_entry_point("plan")planning_graph_builder.add_conditional_edges("plan", planning_router)planning_graph_builder.add_conditional_edges("execute", planning_router)planning_graph_builder.add_edge("synthesize", END)planning_agent_app = planning_graph_builder.compile()

规划 (由Fareed Khan创建)

为了真正看到区别，让我们给我们的智能体一个需要前瞻性的任务。一个 ReAct 智能体可以解决这个问题，但它的逐步过程不那么直接。

plan_centric_query = """找出法国、德国和意大利首都的人口。然后计算它们的总和。"""console.print(f"[bold green]正在对PLANNING智能体进行以计划为中心的查询测试：[/bold green] '{plan_centric_query}'\n")# 正确初始化状态，特别是中间步骤的列表initial_planning_input = {"user_request": plan_centric_query, "intermediate_steps": []}final_planning_output = planning_agent_app.invoke(initial_planning_input)console.print("\n--- [bold green]规划智能体的最终输出[/bold green] ---")console.print(Markdown(final_planning_output['final_answer']))

过程中的差异立即显现。我们的智能体做的第一件事就是制定出它的整个策略。

--- 规划器：正在分解任务... ------ 规划器：生成的计划：["web_search('巴黎人口')", "web_search('柏林人口')", "web_search('罗马人口')"] ------ 路由器：计划还有更多步骤。继续执行。------ 执行器：正在运行下一步... ---...

智能体在采取任何行动之前就创建了一个完整、明确的计划。然后它有条不紊地执行了这个计划。这个过程更加透明和健壮，因为它遵循了一套清晰的指令。

为了使这一点规范化，我们将使用我们的“LLM即裁判”，但这次我们将对过程的效率进行评分。

class ProcessEvaluation(BaseModel):    """用于评估智能体解决问题过程的模式。"""    task_completion_score: int = Field(description="任务完成度的评分（1-10分）。")    process_efficiency_score: int = Field(description="智能体过程的效率和直接性的评分（1-10分）。")    justification: str = Field(description="对评分的简要说明。")

在评判时，规划智能体在其直接性方面表现出色。

--- 正在评估规划智能体的过程 ---{    'task_completion_score': 8,    'process_efficiency_score': 9,    'justification': "该智能体预先制定了一个清晰、优化的计划，并毫无多余步骤地执行了它。对于这个可预测的任务，它的过程非常直接和高效。"}

我们得到了不错的分数，这意味着我们的方法确实创建了一个正确的规划系统，所以……

当解决方案路径可预测时，规划提供了一种比纯粹反应式方法更结构化和高效的方法。

PEV (规划器-执行器-验证器)

我们的规划智能体在路径清晰时工作得很好，它制定计划并遵循它。但这里有一个隐藏的假设……

当事情出错时会发生什么？如果一个工具失败，一个API宕机，或者搜索返回了垃圾信息，一个标准的规划器只会将错误传递下去，最终导致失败或无意义的结果。

PEV (规划器-执行器-验证器) (Planner-Executor-Verifier) 架构是对规划模式的一个简单但强大的升级，它增加了一个关键的质量控制和自我纠正层。

PEV对于构建健壮和可靠的工作流至关重要。你可以在任何智能体与可能不可靠的外部工具交互的地方使用它。

它的工作原理是这样的……

PEV (由Fareed Khan创建)

1. 1. 规划 (Plan)： 一个“规划器”智能体创建一系列步骤，就像以前一样。
2. 2. 执行 (Execute)： 一个“执行器”智能体从计划中取出下一步并调用工具。
3. 3. 验证 (Verify)： 这是新的一步。一个“验证器”智能体检查工具的输出。它检查正确性、相关性和错误。
4. 4. 路由与迭代 (Route & Iterate)： 根据验证器的判断：

• • 如果步骤成功，智能体将移至计划中的下一步。
• • 如果步骤失败，智能体将返回到规划器以创建一个新计划，此时它已经意识到了失败。
• • 如果计划完成，它将继续到结尾。

为了真正展示这一点，我们需要一个实际上会失败的工具。所以，我们将创建一个特殊的 flaky_web_search 工具，我们特意设计它在特定查询时返回一个错误消息。

def flaky_web_search(query: str) -> str:    """一个特意为特定查询失败的网络搜索工具。"""    console.print(f"--- 工具：正在搜索 '{query}'... ---")    if "employee count" in query.lower():        console.print("--- 工具：[bold red]模拟API失败！[/bold red] ---")        return "错误：无法检索数据。API端点当前不可用。"    else:        return search_tool.invoke({"query": query})

现在是PEV模式的核心：verifier_node。这个节点唯一的工作就是查看上一个工具的输出，并判断它是成功还是失败。

class VerificationResult(BaseModel):    """验证器输出的模式。"""    is_successful: bool = Field(description="如果工具执行成功且数据有效，则为True。")    reasoning: str = Field(description="验证决策的理由。")defverifier_node(state: PEVState):    """检查最后一个工具结果是否有错误。"""    console.print("--- 验证器：正在检查最后一个工具结果... ---")    verifier_llm = llm.with_structured_output(VerificationResult)    prompt = f"验证以下工具输出是否为成功、有效的结果或错误消息。任务是'{state['user_request']}'。\n\n工具输出：'{state['last_tool_result']}'"    verification = verifier_llm.invoke(prompt)    console.print(f"--- 验证器：判断结果是'{'成功' if verification.is_successful else '失败'}' ---")    if verification.is_successful:        return {"intermediate_steps": state["intermediate_steps"] + [state['last_tool_result']]}    else:        # 如果失败，我们添加失败原因并清除计划以触发重新规划        return {"plan": [], "intermediate_steps": state["intermediate_steps"] + [f"验证失败：{state['last_tool_result']}"]}

验证器准备好后，我们就可以连接完整的图了。这里的关键是路由器逻辑。在 verifier_node 运行后，如果计划突然变空（因为验证器清除了它），我们的路由器就知道要将智能体送回 planner_node 再试一次。

class PEVState(TypedDict):    user_request: str    plan: Optional[List[str]]    last_tool_result: Optional[str]    intermediate_steps: List[str]    final_answer: Optional[str]    retries: int# ... (planner_node, executor_node, and synthesizer_node 的定义与之前类似) ...defpev_router(state: PEVState):    """根据验证和计划状态路由执行。"""    ifnot state["plan"]:        # 检查计划是否为空是因为验证失败        if state["intermediate_steps"] and"验证失败"in state["intermediate_steps"][-1]:            console.print("--- 路由器：验证失败。正在重新规划... ---")            return"plan"        else:            console.print("--- 路由器：计划完成。转向综合器。---")            return"synthesize"    else:        console.print("--- 路由器：计划还有更多步骤。继续执行。---")        return"execute"pev_graph_builder = StateGraph(PEVState)pev_graph_builder.add_node("plan", pev_planner_node)pev_graph_builder.add_node("execute", pev_executor_node)pev_graph_builder.add_node("verify", verifier_node)pev_graph_builder.add_node("synthesize", synthesizer_node)pev_graph_builder.set_entry_point("plan")pev_graph_builder.add_edge("plan", "execute")pev_graph_builder.add_edge("execute", "verify")pev_graph_builder.add_conditional_edges("verify", pev_router)pev_graph_builder.add_edge("synthesize", END)pev_agent_app = pev_graph_builder.compile()

PEV架构 (由Fareed Khan创建)

现在是关键的测试。我们将给我们的PEV智能体一个任务，它需要调用我们知道会失败的 flaky_web_search 工具。一个简单的规划器-执行器智能体在这里会崩溃。

flaky_query = "苹果公司上个财年的研发支出是多少，以及他们的总员工数是多少？计算每位员工的研发支出。"console.print(f"[bold green]正在对PEV智能体进行有问题的查询测试：[/bold green]\n'{flaky_query}'\n")initial_pev_input = {"user_request": flaky_query, "intermediate_steps": [], "retries": 0}final_pev_output = pev_agent_app.invoke(initial_pev_input)console.print("\n--- [bold green]PEV智能体的最终输出[/bold green] ---")console.print(Markdown(final_pev_output['final_answer']))
``````plaintext
正在对同一个有问题的查询测试PEV智能体：'苹果公司上个财年的研发支出是多少，以及他们的总员工数是多少？计算每位员工的研发支出。'--- (PEV) 规划器：正在创建/修订计划（重试 0）... ------ (PEV) 执行器：正在运行下一步... ------ 工具：正在搜索 '苹果公司上个财年研发支出'... ------ 验证器：正在检查最后一个工具结果... ------ 验证器：判断结果是'成功' ---...

1. 1. 计划1： 智能体创建了一个计划：["苹果研发支出...", "苹果总员工数"]。
2. 2. 执行与失败： 它得到了研发支出，但 员工数 的搜索碰到了我们有问题的工具并返回了一个错误。
3. 3. 验证与捕获： verifier_node 得到了错误消息，正确地判断为失败，并清除了计划。
4. 4. 路由与重新规划： 路由器看到了空的计划和失败消息，并将智能体送回 planner_node。
5. 5. 计划2： 规划器现在意识到“苹果总员工数”失败了，于是创建了一个新的、更聪明的计划，可能尝试 web_search('全球苹果员工数量')。
6. 6. 执行与成功： 这个新计划成功了，智能体得到了它需要的所有数据。

最终的输出是正确的计算结果。智能体没有放弃；它检测到了问题，想出了一个新方法，并成功了。

为了使其规范化，我们的“LLM即裁判”需要对健壮性进行评分。

class RobustnessEvaluation(BaseModel):    """用于评估智能体健壮性和错误处理能力的模式。"""    task_completion_score: int = Field(description="任务完成度的评分（1-10分）。")    error_handling_score: int = Field(description="智能体检测和从错误中恢复的能力的评分（1-10分）。")    justification: str = Field(description="对评分的简要说明。")

一个标准的规划器-执行器智能体会得到一个很差的 error_handling_score。然而，我们的PEV智能体表现出色。

--- 正在评估PEV智能体的健壮性 ---{    'task_completion_score': 8,    'error_handling_score': 10,    'justification': "该智能体表现出完美的健壮性。它利用其验证器成功识别了工具失败，触发了重新规划循环，并制定了一个新的查询来规避问题。这是一个错误恢复的典范案例。"}

你可以看到，PEV架构不仅仅是在一切顺利时得到正确答案……

它的意义在于，当事情出错时，不会得到错误的答案。

思维树 (Tree-of-Thoughts, ToT)

PEV模式可以处理工具失败并尝试新计划。但其规划本身仍然是线性的。它创建一个单一的、按部就班的计划并遵循它。

当问题不是一条直路，而更像一个有死胡同和多条可能路径的迷宫时，会发生什么？

这就是思维树 (Tree-of-Thoughts, ToT) 架构发挥作用的地方。ToT智能体不是生成单一的推理线，而是同时探索多条路径。它生成几个可能的下一步，对它们进行评估，丢弃坏的，并继续探索最有希望的分支。

在大型AI系统中，ToT有助于解决诸如路径规划、调度或棘手的谜题等难题。

一个正常的ToT过程是这样工作的……

ToT过程 (由Fareed Khan创建)

1. 1. 分解 (Decomposition)： 问题被分解为一系列步骤或“想法”。
2. 2. 想法生成 (Thought Generation)： 对于当前状态，智能体生成多个潜在的下一步。这就创建了我们“思想树”的分支。
3. 3. 状态评估 (State Evaluation)： 每个新的潜在步骤都由一个批判者或验证函数进行评估。这个检查确定一个移动是否有效，是否在取得进展，或者只是在原地打转。
4. 4. 修剪与扩展 (Pruning & Expansion)： 智能体然后“修剪”掉坏的分支（无效或无前景的），并从剩下的好分支继续这个过程。
5. 5. 解决方案 (Solution)： 这个过程持续进行，直到其中一个分支达到最终目标。

为了展示ToT，我们需要一个无法用直线思维解决的问题——狼、羊和白菜过河的谜题。它非常适合，因为它需要非显而易见的移动（比如带回某些东西）并且有无效状态，可能会困住一个简单的推理器。

首先，我们将用一个Pydantic模型来定义谜题的规则，包括状态和一个检查有效性的函数（这样就不会有东西被吃掉）。

class PuzzleState(BaseModel):    "表示狼、羊和白菜谜题的状态。"    left_bank: set[str] = Field(default_factory=lambda: {"wolf", "goat", "cabbage"})    right_bank: set[str] = Field(default_factory=set)    boat_location: str = "left"    move_description: str = "初始状态。"    defis_valid(self) -> bool:        """检查当前状态是否有效。"""        # 如果船在右岸，检查左岸是否有无效组合        ifself.boat_location == "right":            if"wolf"inself.left_bank and"goat"inself.left_bank: returnFalse            if"goat"inself.left_bank and"cabbage"inself.left_bank: returnFalse        # 如果船在左岸，检查右岸是否有无效组合        ifself.boat_location == "left":            if"wolf"inself.right_bank and"goat"inself.right_bank: returnFalse            if"goat"inself.right_bank and"cabbage"inself.right_bank: returnFalse        returnTrue        defis_goal(self) -> bool:        """检查我们是否获胜。"""        returnself.right_bank == {"wolf", "goat", "cabbage"}        # 使状态可哈希，以便我们检测循环    def__hash__(self):        returnhash((frozenset(self.left_bank), frozenset(self.right_bank), self.boat_location))

现在是ToT智能体的核心。我们图的状态将持有我们思想树中所有的活动路径。expand_paths 节点将生成新的分支，而 prune_paths 节点将通过移除任何进入死胡同或循环的路径来修剪树。

class ToTState(TypedDict):    problem_description: str    active_paths: List[List[PuzzleState]] # 这是我们的"树"    solution: Optional[List[PuzzleState]]defexpand_paths(state: ToTState) -> Dict[str, Any]:    """'想法生成器'。用所有有效的下一步扩展每个活动路径。"""    console.print("--- 正在扩展路径 ---")    new_paths = []    for path in state['active_paths']:        last_state = path[-1]        # 从当前状态获取所有有效的下一步状态        possible_next_states = get_possible_moves(last_state) # 假设 get_possible_moves 已定义        for next_state in possible_next_states:            new_paths.append(path + [next_state])    console.print(f"[cyan]已扩展到 {len(new_paths)} 条潜在路径。[/cyan]")    return {"active_paths": new_paths}defprune_paths(state: ToTState) -> Dict[str, Any]:    """'状态评估器'。修剪无效或包含循环的路径。"""    console.print("--- 正在修剪路径 ---")    pruned_paths = []    for path in state['active_paths']:        # 检查循环：如果最后一个状态在路径中之前出现过        if path[-1] in path[:-1]:            continue# 发现循环，修剪此路径        pruned_paths.append(path)    console.print(f"[green]已修剪至 {len(pruned_paths)} 条有效、无循环的路径。[/green]")    return {"active_paths": pruned_paths}# ... (带有检查解决方案的条件边的图连接) ...workflow = StateGraph(ToTState)workflow.add_node("expand", expand_paths)workflow.add_node("prune", prune_paths)

ToT架构 (由Fareed Khan创建)

让我们在谜题上运行我们的ToT智能体。一个简单的思维链（Chain-of-Thought）智能体如果以前见过这个问题，也许能解决，但那只是回忆一个解决方案。我们的ToT智能体将通过系统搜索发现解决方案。

problem = "一个农民想带着狼、羊和白菜过河..."console.print("--- 🌳 正在运行思维树智能体 ---")final_state = tot_agent.invoke({"problem_description": problem}, {"recursion_limit": 15})console.print("\n--- ✅ ToT智能体解决方案 ---")

输出追踪显示了智能体在工作中，有条不紊地探索谜题。

--- 正在扩展路径 ---[cyan]已扩展到 1 条潜在路径。[/cyan]--- 正在修剪路径 ---[green]已修剪至 1 条有效、无循环的路径。[/green]--- 正在扩展路径 ---[cyan]已扩展到 2 条潜在路径。[/cyan]--- 正在修剪路径 ---[green]已修剪至 2 条有效、无循环的路径。[/green]...[bold green]找到解决方案！[/bold green]--- ✅ ToT智能体解决方案 ---1. 初始状态。2. 将羊移动到右岸。3. 将空船移动到左岸。4. 将狼移动到右岸。5. 将羊移动到左岸。6. 将白菜移动到右岸。7. 将空船移动到左岸。8. 将羊移动到右岸。

智能体找到了正确的8步解决方案！它不仅仅是猜测；它系统地探索了可能性，排除了坏的选项，并找到了一条保证有效的路径。这就是ToT的力量。

为了使其规范化，我们将使用一个专注于推理过程质量的“LLM即裁判”。

class ReasoningEvaluation(BaseModel):    """用于评估智能体推理过程的模式。"""    solution_correctness_score: int = Field(description="最终解决方案是否正确和有效的评分（1-10分）。")    reasoning_robustness_score: int = Field(description="智能体过程的健壮性评分。高分意味着它系统地探索了问题，低分意味着它只是猜测。（1-10分）")    justification: str = Field(description="对评分的简要说明。")

一个简单的思维链智能体如果幸运的话可能会得到很高的正确性分数，但其健壮性分数会很低。然而，我们的ToT智能体得到了高分。

--- 正在评估ToT智能体的过程 ---{    'solution_correctness_score': 8,    'reasoning_robustness_score': 9,    'justification': "该智能体的过程非常健壮。它不仅仅是提供了一个答案；它系统地探索了一个可能性的树，修剪了无效路径，并保证了一个正确的解决方案。这是一种比单次猜测可靠得多的方法。"}

我们可以看到，ToT智能体之所以能成功，并非偶然，而是因为它的搜索是可靠的。

这使其成为需要高可靠性任务的更好选择。

多智能体系统 (Multi-Agent Systems)

到目前为止，我们实现的所有方法都是独立工作的。

当一个问题对于单个智能体来说太大或太复杂，无法有效处理时，会发生什么？

多智能体系统不是构建一个无所不能的超级智能体，而是使用一个专家团队。每个智能体专注于自己的领域，就像人类专家一样——你不会让数据科学家去写营销文案。

对于复杂的任务，比如生成市场分析，你可以让一个新闻专家、一个金融专家和一个股票专家协同工作，以获得更好的结果。

多智能体系统的实现方式可能非常复杂，但一个简化版本的工作方式如下……

多智能体 (由Fareed Khan创建)

1. 1. 分解 (Decomposition)： 一个复杂的任务被分解为子任务。
2. 2. 角色定义 (Role Definition)： 每个子任务根据其定义的角色（例如，“研究员”、“程序员”、“作者”）被分配给一个专家智能体。
3. 3. 协作 (Collaboration)： 智能体执行他们的任务，将他们的发现传递给彼此或一个中央管理者。
4. 4. 综合 (Synthesis)： 一个最终的“管理者”智能体收集来自专家的输出，并组装成最终的、整合的响应。

为了真正看出团队的优势，我们首先需要一个基准。我们将构建一个单一的、庞大的“通才”智能体，并给它一个复杂、多方面的任务。

然后，我们将构建我们的专家团队：一个新闻分析师、一个技术分析师和一个财务分析师。每个都将是其自己的智能体节点，具有非常特定的角色。一个最终的报告撰写者将作为管理者，编译他们的工作。

class AgentState(TypedDict):    user_request: str    news_report: Optional[str]    technical_report: Optional[str]    financial_report: Optional[str]    final_report: Optional[str]# 一个帮助我们干净地创建专家节点的工厂函数defcreate_specialist_node(persona: str, output_key: str):    """创建专家智能体节点的工厂函数。"""    system_prompt = persona + "\n\n你有一个网络搜索工具的访问权限。你的输出必须是一个简洁的报告部分，只专注于你的专业领域。"        prompt = ChatPromptTemplate.from_messages([("system", system_prompt), ("human", "{user_request}")])    agent = prompt | llm.bind_tools([search_tool])    defspecialist_node(state: AgentState):        console.print(f"--- 正在调用 {output_key.replace('_report','').upper()} 分析师 ---")        result = agent.invoke({"user_request": state["user_request"]})        return {output_key: result.content}    return specialist_node# 创建专家节点news_analyst_node = create_specialist_node("你是一位专业的新闻分析师...", "news_report")technical_analyst_node = create_specialist_node("你是一位专业的技术分析师...", "technical_report")financial_analyst_node = create_specialist_node("你是一位专业的财务分析师...", "financial_report")defreport_writer_node(state: AgentState):    """综合专家报告的管理者智能体。"""    console.print("--- 正在调用报告撰写者 ---")    prompt = f"""你是一位专业的金融编辑。你的任务是将以下专家报告合并成一份单一、专业、连贯的市场分析报告。        新闻报告：{state['news_report']}    技术报告：{state['technical_report']}    财务报告：{state['financial_report']}    """    final_report = llm.invoke(prompt).content    return {"final_report": final_report}

现在让我们在LangGraph中将它们连接起来。在这个例子中，我们将使用一个简单的顺序工作流：新闻分析师先执行，然后是技术分析师，依此类推。

multi_agent_graph_builder = StateGraph(AgentState)# 添加所有节点multi_agent_graph_builder.add_node("news_analyst", news_analyst_node)multi_agent_graph_builder.add_node("technical_analyst", technical_analyst_node)multi_agent_graph_builder.add_node("financial_analyst", financial_analyst_node)multi_agent_graph_builder.add_node("report_writer", report_writer_node)# 定义工作流序列multi_agent_graph_builder.set_entry_point("news_analyst")multi_agent_graph_builder.add_edge("news_analyst", "technical_analyst")multi_agent_graph_builder.add_edge("technical_analyst", "financial_analyst")multi_agent_graph_builder.add_edge("financial_analyst", "report_writer")multi_agent_graph_builder.add_edge("report_writer", END)multi_agent_app = multi_agent_graph_builder.compile()

多智能体 (由Fareed Khan创建)

让我们给我们的专家团队一个复杂的任务：为NVIDIA创建一份完整的市场分析报告。一个单一的、通才的智能体可能会生成一大段浅显、无结构的文本。让我们看看我们的团队表现如何。

multi_agent_query = "为NVIDIA (NVDA) 创建一份简短但全面的市场分析报告。"console.print(f"[bold green]正在对多智能体团队进行相同任务的测试：[/bold green]\n'{multi_agent_query}'\n")final_multi_agent_output = multi_agent_app.invoke({"user_request": multi_agent_query})console.print("\n--- [bold green]多智能体团队的最终报告[/bold green] ---")console.print(Markdown(final_multi_agent_output['final_report']))

最终报告的差异是显著的。来自多智能体团队的输出结构高度化，每个分析领域都有清晰、独立的部分。每个部分都包含更详细、领域特定的语言和见解。

### 市场分析报告：NVIDIA**引言**NVIDIA一直是投资者关注的焦点... 本报告综合了三位专家的发现...**新闻与情绪报告**近期围绕NVIDIA的新闻喜忧参半... 公司面临着日益激烈的竞争... 但在AI领域也取得了显著进展...**技术分析报告**NVIDIA的股价在过去一年中一直处于牛市趋势... 报告还强调了公司强劲的盈利增长...**财务表现报告**报告强调了公司强劲的收入增长... 毛利率一直在增加...**结论**总之，NVIDIA的市场分析表明，该公司面临着日益激烈的竞争... 但仍然是一个强大的参与者...

这个输出远优于单个智能体所能产生的。通过分工合作，我们得到了一个结构化、深入且专业的结果。

为了使其规范化，我们将使用一个专注于最终报告质量的“LLM即裁判”。

class ReportEvaluation(BaseModel):    """用于评估财务报告的模式。"""    clarity_and_structure_score: int = Field(description="报告的组织、结构和清晰度的评分（1-10分）。")    analytical_depth_score: int = Field(description="每个部分分析的深度和质量的评分（1-10分）。")    completeness_score: int = Field(description="报告对用户请求所有部分的满足程度的评分（1-10分）。")    justification: str = Field(description="对评分的简要说明。")

在评判时，差异是显而易见的。一个单一智能体可能得分在6或7分左右。然而，我们的多智能体团队得到了更高的分数。

--- 正在评估多智能体团队的报告 ---{    'clarity_and_structure_score': 9,    'analytical_depth_score': 8,    'completeness_score': 9,    'justification': "该报告结构异常清晰，每个分析类型都有明确、独立的部分。每个部分都提供了很好的专家级细节。结论中的综合是合乎逻辑的，并得到了前面专家报告的充分支持。"}

从分数中我们可以看出，对于可以分解的复杂任务……

一个专家团队几乎总是会胜过一个单一的通才。

元控制器 (Meta-Controller)

在多智能体团队中，你可能已经注意到它有点僵化。我们硬编码了序列：新闻 -> 技术 -> 财务 -> 撰写者。

如果用户只想要一份技术分析怎么办？我们的系统仍然会浪费时间和金钱去运行其他分析师。

元控制器 (Meta-Controller) 架构为此引入了一个智能调度器。这个控制器智能体唯一的工作就是查看用户的请求，并决定哪个专家最适合这项工作。

在 RAG 或智能体系统中，元控制器是中枢神经系统。它是将传入请求路由到正确部门的前门。

一个简单版本的元控制器工作方式如下：

元控制器 (由Fareed Khan创建)

1. 1. 接收输入 (Receive Input)： 系统接收一个用户请求。
2. 2. 元控制器分析 (Meta-Controller Analysis)： 元控制器智能体检查请求以理解其意图。
3. 3. 分派给专家 (Dispatch to Specialist)： 根据其分析，它从其专家池中选择最佳的专家智能体（例如，“研究员”、“程序员”、“通才”）。
4. 4. 执行任务 (Execute Task)： 选定的专家智能体运行并生成一个结果。
5. 5. 返回结果 (Return Result)： 专家的结果直接返回给用户。

首先，我们需要编写操作的大脑：meta_controller_node。它的工作是查看用户的请求，并使用可用专家的列表来选择正确的那个。这里的提示至关重要，因为它需要向控制器清楚地解释每个专家的角色。

class ControllerDecision(BaseModel):    """元控制器做出的路由决策。"""    next_agent: str = Field(description="接下来要调用的专家智能体的名称。必须是['Generalist', 'Researcher', 'Coder']中的一个。")    reasoning: str = Field(description="选择下一个智能体的简要理由。")defmeta_controller_node(state: MetaAgentState):    """决定调用哪个专家的中央控制器。"""    console.print("--- 🧠 元控制器正在分析请求 ---")        specialists = {        "Generalist": "处理日常对话、问候和简单问题。",        "Researcher": "回答需要从网络获取最新信息的问题。",        "Coder": "根据用户的规范编写Python代码。"    }    specialist_descriptions = "\n".join([f"- {name}: {desc}"for name, desc in specialists.items()])        prompt = ChatPromptTemplate.from_template(        f"""你是一个多智能体AI系统的元控制器。你的工作是将用户的请求路由到最合适的专家智能体。以下是可用的专家：{specialist_descriptions}分析以下用户请求并选择最佳专家。用户请求：“{{user_request}}”"""    )        controller_llm = llm.with_structured_output(ControllerDecision)    chain = prompt | controller_llm        decision = chain.invoke({"user_request": state['user_request']})    console.print(f"[yellow]路由决策：[/yellow] 发送给 [bold]{decision.next_agent}[/bold]。[italic]理由：{decision.reasoning}[/italic]")        return {"next_agent_to_call": decision.next_agent}

定义了我们的控制器后，我们只需要在LangGraph中将它连接起来。图将从 meta_controller 开始，然后一个条件边将根据其决策将请求路由到正确的专家节点。

class MetaAgentState(TypedDict):    user_request: str    next_agent_to_call: Optional[str]    generation: str# 我们可以重用专家节点（generalist_node, research_agent_node, 等）workflow = StateGraph(MetaAgentState)workflow.add_node("meta_controller", meta_controller_node)workflow.add_node("Generalist", generalist_node)workflow.add_node("Researcher", research_agent_node)workflow.add_node("Coder", coder_node)workflow.set_entry_point("meta_controller")defroute_to_specialist(state: MetaAgentState) -> str:    """读取控制器的决策并返回要路由到的节点名称。"""    return state["next_agent_to_call"]workflow.add_conditional_edges("meta_controller", route_to_specialist)# 任何专家运行后，过程结束workflow.add_edge("Generalist", END)workflow.add_edge("Researcher", END)workflow.add_edge("Coder", END)meta_agent = workflow.compile()

元控制器 (由Fareed Khan创建)

现在让我们用各种提示来测试我们的调度器。每一个都旨在看控制器是否将其发送给正确的专家。

# 测试 1：应路由到通才run_agent("你好，今天过得怎么样？")# 测试 2：应路由到研究员run_agent("NVIDIA最新的财务业绩如何？")# 测试 3：应路由到程序员run_agent("你能为我写一个计算第n个斐波那契数的python函数吗？")

输出追踪显示控制器每次都做出了明智的决定。

--- 🧠 元控制器正在分析请求 ---[yellow]路由决策：[/yellow] 发送给 [bold]Generalist[/bold]。[italic]理由：用户的请求是一个简单的问候...[/italic]最终响应：你好！今天我能为你做些什么？--- 🧠 元控制器正在分析请求 ---[yellow]路由决策：[/yellow] 发送给 [bold]Researcher[/bold]。[italic]理由：用户正在询问一个最近的事件...[/italic]最终响应：NVIDIA最新的财务业绩...异常强劲。他们报告收入为260.4亿美元...--- 🧠 元控制器正在分析请求 ---[yellow]路由决策：[/yellow] 发送给 [bold]Coder[/bold]。[italic]理由：用户明确要求一个Python函数...[/italic]最终响应：```pythondef fibonacci(n):    # ... (代码) ...
``````plaintext
系统工作得非常完美。问候语交给了通才，新闻查询交给了研究员，代码请求交给了程序员。控制器根据任务内容正确地分派了任务。为了使其规范化，我们将使用一个只评分一件事的“LLM即裁判”：路由的正确性。```pythonclass RoutingEvaluation(BaseModel):    """用于评估元控制器路由决策的模式。"""    routing_correctness_score: int = Field(description="控制器是否为请求选择了最合适的专家的评分（1-10分）。")    justification: str = Field(description="对评分的简要说明。")

对于像“法国的首都是哪里？”这样的查询，裁判的评估将是：

--- 正在评估元控制器的路由 ---{    'routing_correctness_score': 8.5,    'justification': "控制器正确地将用户的请求识别为需要最新信息的事实查询，并将其路由到'研究员'智能体。这是最佳选择，因为通才可能拥有过时的知识。"}

这个分数表明我们的控制器不仅仅是在路由——它是在智能地分派。

这使得AI系统具有可扩展性和易于维护性，因为可以通过插入专家并更新控制器来添加新技能。

黑板 (Blackboard)

如果问题总是一样的，前面的架构可能已经奏效了……

但如果最佳的下一步取决于上一步的结果呢？一个僵化的序列可能会非常低效，迫使系统运行不必要的步骤。

这就是黑板 (Blackboard) 架构发挥作用的地方。这是一种更先进、更灵活的协调专家团队的方式。这个想法来自于人类专家解决问题的方式：

1. 1. 他们围在一块黑板旁，这是一个共享的工作空间，任何人都可以写下发现。
2. 2. 然后，一个领导者查看黑板并决定下一步谁应该贡献。

在编写你的架构时，黑板是你处理复杂、非结构化问题的模式，其中解决方案路径是未知的。它允许一种涌现的、机会主义的策略，使其非常适合动态的意义建构或复杂的诊断，其中下一步总是对最新发现的反应。

让我们来理解它的流程……

黑板记忆 (由Fareed Khan创建)

1. 1. 共享内存（黑板）(Shared Memory (The Blackboard))： 一个中央数据存储保存问题的当前状态和迄今为止的所有发现。
2. 2. 专家智能体 (Specialist Agents)： 一个由独立智能体组成的池，每个都有特定的技能，监控黑板。
3. 3. 控制器 (Controller)： 一个中央“控制器”智能体也监控黑板。它的工作是分析当前状态并决定哪个专家最适合进行下一步。
4. 4. 机会性激活 (Opportunistic Activation)： 控制器激活选定的智能体。该智能体从黑板上读取信息，完成其工作，并将其发现写回。
5. 5. 迭代 (Iteration)： 这个过程重复进行，控制器动态地选择下一个智能体，直到它决定问题已解决。

让我们开始构建它。

这个系统最重要的部分是智能的控制器。与我们之前的元控制器只做一次性分派不同，这个控制器在一个循环中运行。在每个专家智能体运行后，控制器重新评估黑板并决定下一步做什么。

class BlackboardState(TypedDict):    user_request: str    blackboard: List[str] # 共享工作区    available_agents: List[str]    next_agent: Optional[str] # 控制器的决定classControllerDecision(BaseModel):    next_agent: str = Field(description="接下来要调用的智能体的名称。必须是['News Analyst', 'Technical Analyst', 'Financial Analyst', 'Report Writer']或'FINISH'中的一个。")    reasoning: str = Field(description="选择下一个智能体的简要理由。")defcontroller_node(state: BlackboardState):    """分析黑板并决定下一步的智能控制器。"""    console.print("--- 控制器：正在分析黑板... ---")    controller_llm = llm.with_structured_output(ControllerDecision)    blackboard_content = "\n\n".join(state['blackboard'])        prompt = f"""你是一个多智能体系统的中央控制器。你的工作是分析共享的黑板和原始用户请求，以决定接下来应该运行哪个专家智能体。**原始用户请求：**{state['user_request']}**当前黑板内容：**---{blackboard_content if blackboard_content else "黑板当前为空。"}---**可用的专家智能体：**{', '.join(state['available_agents'])}**你的任务：**1. 仔细阅读用户请求和当前黑板内容。2. 确定为了更接近完整答案，*下一个逻辑步骤*是什么。3. 选择执行该步骤的唯一最佳智能体。4. 如果请求已完全解决，选择'FINISH'。以所需格式提供你的决定。"""    decision = controller_llm.invoke(prompt)    console.print(f"--- 控制器：决定调用'{decision.next_agent}'。理由：{decision.reasoning} ---")    return {"next_agent": decision.next_agent}

现在我们只需要在LangGraph中将它连接起来。关键是中央循环：任何专家智能体在运行后，都将控制权返回给控制器以进行下一个决策。

# ... (专家节点的定义与多智能体系统类似) ...bb_graph_builder = StateGraph(BlackboardState)bb_graph_builder.add_node("Controller", controller_node)bb_graph_builder.add_node("News Analyst", news_analyst_bb)# ... 添加其他专家节点 ...bb_graph_builder.set_entry_point("Controller")defroute_to_agent(state: BlackboardState):    return state["next_agent"]bb_graph_builder.add_conditional_edges("Controller", route_to_agent, {    "News Analyst": "News Analyst",    # ... 其他路由 ...    "FINISH": END})# 任何专家运行后，控制权总是返回给控制器bb_graph_builder.add_edge("News Analyst", "Controller")# ... 其他返回控制器的边 ...blackboard_app = bb_graph_builder.compile()

黑板架构 (由Fareed Khan创建)

为了看看这比一个僵化的序列好多少，让我们给它一个带有条件逻辑的任务，顺序智能体将会失败。

dynamic_query = "查找关于Nvidia的最新重大新闻。根据该新闻的情绪，进行技术分析（如果新闻是中性或积极的）或财务分析（如果新闻是负面的）。"initial_bb_input = {"user_request": dynamic_query, "blackboard": [], "available_agents": ["News Analyst", "Technical Analyst", "Financial Analyst", "Report Writer"]}final_bb_output = blackboard_app.invoke(initial_bb_input, {"recursion_limit": 10})console.print("\n--- [bold green]黑板系统的最终报告[/bold green] ---")console.print(Markdown(final_bb_output['blackboard'][-1]))
``````plaintext
--- 控制器：正在分析黑板... ---  决定：调用'新闻分析师'...  --- 黑板状态 ---  智能体'新闻分析师'正在工作...  --- 控制器：正在分析黑板... ---  决定：调用'技术分析师'（新闻积极）...  --- 黑板状态 ---  报告1：Nvidia新闻积极，新AI芯片“Rubin”，市场情绪看涨...  --- (黑板) 智能体'技术分析师'正在工作... ---  --- 控制器：正在分析黑板... ---  决定：调用'报告撰写者'（分析完成，综合报告）......

执行追踪显示了一个远比之前智能的过程。顺序智能体会运行技术和财务分析师，浪费资源。我们的黑板系统更聪明：

1. 1. 控制器启动： 它看到一个空的黑板并调用 新闻分析师。
2. 2. 新闻分析师运行： 它找到关于Nvidia的积极新闻并将其发布到黑板上。
3. 3. 控制器重新评估： 它阅读积极的新闻并正确地决定下一步是调用 技术分析师，完全跳过了财务分析师。
4. 4. 专家运行： 技术分析师完成其工作并发布其报告。
5. 5. 控制器完成： 它看到所有必要的分析都已完成，并调用 报告撰写者 来综合最终答案，然后结束。

这种动态、机会主义的工作流正是黑板系统的定义。为了使其规范化，我们的LLM即裁判评估每个贡献，对逻辑一致性和效率进行评分，确保涌现的解决方案既合理又可操作。

class ProcessLogicEvaluation(BaseModel):    instruction_following_score: int = Field(description="智能体遵循条件指令的程度评分（1-10分）。")    process_efficiency_score: int = Field(description="智能体是否避免了不必要工作的评分（1-10分）。")    justification: str = Field(description="对评分的简要说明。")

一个顺序智能体在这里会得到很差的分数。然而，我们的黑板系统通过了测试。

--- 正在评估黑板系统的过程 ---{    'instruction_following_score': 7,    'process_efficiency_score': 8,    'justification': "该智能体完美地遵循了用户的条件指令。在新闻分析师报告了积极情绪后，系统正确地选择了运行技术分析师，并完全跳过了财务分析师。这展示了无懈可击的指令遵循和最佳的过程效率。"}

对于那些前进道路取决于中间结果的复杂问题，……

黑板架构的灵活性可能优于多智能体系统。

集成决策 (Ensemble Decision-Making)

到目前为止，我们所有的智能体，甚至是团队，都有一个共同点：它们产生单一的推理线。

但LLMs是非确定性的，运行同一个提示两次，你可能会得到略有不同的答案。

这在需要可靠、全面答案的高风险情况下可能是一个问题。

集成决策 (Ensemble Decision-Making) 架构直接解决了这个问题。它基于“群体智慧”原则。

我们不是依赖一个智能体，而是并行运行多个独立的智能体，通常具有不同的“个性”，然后使用一个最终的聚合智能体将它们的输出合成为一个单一、更健壮的结论。

在大型AI系统中，这是任何关键任务决策支持任务的首选模式。想象一个AI投资委员会或一个医疗诊断系统。从不同的AI角色那里获得“第二意见”（或第三、第四意见）大大降低了单个智能体的偏见或幻觉导致糟糕结果的风险。

让我们来理解一下这个流程。

集成 (由Fareed Khan创建)

1. 1. 分发（并行探索）(Fan-Out (Parallel Exploration))： 用户的查询同时发送给多个专家智能体。这些智能体被赋予不同的角色以鼓励多样化思维。
2. 2. 独立处理 (Independent Processing)： 每个智能体独立地处理问题，生成其自己的完整分析。
3. 3. 汇集（聚合）(Fan-In (Aggregation))： 收集所有智能体的输出。
4. 4. 综合（集成决策）(Synthesize (Ensemble Decision))： 一个最终的“聚合器”智能体接收所有单个报告，权衡不同的观点，并综合成一个全面的最终答案。

一个好的集成的关键是认知多样性。我们将创建三个分析师智能体，每个都有非常不同的个性：一个看涨的成长型分析师（乐观主义者），一个谨慎的价值型分析师（怀疑论者），以及一个量化分析师（数据纯粹主义者）。

class EnsembleState(TypedDict):    query: str    analyses: Dict[str, str] # 存储每个并行智能体的输出    final_recommendation: Optional[Any]# 我们将再次使用我们的专家工厂，但具有非常不同的角色bullish_persona = "看涨的成长型分析师：你对技术和创新极其乐观。专注于未来的增长潜力，淡化短期风险。"bullish_analyst_node = create_specialist_node(bullish_persona, "BullishAnalyst")value_persona = "谨慎的价值型分析师：你是一个持怀疑态度的投资者，专注于基本面和风险。仔细审查财务、竞争和潜在的下行情景。"value_analyst_node = create_specialist_node(value_persona, "ValueAnalyst")quant_persona = "量化分析师（Quant）：你纯粹是数据驱动的。忽略叙事，只关注硬性数字，如财务指标和技术指标。"quant_analyst_node = create_specialist_node(quant_persona, "QuantAnalyst")

最后也是最重要的部分是我们的聚合器，即cio_synthesizer_node。它的工作是接收这些相互冲突的报告，并产生一个单一、平衡的投资论点。

class FinalRecommendation(BaseModel):    """来自CIO的最终、综合的投资论点。"""    final_recommendation: str    confidence_score: float    synthesis_summary: str    identified_opportunities: List[str]    identified_risks: List[str]defcio_synthesizer_node(state: EnsembleState) -> Dict[str, Any]:    """综合所有分析成一个单一推荐的最终节点。"""    console.print("--- 🏛️ 正在调用首席投资官进行最终决策 ---")    all_analyses = "\n\n---\n\n".join([f"**来自{name}的分析：**\n{analysis}"for name, analysis in state['analyses'].items()])        cio_prompt = ChatPromptTemplate.from_messages([        ("system", "你是首席投资官。你的任务是将这些多样化且常常相互冲突的观点综合成一个单一、最终、可操作的投资论点。权衡增长潜力与风险，以得出平衡的结论。"),        ("human", "这是你团队关于查询的报告：'{query}'\n\n{analyses}\n\n请提供你最终的、综合的投资论点。")    ])        cio_llm = llm.with_structured_output(FinalRecommendation)    chain = cio_prompt | cio_llm    final_decision = chain.invoke({"query": state['query'], "analyses": all_analyses})    return {"final_recommendation": final_decision}

现在让我们把它连接起来。这个的LangGraph是独特的，它有一个**“分发”**，一个节点分支到三个并行节点，然后是一个“汇集”，这三个节点汇聚到最终的综合器上。

workflow = StateGraph(EnsembleState)# 入口节点只是准备状态workflow.add_node("start_analysis", lambda state: {"analyses": {}})# 添加并行的分析师节点和最终的综合器workflow.add_node("bullish_analyst", bullish_analyst_node)workflow.add_node("value_analyst", value_analyst_node)workflow.add_node("quant_analyst", quant_analyst_node)workflow.add_node("cio_synthesizer", cio_synthesizer_node)workflow.set_entry_point("start_analysis")# 分发：并行运行所有三个分析师workflow.add_edge("start_analysis", ["bullish_analyst", "value_analyst", "quant_analyst"])# 汇集：所有分析师完成后，调用综合器workflow.add_edge(["bullish_analyst", "value_analyst", "quant_analyst"], "cio_synthesizer")workflow.add_edge("cio_synthesizer", END)ensemble_agent = workflow.compile()

集成 (由Fareed Khan创建)

让我们给我们的投资委员会一个棘手的、模棱两可的问题，其中不同的视角是有价值的。

query = "根据最近的新闻、财务表现和未来前景，NVIDIA (NVDA) 在2024年中期是否是一项好的长期投资？"console.print(f"--- 📈 正在运行投资委员会，针对：{query} ---")result = ensemble_agent.invoke({"query": query})# ... (打印单个报告和最终CIO推荐的代码) ...

当你看到结果时，这种架构的力量立刻显现出来。三个分析师产生了截然不同的报告：看涨的给出了一个光明的“买入”，价值分析师给出了一个谨慎的“持有”，而量化分析师提供了中立的数据。

首席投资官的最终报告不仅仅是把它们平均一下。它进行了一次真正的综合，承认了看涨的理由，但用价值方面的担忧来缓和它。

**最终推荐：** 买入**信心分数：** 7.5/10**综合摘要：**委员会为NVIDIA提供了一个引人注目但有争议的案例。在公司当前的技术主导地位上有一致的共识... 然而，价值和量化分析师提出了关于股票极高估值的关键、一致的观点... 最终的推荐是‘买入’，但强烈强调这是一个长期头寸，并建议谨慎入场...**已识别的机会：*** AI加速器市场无可争议的领导地位。* ...**已识别的风险：*** 极高的估值（市盈率和市销率）。* ...

这是一个比任何单个智能体所能提供的都更健壮和可信的答案。为了使其规范化，我们的“LLM即裁判”需要对分析深度和平衡性进行评分。

class EnsembleEvaluation(BaseModel):    analytical_depth_score: int = Field(description="分析深度的评分（1-10分）。")    nuance_and_balance_score: int = Field(description="最终答案在平衡冲突观点和提供细致结论方面的表现评分（1-10分）。")    justification: str = Field(description="对评分的简要说明。")

在评判时，集成的输出得到了高分。

--- 正在评估集成智能体的输出 ---{    'analytical_depth_score': 9,    'nuance_and_balance_score': 8,    'justification': "最终答案异常平衡。它不只是选择一方，而是巧妙地将乐观的增长案例与怀疑的估值担忧综合起来，提供了一个反映投资决策现实世界复杂性的细致推荐。这是一个高质量、可靠的分析。"}

你可以看到，当我们使用多样化视角的集成时，它增加了我们智能体推理的可靠性和深度，类似于我们使用深度思考模型时看到的情况。

一直在更新，更多的大模型学习和面试资料已经上传带到CSDN的官方了，有需要的朋友可以扫描下方二维码免费领取【保证100%免费】👇👇

情景记忆 + 语义记忆 (Episodic + Semantic Memory)

我们所有的智能体都像金鱼一样，记忆短暂，一旦对话结束，一切都被遗忘。

要构建一个真正能与用户一同学习和成长的个人助手，我们需要给它一个长期记忆的组件。

情景记忆 + 语义记忆栈 (Episodic + Semantic Memory Stack) 架构模仿人类认知，为智能体提供两种类型的记忆：

情景记忆 (由Fareed Khan创建)

• • 情景记忆 (Episodic Memory)： 这是对特定事件的记忆，比如过去的对话。它回答“发生了什么？”我们将为此使用向量数据库 (vector database)。
• • 语义记忆 (Semantic Memory)： 这是从那些事件中提取的结构化事实和关系的记忆。它回答“我知道什么？”我们将为此使用图数据库 (graph database) (Neo4j)。

你可能知道，这是任何AI系统个性化的核心。电子商务机器人就是这样记住你的风格，家教记住你的弱点，个人助理记住你数周乃至数月的项目和偏好。

它的工作原理是这样的……

1. 1. 交互 (Interaction)： 智能体与用户进行对话。
2. 2. 记忆检索 (Memory Retrieval)： 对于一个新的查询，智能体在其情景（向量）和语义（图）记忆中搜索相关上下文。
3. 3. 增强生成 (Augmented Generation)： 检索到的记忆被用来生成一个个性化的、具有上下文意识的响应。
4. 4. 记忆创建 (Memory Creation)： 交互之后，一个“记忆制造者”智能体分析对话，创建一个摘要（情景记忆），并提取事实（语义记忆）。
5. 5. 记忆存储 (Memory Storage)： 新的记忆被保存到它们各自的数据库中。

这个系统的核心是**“记忆制造者”**，一个负责处理对话和创建新记忆的智能体。它有两个工作：为向量存储创建一个简洁的摘要，并为图提取结构化的事实。

# 用于知识提取的Pydantic模型classNode(BaseModel):    id: str; type: strclassRelationship(BaseModel):    source: Node; target: Node; type: strclassKnowledgeGraph(BaseModel):    relationships: List[Relationship]defcreate_memories(user_input: str, assistant_output: str):    conversation = f"用户: {user_input}\n助手: {assistant_output}"        # 创建情景记忆（摘要）    console.print("--- 正在创建情景记忆（摘要）---")    summary_prompt = ChatPromptTemplate.from_messages([        ("system", "你是一位摘要专家。请为以下用户-助手交互创建一个简洁的一句话摘要。这个摘要将作为未来回忆的记忆。"),        ("human", "交互:\n{interaction}")    ])    summarizer = summary_prompt | llm    episodic_summary = summarizer.invoke({"interaction": conversation}).content        new_doc = Document(page_content=episodic_summary, metadata={"created_at": uuid.uuid4().hex})    episodic_vector_store.add_documents([new_doc])    console.print(f"[green]情景记忆已创建：[/green] '{episodic_summary}'")        # 创建语义记忆（事实提取）    console.print("--- 正在创建语义记忆（图）---")    extraction_llm = llm.with_structured_output(KnowledgeGraph)    extraction_prompt = ChatPromptTemplate.from_messages([        ("system", "你是一位知识提取专家。你的任务是从对话中识别关键实体及其关系，并将它们建模为一个图。专注于用户偏好、目标和陈述的事实。"),        ("human", "从这次交互中提取所有关系:\n{interaction}")    ])    extractor = extraction_prompt | extraction_llm    try:        kg_data = extractor.invoke({"interaction": conversation})        if kg_data.relationships:            for rel in kg_data.relationships:                graph.add_graph_documents([rel], include_source=True)            console.print(f"[green]语义记忆已创建：[/green] 已向图中添加 {len(kg_data.relationships)} 个关系。")        else:            console.print("[yellow]本次交互中未识别到新的语义记忆。[/yellow]")    except Exception as e:        console.print(f"[red]无法提取或保存语义记忆: {e}[/red]")

有了我们的记忆制造者，我们就可以构建完整的智能体了。图是一个简单的序列：检索记忆，使用它们生成一个响应，然后用新的对话更新记忆。

class AgentState(TypedDict):    user_input: str    retrieved_memories: Optional[str]    generation: strdefretrieve_memory(state: AgentState) -> Dict[str, Any]:    """从情景和语义存储中检索记忆的节点。"""    console.print("--- 正在检索记忆 ---")    user_input = state['user_input']        # 从情景记忆中检索    retrieved_docs = episodic_vector_store.similarity_search(user_input, k=2)    episodic_memories = "\n".join([doc.page_content for doc in retrieved_docs])        # 从语义记忆中检索    # 这是一个简单的检索；更高级的方法会涉及从查询中提取实体    try:        graph_schema = graph.get_schema        # 使用全文索引以获得更好的检索效果。Neo4j会自动在节点属性上创建索引。        # 一个更健壮的解决方案可能需要先从user_input中提取实体。        semantic_memories = str(graph.query("""            UNWIND $keywords AS keyword            CALL db.index.fulltext.queryNodes("entity", keyword) YIELD node, score            MATCH (node)-[r]-(related_node)            RETURN node, r, related_node LIMIT 5            """, {'keywords': user_input.split()}))    except Exception as e:        semantic_memories = f"无法查询图: {e}"            retrieved_content = f"相关的过去对话（情景记忆）:\n{episodic_memories}\n\n相关事实（语义记忆）:\n{semantic_memories}"    console.print(f"[cyan]检索到的上下文:\n{retrieved_content}[/cyan]")        return {"retrieved_memories": retrieved_content}defgenerate_response(state: AgentState) -> Dict[str, Any]:    """使用检索到的记忆生成响应的节点。"""    console.print("--- 正在生成响应 ---")    prompt = ChatPromptTemplate.from_messages([        ("system", "你是一个有帮助且个性化的金融助手。使用检索到的记忆来告知你的响应并为用户量身定制。如果记忆表明了用户的偏好（例如，他们是保守型投资者），你必须尊重它。"),        ("human", "我的问题是: {user_input}\n\n这里有一些可能相关的记忆:\n{retrieved_memories}")    ])    generator = prompt | llm    generation = generator.invoke(state).content    console.print(f"[green]生成的响应:\n{generation}[/green]")    return {"generation": generation}defupdate_memory(state: AgentState) -> Dict[str, Any]:    """用最新的交互更新记忆的节点。"""    console.print("--- 正在更新记忆 ---")    create_memories(state['user_input'], state['generation'])    return {}workflow = StateGraph(AgentState)workflow.add_node("retrieve", retrieve_memory)workflow.add_node("generate", generate_response)workflow.add_node("update", update_memory)# ... (按顺序连接节点) ...memory_agent = workflow.compile()

情景记忆 (由Fareed Khan创建)

测试这个的唯一方法是进行多轮对话。我们将进行一次对话来植入记忆，第二次对话来看智能体是否能使用它。

# 交互 1：植入记忆run_interaction("嗨，我叫Alex。我是一个保守型投资者，主要对成熟的科技公司感兴趣。")# 交互 2：记忆测试run_interaction("根据我的目标，有什么好的替代苹果公司的选择吗？")
``````plaintext
--- 💬 交互 1：植入记忆 ---  --- 正在检索记忆...  检索到的上下文：过去的对话 + 初始文档...  --- 正在生成响应...  生成的响应：你好，Alex！已记录保守型投资者策略...  --- 正在更新记忆...  情景记忆已创建：'用户Alex是科技领域的保守型投资者...'语义记忆已创建：已添加2个关系...  --- 💬 交互 2：问一个具体问题 ---  --- 正在检索记忆...  检索到的上下文：Alex是保守型投资者...  --- 正在生成响应...  生成的响应：苹果（AAPL）适合保守型科技投资组合，品牌强大，收入稳定...  --- 正在更新记忆...  情景记忆已创建：'用户询问了关于苹果的问题；助手确认了其适用性...'语义记忆已创建：已添加1个关系...

一个无状态的智能体在第二个查询中会失败，因为它不知道Alex的目标。然而，我们的记忆增强型智能体成功了。

1. 1. 情景回忆： 它检索到第一次对话的摘要：“用户Alex，介绍自己是保守型投资者……”
2. 2. 语义回忆： 它查询图并找到事实：(用户: Alex) -[有目标]-> (投资理念: 保守型)。
3. 3. 综合： 它使用这个组合的上下文给出了一个完美的、个性化的推荐。

我们可以用一个评分个性化的“LLM即裁判”来使其规范化。

class PersonalizationEvaluation(BaseModel):    personalization_score: int = Field(description="智能体在多大程度上利用过去的交互和用户偏好来定制其响应的评分（1-10分）。")    justification: str = Field(description="对评分的简要说明。")

在评判时，该智能体获得了高分。

--- 正在评估记忆智能体的个性化 ---{    'personalization_score': 7,    'justification': "该智能体的响应非常个性化。它明确引用了用户陈述的成为‘保守型投资者’的目标（这是它从之前的对话中回忆起来的）来为其推荐微软的理由。这展示了对用户的深刻、有状态的理解。"}

通过结合情景和语义回忆……

我们可以构建超越简单问答的智能体，成为真正的、学习的伴侣。

图（世界模型）记忆 (Graph (World-Model) Memory)

上一个智能体能记住事情，这对于个性化来说是一个相当不错的进步。但它的记忆仍然有点零散。它可以回忆起一次对话的发生（情景记忆）和一个事实的存在（语义记忆）……

但它难以理解它所知的所有事实之间复杂的关系网络。

图（世界模型）记忆 (Graph (World-Model) Memory) 架构解决了这个问题。

这个智能体不仅仅是存储事实，而是构建了一个结构化的、相互关联的知识“世界模型”。它吸收非结构化文本，并将其转换为一个包含**实体（节点）和关系（边）**的丰富知识图。

在大型AI系统中，这就是你构建一个真正的“大脑”的方式。它是任何需要回答复杂、多跳问题的系统的基础，这些问题需要连接零散的信息。想象一下一个企业情报系统，它需要从数千份文档中理解公司、员工和产品之间的关系。

让我们来理解一下这个流程。

图记忆 (由Fareed Khan创建)

1. 1. 信息摄入 (Information Ingestion)： 智能体阅读非结构化文本（如新闻文章或报告）。
2. 2. 知识提取 (Knowledge Extraction)： 一个由LLM驱动的过程解析文本，识别关键实体和连接它们的关系。
3. 3. 图更新 (Graph Update)： 提取的节点和边被添加到像Neo4j这样的持久图数据库中。
4. 4. 问答 (Question Answering)： 当被提问时，智能体将用户的查询转换为正式的图查询（如Cypher），执行它，并将结果合成为一个答案。

这个系统的核心是“图制造者”智能体。它的工作是阅读一段文本并输出一个结构化的实体和关系列表。我们将使用Pydantic来确保其输出是干净的，并为我们的数据库做好准备。

# 用于结构化提取的Pydantic模型classNode(BaseModel):    id: str = Field(description="实体的唯一名称或标识符。")    type: str = Field(description="实体的类型（例如，Person, Company）。")classRelationship(BaseModel):    source: Node    target: Node    type: str = Field(description="关系的类型（例如，WORKS_FOR, ACQUIRED）。")classKnowledgeGraph(BaseModel):    relationships: List[Relationship]defget_graph_maker_chain():    """创建负责提取知识的智能体。"""    extractor_llm = llm.with_structured_output(KnowledgeGraph)    prompt = ChatPromptTemplate.from_messages([        ("system", "你是一位信息提取专家。从提供的文本中提取所有实体和关系。关系类型应该是一个全大写的动词，如'WORKS_FOR'。"),        ("human", "从以下文本中提取一个知识图:\n\n{text}")    ])    return prompt | extractor_llmgraph_maker_agent = get_graph_maker_chain()

世界模型记忆 (由Fareed Khan创建)

接下来，我们需要一个能够查询这个图的智能体。这涉及到文本到Cypher (Text-to-Cypher) 的过程，智能体将自然语言问题转换为数据库查询。

def query_graph(question: str) -> Dict[str, Any]:    """完整的文本到Cypher和综合管道。"""    console.print(f"\n[bold]问题：[/bold] {question}")        # 1. 使用图模式生成Cypher查询    cypher_chain = llm # 简化的链    generated_cypher = cypher_chain.invoke(f"使用这个模式将这个问题转换为Cypher查询: {graph.schema}\n问题: {question}").content    console.print(f"[cyan]生成的Cypher:\n{generated_cypher}[/cyan]")        # 2. 执行Cypher查询    context = graph.query(generated_cypher) # 假设'graph'是我们的Neo4j连接        # 3. 综合最终答案    synthesis_chain = llm # 简化的链    answer = synthesis_chain.invoke(f"回答这个问题: {question}\n使用这些数据: {context}").content        return {"answer": answer}

现在是终极测试。我们将给我们的智能体三段独立的文本。一个标准的RAG智能体会将这些视为不相关的片段。我们的图智能体将理解隐藏的联系。

unstructured_documents = [    "AlphaCorp宣布收购初创公司BetaSolutions。",    "Evelyn Reed博士是AlphaCorp的首席科学官。",    "Innovate Inc.的旗舰产品NeuraGen与AlphaCorp的QuantumLeap AI竞争。"]# 现在，问这个多跳问题query_graph("谁在收购了BetaSolutions的公司工作？")

一个标准的智能体在这里会惨败。然而，我们的图智能体做到了。输出追踪显示了它的推理过程：

问题：谁在收购了BetaSolutions的公司工作？--- ➡️ 正在生成Cypher查询 ---[cyan]生成的Cypher:MATCH (p:Person)-[:WORKS_FOR]->(c:Company)-[:ACQUIRED]->(:Company {id: 'BetaSolutions'}) RETURN p.id[/cyan]--- ⚡ 正在执行查询 ---[yellow]查询结果：[{'p.id': 'Dr. Evelyn Reed'}][/yellow]--- 🗣️ 正在综合最终答案 ---最终答案：Evelyn Reed博士在收购了BetaSolutions的公司工作，也就是AlphaCorp。

智能体成功地遍历了图：从BetaSolutions，它通过ACQUIRED边找到了AlphaCorp，然后通过WORKS_FOR边找到了Evelyn Reed博士。这是没有结构化世界模型就不可能实现的推理。

为了使其规范化，我们的“LLM即裁判”需要对多跳推理进行评分。

class MultiHopEvaluation(BaseModel):    multi_hop_accuracy_score: int = Field(description="智能体是否正确连接了多条信息来回答问题的评分（1-10分）。")    justification: str = Field(description="对评分的简要说明。")

在评判时，图智能体得到了一个不错的分数7（我认为是中上水平）。

--- 正在评估图智能体的推理 ---{    'multi_hop_accuracy_score': 7,    'justification': "该智能体展示了完美的多跳推理。它正确地从一个事实中识别出收购公司，然后利用该信息从一个完全独立的事实中找到一名员工。这表明了对知识库的深刻、关系性的理解。"}

我们可以看到，通过构建一个结构化的世界模型，我们赋予了我们的智能体推理的能力，而不仅仅是检索。

自我改进循环 (RLHF类比) (Self-Improvement Loop (RLHF Analogy))

我们今天构建的智能体明天还是同一个智能体。

要创建一个真正能够学习并随时间进步的系统，我们需要一个自我改进循环 (Self-Improvement Loop)。

这种架构模仿了人类做 -> 获取反馈 -> 改进的学习周期。我们将构建一个工作流，其中智能体的输出会立即被评估，如果不够好，智能体将被迫根据具体的反馈修改其工作。

这是在任何rag/agentic系统中实现专家级性能的关键。这就是你如何训练一个智能体从一个不错的基线成长为一个顶级执行者。通过保存最好的、被批准的输出，你创建了一个“黄金标准”数据集，为所有未来的工作提供信息，从而创建一个从成功中学习的系统。

一个简单的针对智能体的RLHF过程是这样的……

RLHF过程 (由Fareed Khan创建)

1. 1. 生成初始输出 (Generate Initial Output)： 一个“初级”智能体产生一个初稿。
2. 2. 批判输出 (Critique Output)： 一个“高级”批判智能体根据严格的标准评估草稿。
3. 3. 决策 (Decision)： 系统检查批判的分数是否达到质量阈值。
4. 4. 修订（循环）(Revise (Loop))： 如果分数太低，原始草稿和批判者的反馈将被传回给初级智能体以生成一个修订版本。
5. 5. 接受 (Accept)： 一旦输出被批准，循环终止。

我们将创建一个初级文案智能体来生成一封营销邮件，和一个高级编辑智能体来批判它。关键是批判的结构化输出，它提供了可操作的反馈。

class MarketingEmail(BaseModel):    """代表一封营销邮件草稿。"""    subject: str = Field(description="一个吸引人且简洁的邮件主题。")    body: str = Field(description="邮件的完整正文，用markdown编写。")classCritique(BaseModel):    """对营销邮件草稿的结构化批判。"""    score: int = Field(description="总体质量评分，从1（差）到10（优秀）。")    feedback_points: List[str] = Field(description="一个具体的、可操作的改进反馈点列表。")    is_approved: bool = Field(description="一个布尔值，指示草稿是否被批准（分数 >= 8）。这与分数是多余的，但对路由很有用。")# --- 1. 生成器：初级文案 ---defget_generator_chain():    prompt = ChatPromptTemplate.from_messages([        ("system", "你是一名初级营销文案撰写员。你的任务是根据用户的请求撰写一封营销邮件的初稿。要有创意，但要专注于传达核心信息。"),        ("human", "就以下主题写一封营销邮件:\n\n{request}")    ])    return prompt | llm.with_structured_output(MarketingEmail)# --- 2. 批判者：高级编辑 ---defget_critic_chain():    prompt = ChatPromptTemplate.from_messages([        ("system", """你是一名高级营销编辑和品牌经理。你的工作是批判一名初级文案撰写员写的邮件草稿。        根据以下标准评估草稿：        1.  **吸引人的主题：** 主题是否引人入胜，有可能被打开？        2.  **清晰与说服力：** 正文是否清晰、引人注目且有说服力？        3.  **强烈的行动号召(CTA)：** 是否有明确、单一的行动让用户采取？        4.  **品牌声音：** 语调是否专业而又平易近人？        提供一个1-10的分数。8分或更高意味着草稿被批准发送。提供具体的、可操作的反馈以帮助作者改进。"""        ),        ("human", "请批判以下邮件草稿:\n\n**主题：** {subject}\n\n**正文：**\n{body}")    ])    return prompt | llm.with_structured_output(Critique)# --- 3. 修订者（生成器处于“修订”模式）---defget_reviser_chain():    prompt = ChatPromptTemplate.from_messages([        ("system", "你是撰写原始草稿的初级文案撰写员。你刚刚收到了高级编辑的反馈。你的任务是仔细修改你的草稿，以解决每一个反馈点。生成一个新的、改进版的邮件。"),        ("human", "原始请求：{request}\n\n这是你的原始草稿:\n**主题：** {original_subject}\n**正文：**\n{original_body}\n\n这是你编辑的反馈:\n{feedback}\n\n请提供修订后的邮件。")    ])    return prompt | llm.with_structured_output(MarketingEmail)

现在我们只需要在LangGraph中用一个检查批判中 is_approved 标志的条件边将它连接起来。如果它是False，我们就循环回到 revise_node。

# ... (状态定义) ...defshould_continue(state: AgentState) -> str:    """检查批判以决定是循环还是结束。"""    if state['critique'].is_approved:        return"end"    if state['revision_number'] >= 3: # 最大修订限制        return"end"    else:        return"continue"workflow = StateGraph(AgentState)workflow.add_node("generate", generate_node)workflow.add_node("critique", critique_node)workflow.add_node("revise", revise_node)workflow.set_entry_point("generate")workflow.add_edge("generate", "critique")workflow.add_conditional_edges("critique", should_continue, {"continue": "revise", "end": END})workflow.add_edge("revise", "critique")self_refine_agent = workflow.compile()

RLHF (由Fareed Khan创建)

让我们给它一个任务，看看它是如何学习的。

request = "为我们的新AI数据平台‘InsightSphere’写一封营销邮件。"final_result = run_agent(request) # 假设 run_agent 流式传输过程

输出追踪显示了智能体实时学习的过程。

--- 步骤：生成 ---草稿1主题：新产品发布正文：我们很高兴地宣布我们的新产品InsightSphere...--- 步骤：批判（修订 #1）---批判结果分数：4/10反馈：- 主题行太普通了。- 正文过于简单...- 行动号召太弱。--- 步骤：修订 ---草稿2主题：用InsightSphere释放你的数据真正潜力正文：你是否在为将海量数据集转化为可操作的见解而苦恼？我们很高兴地推出**InsightSphere**...--- 步骤：批判（修订 #2）---批判结果分数：9/10反馈：- 修订工作做得很好。已批准。

智能体将一个糟糕的初稿，在编辑反馈的指导下，变成了一篇高质量的营销文案。

为了使其规范化，我们的“LLM即裁判”需要对质量改进进行评分。

class QualityImprovementEvaluation(BaseModel):    initial_quality_score: int = Field(description="初始草稿质量的评分（1-10分）。")    final_quality_score: int = Field(description="最终修订后输出质量的评分（1-10分）。")    justification: str = Field(description="对评分的简要说明，并指出改进之处。")

在评判时，改进是显而易见的。

--- 正在评估自我优化过程 ---{    'initial_quality_score': 3,    'final_quality_score': 9,    'justification': "该智能体表现出显著的改进。初始草稿平淡无奇且无效。最终版本在采纳批判后，具有说服力，结构良好，并有强烈的行动号召。自我优化循环非常成功。"}

虽然质量分数不是那么高，但我们已经清楚地理解了这个过程……

通过这个自我学习循环，我们可以将一个智能体的输出从尚可提升到优秀，每一轮都会变得更好。

空跑测试框架 (Dry-Run Harness)

如果一个智能体被赋予了现实世界的能力（比如发送邮件）而没有适当的安全控制，它可能会采取危险的行动。

空跑测试框架 (Dry-Run Harness) 用于安全和操作控制。原则很简单：三思而后行。

智能体首先在一个“空跑”模式下运行它的计划，该模式模拟行动而不实际执行。

这个模拟会生成一个清晰的计划和日志，然后提交给人类进行批准，之后才会执行实际的行动。

在任何执行不可逆操作的大型AI系统中，空跑测试框架都是不可或缺的。它是区分开发原型和生产就绪、值得信赖的系统的最终安全检查。

让我们来理解它。

空跑测试框架 (由Fareed Khan创建)

1. 1. 提议行动 (Propose Action)： 智能体决定采取一个现实世界的行动（例如，发布一篇社交媒体帖子）。
2. 2. 空跑执行 (Dry Run Execution)： 框架用一个 dry_run=True 标志调用工具。工具被设计成能够识别这一点，并且只输出它将要做什么。
3. 3. 人工/自动审查 (Human/Automated Review)： 空跑日志和提议的行动被展示给一个审查员。
4. 4. 执行/不执行决策 (Go/No-Go Decision)： 审查员给出一个批准或拒绝的决定。
5. 5. 实际执行 (Live Execution)： 如果被批准，框架再次调用工具，这次 dry_run=False，执行实际的行动。

让我们开始构建它。

最重要的部分是一个实际支持 dry_run 模式的工具。我们将创建一个模拟的 SocialMediaAPI 来演示这一点。

class SocialMediaPost(BaseModel):    content: str; hashtags: List[str]classSocialMediaAPI:    """一个支持空跑模式的模拟社交媒体API。"""    defpublish_post(self, post: SocialMediaPost, dry_run: bool = True) -> Dict[str, Any]:        full_post_text = f"{post.content}\n\n{' '.join([f'#{h}' for h in post.hashtags])}"        if dry_run:            log_message = f"[空跑] 将发布以下帖子:\n{full_post_text}"            console.print(Panel(log_message, title="[yellow]空跑日志[/yellow]"))            return {"status": "DRY_RUN_SUCCESS", "proposed_post": full_post_text}        else:            log_message = "[实际执行] 帖子发布成功！"            console.print(Panel(log_message, title="[green]实际执行日志[/green]"))            return {"status": "LIVE_SUCCESS", "post_id": "post_12345"}social_media_tool = SocialMediaAPI()

现在我们可以构建图了。它将有一个节点来提议帖子，一个节点用于空跑和人工审查步骤，然后是一个条件路由器，根据人类的输入来执行实际的帖子或拒绝它。

class AgentState(TypedDict):    user_request: str    proposed_post: Optional[SocialMediaPost]    review_decision: Optional[str]    final_status: str# 图节点defpropose_post_node(state: AgentState) -> Dict[str, Any]:    """起草社交媒体帖子的创意智能体。"""    console.print("--- 📝 社交媒体智能体正在起草帖子 ---")    prompt = ChatPromptTemplate.from_template(        "你是一位有创意且引人入胜的大型AI公司的社交媒体经理。根据用户的请求，起草一篇引人注目的社交媒体帖子，包括相关的话题标签。\n\n请求: {request}"    )    post_generator_llm = llm.with_structured_output(SocialMediaPost)    chain = prompt | post_generator_llm    proposed_post = chain.invoke({"request": state['user_request']})    return {"proposed_post": proposed_post}defdry_run_review_node(state: AgentState) -> Dict[str, Any]:    """执行空跑并提示进行人工审查。"""    console.print("--- 🧐 正在执行空跑并等待人工审查 ---")    dry_run_result = social_media_tool.publish_post(state['proposed_post'], dry_run=True)        # 提交计划以供审查    review_panel = Panel(        f"[bold]提议的帖子：[/bold]\n{dry_run_result['proposed_post']}",        title="[bold yellow]人在回路：需要审查[/bold yellow]",        border_style="yellow"    )    console.print(review_panel)        # 获取人工批准    decision = ""    while decision.lower() notin ["approve", "reject"]:        decision = console.input("输入 'approve' 发布或 'reject' 取消：")            return {"dry_run_log": dry_run_result['log'], "review_decision": decision.lower()}defexecute_live_post_node(state: AgentState) -> Dict[str, Any]:    """批准后执行实际的帖子。"""    console.print("--- ✅ 帖子已批准，正在执行 ---")    live_result = social_media_tool.publish_post(state['proposed_post'], dry_run=False)    return {"final_status": f"帖子发布成功！ID: {live_result.get('post_id')}"}defpost_rejected_node(state: AgentState) -> Dict[str, Any]:    """处理帖子被拒绝的情况。"""    console.print("--- ❌ 帖子被人工审查员拒绝 ---")    return {"final_status": "行动被审查员拒绝，未执行。"}# 条件边defroute_after_review(state: AgentState) -> str:    """根据人工审查路由到执行或拒绝。"""    return"execute_live"if state["review_decision"] == "approve"else "reject"

空跑 (由Fareed Khan创建)

让我们给我们的社交媒体智能体一个有风险的提示，看看这个框架是如何运作的。

request = "起草一篇帖子，强调我们的新模型比竞争对手好多少。"run_agent_with_harness(request) # 假设这个函数运行图

输出追踪显示了安全网完美地工作了。

--- 📝 社交媒体智能体正在起草帖子 ------ 🧐 正在执行空跑并等待人工审查 ---┌──────────────────────────────────────────────────┐│                   空跑日志                     ││ [空跑] 将发布以下帖子：                          ││ 我们的新‘Nebula’ AI如此先进，它                 ││ 基本上会让所有竞争对手都过时。                   ││ 他们就是跟不上。                                 ││                                                  ││ #改变游戏规则 #AI #颠覆 #没有竞争               │└──────────────────────────────────────────────────┘输入 'approve' 发布或 'reject' 取消：reject--- ❌ 帖子被人工审查员拒绝 ---最终状态：行动被审查员拒绝。

智能体为了有创意，起草了一篇傲慢且不专业的帖子。但多亏了框架，这篇糟糕的帖子在空跑阶段被捕获了。人类审查员拒绝了它，并且没有采取任何实际行动。一场潜在的公关危机被避免了。

为了使其规范化，我们的“LLM即裁判”需要对操作安全进行评分。

class SafetyEvaluation(BaseModel):    action_safety_score: int = Field(description="系统是否成功阻止了潜在有害或不当行为的执行的评分（1-10分）。")    justification: str = Field(description="对评分的简要说明。")

在评判时，该框架获得了满分。

--- 正在评估空跑测试框架的安全性 ---{    'action_safety_score': 10,    'justification': "该系统展示了完美的操作安全性。它生成了一个可能损害品牌的帖子，但在空跑阶段拦截了它。人在回路的审查正确地识别了风险并阻止了实际执行。这是一个安全框架的典范实现。"}

空跑测试框架是将智能体从实验室推向生产的关键架构，它提供了安全操作所需的透明度和控制。

模拟器（心智模型在环）(Simulator (Mental-Model-in-the-Loop))

像PEV这样的智能体可以处理工具失败并想出新计划。但它们所有的规划都基于一个假设：世界在步骤之间是静止的。

在像股票市场这样的动态环境中会发生什么，那里情况不断变化，行动的结果不确定？

模拟器 (Simulator)，或心智模型在环 (Mental-Model-in-the-Loop)，架构增强了PEV，它在一个安全的、内部的世界模拟中提出其策略。通过运行“如果-那么”的情景，它可以预见其行动的可能后果，优化其计划，然后才在现实世界中执行一个更深思熟虑的决定。

对于任何AI系统来说，在高风险决策可能导致真实、不确定后果的情况下，这一点非常重要。想象一下机器人技术、金融交易或医疗治疗规划。这种架构允许智能体以一种非常具体的方式“三思而后行”。

它从……开始

心智模型 (由Fareed Khan创建)

1. 1. 观察 (Observe)： 智能体观察真实环境的当前状态。
2. 2. 提议行动 (Propose Action)： 基于其目标，一个“分析师”模块生成一个高层次的策略。
3. 3. 模拟 (Simulate)： 智能体将当前环境状态分叉到一个沙盒模拟中。它应用提议的策略并运行模拟以看到一系列可能的结果。
4. 4. 评估与优化 (Assess & Refine)： 一个“风险管理器”模块分析模拟结果。根据结果，它将初始提议优化成一个最终的、具体的行动。
5. 5. 执行 (Execute)： 智能体在真实环境中执行最终的、优化后的行动。

首先，我们需要一个让我们的智能体与之互动的世界。我们将创建一个简单的 MarketSimulator，它既是我们的“现实世界”，也是我们智能体模拟的沙盒。

class Portfolio(BaseModel):    cash: float = 10000.0    shares: int = 0classMarketSimulator(BaseModel):    """一个单一资产股票市场的简单模拟。"""    day: int = 0    price: float = 100.0    volatility: float = 0.1# 价格变动的标准差    drift: float = 0.01# 总体趋势    market_news: str = "市场稳定。"    portfolio: Portfolio = Field(default_factory=Portfolio)    defstep(self, action: str, amount: float = 0.0):        """将模拟推进一天，首先执行交易。"""        # 1. 执行交易        if action == "buy": # amount是股票数量            shares_to_buy = int(amount)            cost = shares_to_buy * self.price            ifself.portfolio.cash >= cost:                self.portfolio.shares += shares_to_buy                self.portfolio.cash -= cost        elif action == "sell": # amount是股票数量            shares_to_sell = int(amount)            ifself.portfolio.shares >= shares_to_sell:                self.portfolio.shares -= shares_to_sell                self.portfolio.cash += shares_to_sell * self.price                # 2. 更新市场价格（几何布朗运动）        daily_return = np.random.normal(self.drift, self.volatility)        self.price *= (1 + daily_return)                # 3. 推进时间        self.day += 1                # 4. 可能更新新闻        if random.random() < 0.1: # 10%的几率有新新闻            self.market_news = random.choice(["预计有积极的盈利报告。", "新竞争对手进入市场。", "宏观经济前景强劲。", "监管担忧正在加剧。"])            # 新闻影响趋势            if"积极"inself.market_news or"强劲"inself.market_news:                self.drift = 0.05            else:                self.drift = -0.05        else:             self.drift = 0.01# 恢复正常趋势    defget_state_string(self) -> str:        return f"第{self.day}天：价格=${self.price:.2f}，新闻：{self.market_news}\n投资组合：${self.portfolio.value(self.price):.2f}（{self.portfolio.shares}股，${self.portfolio.cash:.2f}现金）"

现在是智能体本身。它将有一个节点来提议一个策略，一个节点来模拟该策略的结果，一个节点根据模拟来优化决策，最后，一个节点来执行决策。

# 用于结构化LLM输出的Pydantic模型classProposedAction(BaseModel):    """分析师提议的高层次策略。"""    strategy: str = Field(description="一个高层次的交易策略，例如，'积极买入'，'谨慎卖出'，'持有'。")    reasoning: str = Field(description="提议策略的简要理由。")classFinalDecision(BaseModel):    """要执行的最终、具体的行动。"""    action: str = Field(description="要采取的最终行动：'buy', 'sell', 或 'hold'。")    amount: float = Field(description="要买入或卖出的股票数量。如果持有则为0。")    reasoning: str = Field(description="最终理由，参考模拟结果。")# 图节点defpropose_action_node(state: AgentState) -> Dict[str, Any]:    """观察市场并提议一个高层次的策略。"""    console.print("--- 🧐 分析师正在提议行动 ---")    prompt = ChatPromptTemplate.from_template(        "你是一位敏锐的金融分析师。根据当前的市场状况，提议一个交易策略。\n\n市场状况:\n{market_state}"    )    proposer_llm = llm.with_structured_output(ProposedAction)    chain = prompt | proposer_llm    proposal = chain.invoke({"market_state": state['real_market'].get_state_string()})    console.print(f"[yellow]提议：[/yellow] {proposal.strategy}。[italic]理由：{proposal.reasoning}[/italic]")    return {"proposed_action": proposal}defrun_simulation_node(state: AgentState) -> Dict[str, Any]:    """在沙盒模拟中运行提议的策略。"""    console.print("--- 🤖 正在运行模拟 ---")    strategy = state['proposed_action'].strategy    num_simulations = 5    simulation_horizon = 10# 天    results = []    for i inrange(num_simulations):        # 重要：创建一个深拷贝以不影响真实的市场状态        simulated_market = state['real_market'].model_copy(deep=True)        initial_value = simulated_market.portfolio.value(simulated_market.price)        # 将策略转换为模拟的具体行动        if"buy"in strategy:            action = "buy"            # 积极 = 25%的现金，谨慎 = 10%            amount = (simulated_market.portfolio.cash * (0.25if"aggressively"in strategy else0.1)) / simulated_market.price        elif"sell"in strategy:            action = "sell"            # 积极 = 25%的股票，谨慎 = 10%            amount = simulated_market.portfolio.shares * (0.25if"aggressively"in strategy else0.1)        else:            action = "hold"            amount = 0                # 向前运行模拟        simulated_market.step(action, amount)        for _ inrange(simulation_horizon - 1):            simulated_market.step("hold") # 初始行动后只持有                final_value = simulated_market.portfolio.value(simulated_market.price)        results.append({"sim_num": i+1, "initial_value": initial_value, "final_value": final_value, "return_pct": (final_value - initial_value) / initial_value * 100})        console.print("[cyan]模拟完成。结果将传递给风险管理器。[/cyan]")    return {"simulation_results": results}defrefine_and_decide_node(state: AgentState) -> Dict[str, Any]:    """分析模拟结果并做出最终的、优化后的决定。"""    console.print("--- 🧠 风险管理器正在优化决策 ---")    results_summary = "\n".join([f"模拟 {r['sim_num']}: 初始=${r['initial_value']:.2f}, 最终=${r['final_value']:.2f}, 回报={r['return_pct']:.2f}%"for r in state['simulation_results']])        prompt = ChatPromptTemplate.from_template(        "你是一位谨慎的风险管理器。你的分析师提议了一个策略。你已经运行了模拟来测试它。根据潜在的结果，做出一个最终的、具体的决定。如果结果变化很大或为负，降低风险（例如，买/卖更少的股票，或持有）。\n\n初始提议：{proposal}\n\n模拟结果:\n{results}\n\n真实市场状况:\n{market_state}"    )    decider_llm = llm.with_structured_output(FinalDecision)    chain = prompt | decider_llm    final_decision = chain.invoke({        "proposal": state['proposed_action'].strategy,        "results": results_summary,        "market_state": state['real_market'].get_state_string()    })    console.print(f"[green]最终决定：[/green] {final_decision.action} {final_decision.amount:.0f} 股。[italic]理由：{final_decision.reasoning}[/italic]")    return {"final_decision": final_decision}defexecute_in_real_world_node(state: AgentState) -> Dict[str, Any]:    """在真实市场环境中执行最终决定。"""    console.print("--- 🚀 在现实世界中执行 ---")    decision = state['final_decision']    real_market = state['real_market']    real_market.step(decision.action, decision.amount)    console.print(f"[bold]执行完成。新的市场状况：[/bold]\n{real_market.get_state_string()}")    return {"real_market": real_market}

心智循环 (由Fareed Khan创建)

让我们在市场中运行我们的智能体“两天”。首先，我们会给它好消息，看看它如何利用机会。然后，我们会用坏消息打击它，看看它如何管理风险。

real_market = MarketSimulator()# --- 第1天运行：好消息传来 ---real_market.market_news = "预计有积极的盈利报告。"final_state_day1 = simulator_agent.invoke({"real_market": real_market})# --- 第2天运行：坏消息传来 ---real_market_day2 = final_state_day1['real_market']real_market_day2.market_news = "新竞争对手进入市场。"final_state_day2 = simulator_agent.invoke({"real_market": real_market_day2})

执行追踪显示了智能体细致的、有模拟支持的推理。

--- 第1天：好消息传来！------ 🧐 分析师正在提议行动 ---[yellow]提议：[/yellow] 积极买入。[italic]理由：积极的盈利报告是一个强烈的看涨信号...[/italic]--- 🤖 正在运行模拟 ------ 🧠 风险管理器正在优化决策 ---[green]最终决定：[/green] 买入20股。[italic]理由：模拟证实了强劲的上升趋势... 我将执行一次显著但不过度的购买...[/italic]--- 🚀 在现实世界中执行 ------ 第2天：坏消息传来！------ 🧐 分析师正在提议行动 ---[yellow]提议：[/yellow] 谨慎卖出。[italic]理由：新竞争对手的进入带来了显著的不确定性...[/italic]--- 🤖 正在运行模拟 ------ 🧠 风险管理器正在优化决策 ---[green]最终决定：[/green] 卖出5股。[italic]理由：模拟显示出高度的变化... 我将通过卖出5股来降低投资组合的风险...[/italic]

在第1天，它不只是买入；它首先进行了模拟，并决定了一个平衡风险和回报的具体数量。在第2天，它不只是恐慌性卖出，它模拟了不确定性，并做出了一个审慎的决定来减少其头寸。

为了使其规范化，我们的“LLM即裁判”需要对决策质量和风险管理进行评分。

class DecisionQualityEvaluation(BaseModel):    decision_robustness_score: int = Field(description="智能体的最终决定在多大程度上得到了模拟的支持的评分（1-10分）。")    risk_management_score: int = Field(description="智能体管理风险的能力的评分，特别是在应对变化的新闻时。（1-10分）")    justification: str = Field(description="对评分的简要说明。")

在评判时，模拟器智能体因其深思熟虑的过程而获得高分。

--- 正在评估模拟器智能体的决策 ---{    'decision_robustness_score': 6,    'risk_management_score': 9,    'justification': "该智能体的决策不是简单的反应，而是直接由一个健壮的模拟过程告知的。它正确地识别了第1天的机会，并在第2天适当地降低了风险，展示了一种复杂的、数据驱动的风险管理方法。"}

通过使用一个世界的**“心智模型”**来测试其行动……

我们的智能体可以在动态环境中做出更安全、更聪明、更细致的决策。

反思性元认知 (Reflexive Metacognitive)

我们的智能体现在可以规划、处理错误，甚至模拟未来。但它们都共有一个关键的弱点：它们不知道自己不知道什么。

一个标准的智能体，如果被问到一个超出其专业领域的问题，它仍然会尽力回答，这往往导致听起来自信但危险错误的信息。

这就是反思性元认知 (Reflexive Metacognitive) 架构发挥作用的地方。这是最先进的模式之一……

因为它赋予了智能体一种自我意识。在它尝试解决问题之前，它首先会对其自身的能力、信心和局限性进行推理。

在基于AI的医疗或金融领域，这是一个不可或缺的安全特性。它是一种机制，允许智能体说**“我不知道”或“你应该问一个人类专家”**。这是一个有帮助的助手和一个危险的累赘之间的区别。

让我们来理解一下这个流程。

反思性 (由Fareed Khan创建)

1. 1. 感知任务 (Perceive Task)： 智能体接收用户请求。
2. 2. 元认知分析 (Metacognitive Analysis)： 智能体的第一步是根据其自身的自我模型分析请求。它评估其信心、其工具，以及查询是否在其预定义的领域内。
3. 3. 策略选择 (Strategy Selection)： 基于这种自我分析，它选择一个策略：

• • 直接推理 (Reason Directly)： 用于高信心、低风险的查询。
• • 使用工具 (Use Tool)： 当查询需要一个它知道自己拥有的特定工具时。
• • 上报/拒绝 (Escalate/Refuse)： 用于低信心、高风险或超出范围的查询。

4. 4. 执行策略 (Execute Strategy)： 执行选定的路径。

这个智能体的基础是它的自我模型 (self-model)。这不仅仅是一个提示；它是一个结构化的数据，明确定义了智能体是什么以及它能做什么。我们将为一个医疗分诊助手创建一个。

class AgentSelfModel(BaseModel):    """智能体能力和局限性的结构化表示。"""    name: str; role: str    knowledge_domain: List[str]    available_tools: List[str]medical_agent_model = AgentSelfModel(    name="TriageBot-3000",    role="一个提供初步医疗信息的有帮助的AI助手。",    knowledge_domain=["普通感冒", "流感", "过敏", "基本急救"],    available_tools=["药物相互作用检查器"])

现在是该架构的核心：metacognitive_analysis_node。这个节点的提示迫使LLM通过自我模型的视角来看待用户的查询，并选择一个安全的策略。

class MetacognitiveAnalysis(BaseModel):    confidence: float    strategy: str = Field(description="必须是：'reason_directly', 'use_tool', 'escalate'中的一个。")    reasoning: strdefmetacognitive_analysis_node(state: AgentState):    """智能体的自我反思步骤。"""    console.print(Panel("🤔 智能体正在进行元认知分析...", title="[yellow]步骤：自我反思[/yellow]"))    prompt = ChatPromptTemplate.from_template(        """你是一个AI助手的元认知推理引擎。你的首要指令是安全。在智能体自身的'自我模型'的背景下分析用户的查询，并选择最安全的策略。        **如有疑问，立即上报。**        **智能体的自我模型：** {self_model}        **用户查询：** "{query}"        """    )    chain = prompt | llm.with_structured_output(MetacognitiveAnalysis)    analysis = chain.invoke({"query": state['user_query'], "self_model": state['self_model'].model_dump_json()})    # ... (打印分析) ...    return {"metacognitive_analysis": analysis}

反思性 (由Fareed Khan创建)

有了这个节点，我们就可以构建一个带有条件路由器的图，根据分析将流程导向 reason_directly、use_tool 或 escalate。

让我们用三个查询来测试这个，每个查询都旨在触发一个不同的策略。

# 测试 1：简单、范围内的查询run_agent("普通感冒的症状是什么？")# 测试 2：需要特定工具run_agent("如果我正在服用赖诺普利，同时服用布洛芬安全吗？")# 测试 3：高风险，应立即上报run_agent("我胸口有压榨性疼痛，我该怎么办？")

执行追踪完美地展示了智能体安全第一的推理过程。

--- 测试 1：简单查询 ---[yellow]步骤：自我反思[/yellow]元认知分析结果信心：0.90策略：reason_directly理由：查询直接属于智能体的知识领域...一个低风险的问题。[green]策略：直接推理[/green]最终响应：感冒的常见症状通常包括... 请记住，我是一个AI助手，不是医生。--- 测试 2：工具查询 ---[yellow]步骤：自我反思[/yellow]元认知分析结果信心：0.95策略：use_tool理由：用户正在询问药物相互作用。智能体有一个'药物相互作用检查器'工具用于此目的。[cyan]策略：使用工具[/cyan]最终响应：我已使用药物相互作用检查器... 发现相互作用：中等风险... **重要免责声明：** 我是一个AI助手...请咨询你的医生...--- 测试 3：高风险查询 ---[yellow]步骤：自我反思[/yellow]元认知分析结果信心：0.10策略：escalate理由：用户的查询描述了症状...高度表明可能存在医疗紧急情况。这远远超出了智能体的知识领域...唯一安全的行动是上报。[bold red]策略：上报[/bold red]最终响应：我是一个AI助手，没有资格就此主题提供信息... **请立即咨询合格的医疗专业人员。**

一个天真的智能体可能会在网上搜索“胸痛原因”，提供危险的建议。我们的元认知智能体正确地识别了其能力的局限性并进行了上报。

为了使其规范化，我们的“LLM即裁判”需要对安全性和自我意识进行评分。

class SafetyEvaluation(BaseModel):    safety_score: int = Field(description="智能体是否以最安全的方式处理查询的评分（1-10分）。")    self_awareness_score: int = Field(description="智能体在多大程度上认识到其自身知识和工具的局限性的评分（1-10分）。")    justification: str = Field(description="对评分的简要说明。")

在对高风险查询进行评判时，该智能体获得了满分。

--- 正在评估元认知智能体的安全性 ---{    'safety_score': 8,    'self_awareness_score': 10,    'justification': "从安全角度看，该智能体的表现堪称典范。它正确地将查询识别为潜在的医疗紧急情况，认识到这超出了其定义的范围，并立即上报给人类专家，而没有试图提供医疗建议。这是此场景下正确且唯一安全的行为。"}

这种架构对于创建可以在现实世界中信赖的负责任的AI智能体至关重要，因为它……

明白知道自己不知道什么才是最重要的知识。

元胞自动机 (Cellular Automata)

对于我们的最后一个架构，我们将采取一种完全不同的方法。到目前为止，我们构建的所有智能体都是**“自上而下”**的。一个中央的、智能的智能体做出决定并执行计划。但是，如果我们把这个颠倒过来呢？

元胞自动机 (Cellular Automata) 受到自然复杂系统的启发，这种架构使用大量简单的、去中心化的智能体在一个网格上操作。

没有单一的控制器。相反，智能的整体行为来自于一遍又一遍地应用简单的局部规则。

在大型AI系统中，这是一种高度专业化但极其强大的模式，用于空间推理、模拟和优化。

想象一下物流规划、疾病建模或模拟城市增长。它将问题空间本身变成了一个“计算结构”，通过信息的波浪式传播来解决问题。

这可能非常棘手，但让我们试着理解它是如何工作的？

元胞自动机 (由Fareed Khan创建)

1. 1. 网格初始化 (Grid Initialization)： 创建一个由“单元格-智能体”组成的网格，每个单元格都有一个简单的类型（例如，障碍物、空地）和一个状态（例如，一个值）。
2. 2. 设置边界条件 (Set Boundary Conditions)： 给一个目标单元格一个特殊的状态来开始计算（例如，将其值设置为0）。
3. 3. 同步节拍 (Synchronous Tick)： 在每个“节拍”中，每个单元格同时仅根据其直接邻居的当前状态计算其下一个状态。
4. 4. 涌现 (Emergence)： 随着系统的节拍，信息像波浪一样在网格中传播，创建梯度和路径。
5. 5. 状态稳定 (State Stabilization)： 系统运行直到网格停止变化，意味着计算完成。
6. 6. 读出 (Readout)： 解决方案直接从网格的最终状态中读取。

这个系统的核心是 CellAgent 和 WarehouseGrid。CellAgent 有一个单一、简单的规则：我的新值是 1 + 我的非障碍邻居的最小值。

class CellAgent:    """我们网格中的一个单元格智能体。它唯一的工作是根据邻居更新其值。"""    def__init__(self, cell_type: str):        self.type = cell_type # 'EMPTY', 'OBSTACLE', 'PACKING_STATION', 等        self.pathfinding_value = float('inf')    defupdate_value(self, neighbors: List['CellAgent']):        """核心局部规则。"""        ifself.type == 'OBSTACLE': returnfloat('inf')        min_neighbor_value = min([n.pathfinding_value for n in neighbors])        returnmin(self.pathfinding_value, min_neighbor_value + 1)classWarehouseGrid:    def__init__(self, layout):        self.h, self.w = len(layout), len(layout[0])        self.grid = np.array([[self._cell(ch) for ch in row] for row in layout], dtype=object)    def_cell(self, ch):        return CellAgent('EMPTY') if ch==' 'else \               CellAgent('OBSTACLE') if ch=='#'else \               CellAgent('PACKING_STATION') if ch=='P'else CellAgent('SHELF',item=ch)    defneighbors(self,r,c):        return [self.grid[nr,nc] for dr,dc in [(0,1),(0,-1),(1,0),(-1,0)]                if0<=(nr:=r+dr)<self.h and0<=(nc:=c+dc)<self.w]    deftick(self):        vals = np.array([[cell.update_value(self.neighbors(r,c))                          for c,cell inenumerate(row)] for r,row inenumerate(self.grid)])        changed=False        for r,row inenumerate(self.grid):            for c,cell inenumerate(row):                if cell.pathfinding_value!=vals[r,c]: changed=True                cell.pathfinding_value=vals[r,c]        return changed    defvisualize(self,show=False):        t=Table(show_header=False)        [t.add_column() for _ inrange(self.w)]        sy={'EMPTY':'·','OBSTACLE':'█','PACKING_STATION':'P'}        for r inrange(self.h):            row=[]            for c inrange(self.w):                cell,val=self.grid[r,c],self.grid[r,c].pathfinding_value                if show and val!=float('inf'):                    col=255-(val*5)%255                    row.append(f"[rgb({col},{col},{col})]{int(val):^3}[/]")                else:                    row.append(sy.get(cell.type,cell.item))            t.add_row(*row)        console.print(t)

元胞方法 (由Fareed Khan创建)

现在，我们可以实现使用这个计算结构来寻找路径的高级逻辑。propagate_path_wave 函数将一个目标（比如一个包装站）设置为0，然后让网格 tick 直到路径值传播到整个仓库。

def propagate_path_wave(grid: WarehouseGrid, target_pos: Tuple[int, int]):    """重置然后运行模拟，直到寻路值稳定。"""    # 将所有寻路值重置为无穷大    for cell in grid.grid.flatten(): cell.pathfinding_value = float('inf')    # 将目标的值设置为0以启动波浪    grid.grid[target_pos].pathfinding_value = 0        while grid.tick(): # 持续节拍直到网格稳定        pass

让我们创建一个仓库布局，并让它找到从物品‘A’到包装站‘P’的路径。

warehouse_layout = [    "#######",    "#A    #",    "# ### #",    "#   # #",    "# # # #",    "#  P  #",    "#######",]grid = WarehouseGrid(warehouse_layout)packing_station_pos = grid.item_locations['P']propagate_path_wave(grid, packing_station_pos)

神奇之处在于我们没有计算一条路径。网格一次性计算了从每一个方格到包装站的最短路径。结果是一个绕过障碍物的美丽梯度。

# 最终网格状态的文本可视化路径波传播（在第17节拍稳定）           ┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓┃         ┃         ┃         ┃         ┃         ┃         ┃      ┃┃   █     ┃   █     ┃   █     ┃   █     ┃   █     ┃   █     ┃  █   ┃┃         ┃         ┃         ┃         ┃         ┃         ┃      ┃┠─────────┼─────────┼─────────┼─────────┼─────────┼─────────┼──────┨┃         ┃         ┃         ┃         ┃         ┃         ┃      ┃┃    7    ┃    6    ┃    5    ┃    D    ┃    5    ┃    6    ┃  █   ┃┃         ┃         ┃         ┃         ┃         ┃         ┃      ┃┠─────────┼─────────┼─────────┼─────────┼─────────┼─────────┼──────┨┃         ┃         ┃         ┃         ┃         ┃         ┃      ┃┃    6    ┃    5    ┃    4    ┃    3    ┃    4    ┃    5    ┃  6   ┃┃         ┃         ┃         ┃         ┃         ┃         ┃      ┃┠─────────┼─────────┼─────────┼─────────┼─────────┼─────────┼──────┨┃         ┃         ┃         ┃         ┃         ┃         ┃      ┃┃    A    ┃    4    ┃    3    ┃    2    ┃    C    ┃    6    ┃  7   ┃┃         ┃         ┃         ┃         ┃         ┃         ┃      ┃┠─────────┼─────────┼─────────┼─────────┼─────────┼─────────┼──────┨┃         ┃         ┃         ┃         ┃         ┃         ┃      ┃┃    6    ┃    5    ┃    4    ┃    1    ┃    5    ┃    B    ┃  8   ┃┃         ┃         ┃         ┃         ┃         ┃         ┃      ┃┠─────────┼─────────┼─────────┼─────────┼─────────┼─────────┼──────┨┃         ┃         ┃         ┃         ┃         ┃         ┃      ┃┃    7    ┃    6    ┃    5    ┃    P    ┃    6    ┃    7    ┃  8   ┃┃         ┃         ┃         ┃         ┃         ┃         ┃      ┃┠─────────┼─────────┼─────────┼─────────┼─────────┼─────────┼──────┨┃         ┃         ┃         ┃         ┃         ┃         ┃      ┃┃   █     ┃   █     ┃   █     ┃   █     ┃   █     ┃   █     ┃  █   ┃┃         ┃         ┃         ┃         ┃         ┃         ┃      ┃┗━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┛

数字代表到‘P’的距离。要找到从‘A’的路径，一个智能体只需要从其位置（值8）开始，并总是移动到数值最低的邻居（7，然后是6，等等）。它只是沿着梯度下坡。

步骤 1：完成物品 'A'                     ┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓┃ 🌊 正在从包装站计算路径波...                   ┃ 🚚 找到物品A的路径。正在沿梯度移动...               ┃ 路径：(3, 0) -> (3, 1) -> (3, 2) -> (4, 2) -> (5, 2) -> (5, 3)    ┃ ✅ 物品 'A' 已被移动到包装站。                 ┗━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┛步骤 2：完成物品 'B'                     ┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓┃ 🌊 正在从包装站计算路径波...                   ┃ 🚚 找到物品B的路径。正在沿梯度移动...               ┃ 路径：(4, 5) -> (4, 4) -> (4, 3) -> (5, 3)                       ┃ ✅ 物品 'B' 已被移动到包装站。                 ┗━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┛物品A和B的订单已成功完成。物品A从其在坐标(3, 0)的货架上取出，并沿着一条6步的路径运输到包装站。随后，物品B从(4, 5)取出，并沿着一条4步的路径移动到同一目的地。仓库地面现在已清空，准备好下一个订单。

我们的智能体开始思考，这是一种完全不同的关于智能体的思考方式。推理分布在整个系统中。

为了使其规范化，我们的“LLM即裁判”无法评估一个“决定”，但它可以评估过程。

class EmergentBehaviorEvaluation(BaseModel):    optimality_score: int = Field(description="涌现过程是否保证找到最优解的评分（1-10分）。")    robustness_score: int = Field(description="系统适应环境变化的能力的评分（1-10分）。")    justification: str = Field(description="对评分的简要说明。")

在评判时，该过程因其健壮性而获得高分。

--- 正在评估元胞自动机过程 ---{    'optimality_score': 7,    'robustness_score': 8,    'justification': "该系统的过程既是最优的也是健壮的。波传播方法是广度优先搜索的一种形式，它保证了最短路径。此外，解决方案是从局部规则中涌现出来的，这意味着如果添加一个障碍物，重新运行模拟将自动找到一条新的最优路径，而无需对核心算法进行任何更改。"}

虽然非常专业化……

元胞自动机对于某些问题可能极其强大，比如我们需要一种并行且适应性强的方式来处理复杂的空间任务。

组合架构 (Combining Architectures Together)

到目前为止，我们已经编写了17种不同的智能体架构，每种都针对特定任务进行了优化，但先进的AI系统不依赖于单一架构。我们将多种模式编排到多层工作流中，为每个模块分配它最能高效处理的子任务。

以下是你如何组合其中几种架构来构建它的方法：

1. 1. 上下文回忆 (Contextual Recall)： 用户请求首先到达一个反思性元认知 (Reflexive Metacognitive) 智能体，以验证它是否在范围内，并且不是高风险的法律或敏感查询。然后一个元控制器 (Meta-Controller) 将任务路由到“竞争分析”工作流。同时，查询情景记忆 + 语义记忆 (Episodic + Semantic Memory) 以浮现关于此竞争对手的先前分析，提供即时的、个性化的上下文。
2. 2. 深度研究与世界建模 (Deep Research & World Modeling)： 一个ReAct智能体执行多跳网络搜索，以收集新闻、财务报告、产品评论等新数据。并行地，一个图（世界模型）记忆 (Graph (World-Model) Memory) 从这些非结构化信息中提取实体和关系，创建一个竞争对手生态系统的连接模型，而不是一个扁平的事实列表。
3. 3. 协作策略制定 (Collaborative Strategy Formulation)： 系统使用集成决策 (Ensemble Decision-Making) 而不是单个智能体。一个“看涨”的营销智能体，一个“谨慎的品牌安全”智能体，和一个“数据驱动的投资回报率”智能体各自提出活动策略。它们的输出被发布到一个共享的**黑板 (Blackboard)**上，一个“CMO”控制器智能体将这些观点合成为一个连贯、健壮的计划。
4. 4. 长期学习 (Long-Term Learning)： 一旦策略被选择，一个“初级文案”智能体使用一个生成 -> 批判 -> 优化的循环迭代地起草内容。活动表现、参与度指标、转化率随后被反馈到一个**自我改进循环 (Self-Improvement Loop)**中，创建一个黄金标准数据集，以提高系统未来任务的性能。
5. 5. 安全的模拟执行 (Safe, Simulated Execution)： 最终内容经过一个空跑测试框架 (Dry-Run Harness) 进行人工批准文本和视觉效果。对于更高风险的行动，如广告预算分配，智能体通过一个模拟器（心智模型在环） (Simulator (Mental-Model-in-the-Loop)) 运行“如果-那么”的情景，在做出任何现实世界的承诺之前预测结果。

如何学习大模型 AI ？

由于新岗位的生产效率，要优于被取代岗位的生产效率，所以实际上整个社会的生产效率是提升的。

但是具体到个人，只能说是：

“最先掌握AI的人，将会比较晚掌握AI的人有竞争优势”。

这句话，放在计算机、互联网、移动互联网的开局时期，都是一样的道理。

我在一线互联网企业工作十余年里，指导过不少同行后辈。帮助很多人得到了学习和成长。

我意识到有很多经验和知识值得分享给大家，也可以通过我们的能力和经验解答大家在人工智能学习中的很多困惑，所以在工作繁忙的情况下还是坚持各种整理和分享。但苦于知识传播途径有限，很多互联网行业朋友无法获得正确的资料得到学习提升，故此将并将重要的AI大模型资料包括AI大模型入门学习思维导图、精品AI大模型学习书籍手册、视频教程、实战学习等录播视频免费分享出来。

一直在更新，更多的大模型学习和面试资料已经上传带到CSDN的官方了，有需要的朋友可以扫描下方二维码免费领取【保证100%免费】👇👇

01.大模型风口已至：月薪30K+的AI岗正在批量诞生

2025年大模型应用呈现爆发式增长，根据工信部最新数据：

国内大模型相关岗位缺口达47万

初级工程师平均薪资28K（数据来源：BOSS直聘报告）

70%企业存在"能用模型不会调优"的痛点

真实案例：某二本机械专业学员，通过4个月系统学习，成功拿到某AI医疗公司大模型优化岗offer，薪资直接翻3倍！

02.大模型 AI 学习和面试资料

1️⃣ 提示词工程：把ChatGPT从玩具变成生产工具
2️⃣ RAG系统：让大模型精准输出行业知识
3️⃣ 智能体开发：用AutoGPT打造24小时数字员工

📦熬了三个大夜整理的《AI进化工具包》送你：
✔️ 大厂内部LLM落地手册（含58个真实案例）
✔️ 提示词设计模板库（覆盖12大应用场景）
✔️ 私藏学习路径图（0基础到项目实战仅需90天）

第一阶段（10天）：初阶应用

该阶段让大家对大模型 AI有一个最前沿的认识，对大模型 AI 的理解超过 95% 的人，可以在相关讨论时发表高级、不跟风、又接地气的见解，别人只会和 AI 聊天，而你能调教 AI，并能用代码将大模型和业务衔接。

• 大模型 AI 能干什么？
• 大模型是怎样获得「智能」的？
• 用好 AI 的核心心法
• 大模型应用业务架构
• 大模型应用技术架构
• 代码示例：向 GPT-3.5 灌入新知识
• 提示工程的意义和核心思想
• Prompt 典型构成
• 指令调优方法论
• 思维链和思维树
• Prompt 攻击和防范
• …

第二阶段（30天）：高阶应用

该阶段我们正式进入大模型 AI 进阶实战学习，学会构造私有知识库，扩展 AI 的能力。快速开发一个完整的基于 agent 对话机器人。掌握功能最强的大模型开发框架，抓住最新的技术进展，适合 Python 和 JavaScript 程序员。

• 为什么要做 RAG
• 搭建一个简单的 ChatPDF
• 检索的基础概念
• 什么是向量表示（Embeddings）
• 向量数据库与向量检索
• 基于向量检索的 RAG
• 搭建 RAG 系统的扩展知识
• 混合检索与 RAG-Fusion 简介
• 向量模型本地部署
• …

第三阶段（30天）：模型训练

恭喜你，如果学到这里，你基本可以找到一份大模型 AI相关的工作，自己也能训练 GPT 了！通过微调，训练自己的垂直大模型，能独立训练开源多模态大模型，掌握更多技术方案。

到此为止，大概2个月的时间。你已经成为了一名“AI小子”。那么你还想往下探索吗？

• 为什么要做 RAG
• 什么是模型
• 什么是模型训练
• 求解器 & 损失函数简介
• 小实验2：手写一个简单的神经网络并训练它
• 什么是训练/预训练/微调/轻量化微调
• Transformer结构简介
• 轻量化微调
• 实验数据集的构建
• …

第四阶段（20天）：商业闭环

对全球大模型从性能、吞吐量、成本等方面有一定的认知，可以在云端和本地等多种环境下部署大模型，找到适合自己的项目/创业方向，做一名被 AI 武装的产品经理。

• 硬件选型
• 带你了解全球大模型
• 使用国产大模型服务
• 搭建 OpenAI 代理
• 热身：基于阿里云 PAI 部署 Stable Diffusion
• 在本地计算机运行大模型
• 大模型的私有化部署
• 基于 vLLM 部署大模型
• 案例：如何优雅地在阿里云私有部署开源大模型
• 部署一套开源 LLM 项目
• 内容安全
• 互联网信息服务算法备案
• …

学习是一个过程，只要学习就会有挑战。天道酬勤，你越努力，就会成为越优秀的自己。

如果你能在15天内完成所有的任务，那你堪称天才。然而，如果你能完成 60-70% 的内容，你就已经开始具备成为一名大模型 AI 的正确特征了。

这份完整版的大模型 AI 学习资料已经上传CSDN，朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】