探讨 ‘Biological Plausibility in Agent Design’：LangGraph 架构与人类神经系统在信息处理上的深度契合

各位同仁，各位对人工智能与认知科学交叉领域充满热情的探索者们，大家好。

今天，我们将深入探讨一个引人入胜的主题：“Biological Plausibility in Agent Design: LangGraph 架构与人类神经系统在信息处理上的深度契合”。这不仅仅是一个理论探讨，更是一场实践的演练，我们将以编程专家的视角，剖析 LangGraph 如何为我们构建更接近人类认知模式的智能体提供了强大而灵活的框架。

1. 智能体设计的演进与生物学启示

在过去的几年里，大型语言模型（LLMs）以其惊人的生成和理解能力，彻底改变了我们对人工智能的认知。然而，仅仅依靠一个巨大的“黑箱”模型来处理所有任务，往往会遇到效率、可控性和鲁棒性的瓶颈。一个单一的 LLM 就像一个拥有超凡记忆力和联想能力的个体，但缺乏结构化的思考、规划、反思和执行能力。

人类智能并非单一的巨型网络。它是一个高度模块化、层次化、并充满反馈循环的复杂系统。我们的神经系统拥有专门的区域负责感知、记忆、决策、情感和运动。这些区域协同工作，通过精确的信号传递和迭代处理，使我们能够适应复杂多变的世界。

这种生物学的深刻洞察，为我们设计下一代 AI 智能体提供了关键的启示：

• 模块化 (Modularity): 将复杂的任务分解为更小的、可管理的子任务，并分配给专门的模块处理。
• 层次化 (Hierarchy): 存在不同抽象层次的决策和处理，从低级的感知运动到高级的战略规划。
• 反馈循环 (Feedback Loops): 智能体需要能够根据环境的反馈和自身行为的结果进行调整和学习。
• 状态管理 (State Management): 智能体需要维护一个内部状态，反映其当前目标、感知和信念，以便在不同模块间传递上下文。
• 记忆系统 (Memory Systems): 区分短期工作记忆和长期知识存储，以便在需要时检索相关信息。
• 执行控制 (Executive Control): 能够进行规划、决策、优先级排序和自我监控。

这些原则并非抽象概念，而是我们今天探讨 LangGraph 的基石。

2. 人类神经系统信息处理的简化模型

为了更好地理解 LangGraph 如何与生物学概念契合，我们先对人类神经系统的信息处理机制进行一个高度简化的抽象：

人类神经系统功能	核心特征	在认知中的作用
感知输入 (Sensory Input)	接收来自外部世界的信号 (视觉、听觉、触觉等)。	提供关于环境的原始数据。
工作记忆 (Working Memory)	短期、容量有限的存储，用于当前任务的活跃信息。	维持注意力，进行即时推理和决策。
长时记忆 (Long-Term Memory)	长期、大容量的存储，包括事实（语义记忆）和事件（情景记忆）。	提供知识、经验和背景信息，用于理解世界和指导行为。
执行功能 (Executive Functions)	位于前额叶皮层，负责规划、决策、抑制冲动、任务切换等。	高级认知控制，指导目标导向行为。
处理模块 (Processing Modules)	大脑皮层不同区域（如视觉皮层、听觉皮层、语言区等）的专业化处理。	对特定类型的信息进行深入分析和转换。
反馈循环 (Feedback Loops)	神经信号在不同区域之间往复传递，进行迭代修正。	允许自我纠正、适应性学习和行为优化。
行动输出 (Motor Output)	运动皮层向肌肉发送信号，产生行为。	智能体与环境交互，实现其目标。
情感/动机系统 (Emotion/Motivation)	边缘系统等，产生内部驱动力，影响决策和优先级。	赋予行为方向和强度，评估奖励与惩罚。 (在 AI 中通常通过奖励函数或优先级机制模拟)

这个简化的模型将作为我们后续将 LangGraph 架构与生物学概念进行映射的参照系。

3. LangGraph: 赋能架构化认知的框架

LangGraph 是 LangChain 生态系统中的一个强大库，它允许我们通过定义一个有向无环图（DAG）或有向循环图来编排复杂的 LLM 驱动的工作流。其核心思想是将整个智能体行为分解为一系列可重用的“节点”（Nodes），这些节点之间通过“边”（Edges）连接，共同处理一个共享的“状态”（State）。

LangGraph 的核心优势在于：

• 状态管理： 维护一个共享的、可变的状态，在图的节点之间传递信息。
• 节点： 封装特定的逻辑，可以是 LLM 调用、工具使用、自定义函数或人类干预。
• 边： 定义节点之间的信息流，可以是无条件连接，也可以是基于状态或节点输出的条件路由。
• 循环： 允许智能体创建迭代处理、规划、反思和自我修正的循环。

正是这些特性，使得 LangGraph 成为构建具有生物学启发性智能体的理想工具。

3.1 LangGraph 的基本构造块

我们通过一个最小的 LangGraph 示例来理解其基本组件：

from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, END
from langchain_core.messages import BaseMessage

# 1. 定义图的共享状态 (Working Memory)
class AgentState(TypedDict):
    messages: Annotated[List[BaseMessage], lambda x, y: x + y] # 消息列表，新消息追加

# 2. 定义节点 (Processing Modules)
def agent_node(state: AgentState):
    """一个模拟代理人思考的节点"""
    print("代理人正在思考...")
    # 实际场景中这里会调用 LLM，例如：
    # from langchain_openai import ChatOpenAI
    # llm = ChatOpenAI(model="gpt-4o", temperature=0)
    # result = llm.invoke(state['messages'])
    # return {"messages": [result]}

    # 简化模拟：直接添加一个回复
    new_message = BaseMessage(content="好的，我理解了您的请求。", type="ai")
    return {"messages": [new_message]}

def tool_node(state: AgentState):
    """一个模拟工具调用的节点 (例如，执行搜索或API)"""
    print("工具正在执行...")
    # 实际场景中这里会调用工具，例如：
    # from langchain.tools import WikipediaQueryRun
    # tool = WikipediaQueryRun(...)
    # tool_result = tool.invoke(state['messages'][-1].content)
    # return {"messages": [BaseMessage(content=tool_result, type="tool_output")]}

    # 简化模拟：直接添加一个工具输出
    tool_output = BaseMessage(content="工具执行结果：搜索到相关信息。", type="tool")
    return {"messages": [tool_output]}

# 3. 构建图
workflow = StateGraph(AgentState)

# 添加节点
workflow.add_node("agent", agent_node)
workflow.add_node("tool", tool_node)

# 设置入口点
workflow.set_entry_point("agent")

# 添加边 (无条件连接)
workflow.add_edge("agent", "tool") # 代理人思考后，工具执行
workflow.add_edge("tool", END)     # 工具执行后，流程结束

# 编译图
app = workflow.compile()

# 运行图
initial_state = {"messages": [BaseMessage(content="请帮我查找关于 LangGraph 的最新资料。", type="human")]}
final_state = app.invoke(initial_state)

print("n最终状态:")
for msg in final_state["messages"]:
    print(f"{msg.type}: {msg.content}")

输出示例：

代理人正在思考...
工具正在执行...

最终状态:
human: 请帮我查找关于 LangGraph 的最新资料。
ai: 好的，我理解了您的请求。
tool: 工具执行结果：搜索到相关信息。

在这个简单的例子中：

• AgentState 是我们的工作记忆，它在不同节点间传递。
• agent_node 和 tool_node 是两个处理模块，它们执行特定的认知功能。
• add_node, set_entry_point, add_edge 定义了信息流的神经通路。

4. 解构认知：LangGraph 与人类神经系统的深度映射

现在，让我们更深入地将人类认知的各个方面，与 LangGraph 的架构元素进行一一映射。

4.1 感知输入与行动输出：工具（Tools）

• 人类神经系统： 感知系统（视觉、听觉、触觉等）将外部世界的原始信号转换为神经冲动；运动皮层控制肌肉运动，产生行动。
• LangGraph 映射： LangChain 的工具（Tools）是智能体与外部世界交互的接口。它们可以是搜索引擎、API 调用、数据库查询、文件读写，甚至是执行代码。当一个 LangGraph 节点（通常是 LLM 节点）决定需要外部信息或需要执行某个操作时，它会调用一个工具，这就像大脑根据需要激活不同的感知器官或运动指令。

代码示例：集成一个简单的搜索工具

from typing import TypedDict, Annotated, List, Union
from langgraph.graph import StateGraph, END
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, ToolMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
import operator

# 1. 定义工具
@tool
def search_web(query: str) -> str:
    """在网络上搜索信息。"""
    print(f"执行搜索：{query}")
    # 实际应用中这里会调用搜索引擎API
    if "LangGraph" in query:
        return "LangGraph是一个用于构建LLM代理的图框架，支持循环和状态管理。"
    elif "Python" in query:
        return "Python是一种流行的编程语言，广泛应用于AI、Web开发等领域。"
    else:
        return "未找到相关信息。"

# 2. 定义图的共享状态
class AgentState(TypedDict):
    messages: Annotated[List[BaseMessage], operator.add] # 消息列表，新消息追加

# 3. 定义LLM节点
llm = ChatOpenAI(model="gpt-4o", temperature=0)

def llm_agent_node(state: AgentState):
    """代理人思考和决定是否使用工具的节点。"""
    current_messages = state["messages"]
    print(f"n--- LLM Agent 接收到消息：{current_messages[-1].content} ---")

    # 绑定工具到LLM，使其能够调用工具
    llm_with_tools = llm.bind_tools([search_web])

    response = llm_with_tools.invoke(current_messages)

    # LangChain 的 LLM 响应中，如果 LLM 决定使用工具，会返回 FunctionCall 或 ToolCall 消息
    # 这里我们直接返回 LLM 的响应，后续的工具节点会处理
    return {"messages": [response]}

# 4. 定义工具调用节点
def tool_node(state: AgentState):
    """根据LLM的决定执行工具的节点。"""
    print("n--- Tool Node 正在执行 ---")
    last_message = state["messages"][-1]

    if not last_message.tool_calls:
        raise ValueError("Tool node expects a message with tool_calls.")

    tool_outputs = []
    for tool_call in last_message.tool_calls:
        tool_name = tool_call["name"]
        tool_args = tool_call["args"]

        if tool_name == "search_web":
            output = search_web.invoke(tool_args["query"])
            tool_outputs.append(ToolMessage(content=output, tool_call_id=tool_call["id"]))
        else:
            tool_outputs.append(ToolMessage(content=f"未知工具: {tool_name}", tool_call_id=tool_call["id"]))

    return {"messages": tool_outputs}

# 5. 定义路由函数 (Executive Function: Decision Making)
def route_agent(state: AgentState):
    """根据LLM的输出决定下一步是继续LLM思考还是执行工具。"""
    print("n--- Route Agent 正在决策 ---")
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        print("决策：需要执行工具。")
        return "call_tool"
    else:
        print("决策：LLM 思考完成，可以结束或继续LLM思考。")
        return "end_or_continue_llm" # 这里我们选择结束，实际可能继续LLM思考

# 6. 构建图
workflow = StateGraph(AgentState)

workflow.add_node("llm_agent", llm_agent_node)
workflow.add_node("tool_executor", tool_node)

workflow.set_entry_point("llm_agent")

workflow.add_conditional_edges(
    "llm_agent",     # 从 llm_agent 节点出发
    route_agent,     # 使用 route_agent 函数进行条件路由
    {
        "call_tool": "tool_executor",       # 如果需要调用工具，则到 tool_executor
        "end_or_continue_llm": END, # 如果LLM思考完成，则结束
    }
)

workflow.add_edge("tool_executor", "llm_agent") # 工具执行后，将结果反馈给LLM进行进一步思考

app = workflow.compile()

# 运行图
initial_query = "请问LangGraph是什么？"
final_state = app.invoke({"messages": [HumanMessage(content=initial_query)]})

print("n--- 最终消息链 ---")
for msg in final_state["messages"]:
    print(f"{msg.type}: {msg.content}")

输出示例 (部分)：

--- LLM Agent 接收到消息：请问LangGraph是什么？ ---

--- Route Agent 正在决策 ---
决策：需要执行工具。

--- Tool Node 正在执行 ---
执行搜索：LangGraph

--- LLM Agent 接收到消息：LangGraph是一个用于构建LLM代理的图框架，支持循环和状态管理。 ---

--- Route Agent 正在决策 ---
决策：LLM 思考完成，可以结束或继续LLM思考。

--- 最终消息链 ---
human: 请问LangGraph是什么？
ai_tool_call: name='search_web' args={'query': 'LangGraph'} id='call_...'
tool: LangGraph是一个用于构建LLM代理的图框架，支持循环和状态管理。
ai: LangGraph是一个用于构建LLM代理的图框架，它支持循环和状态管理，使得构建复杂的、多步骤的AI工作流成为可能。

在这个例子中，llm_agent_node 扮演了决策中心和语言理解/生成模块的角色。它接收用户输入，并决定是否需要调用外部工具。tool_executor 节点则模拟了实际的行动执行者，它根据 LLM 的指令调用工具，并将结果反馈给 LLM。route_agent 函数是执行功能的体现，它根据 LLM 的输出（是否有工具调用）来决定信息流向。

4.2 工作记忆：图的状态（Graph State）

• 人类神经系统： 工作记忆是一个临时存储和处理信息的系统，它在认知任务中扮演核心角色，如推理、理解和学习。它的容量有限，但信息活跃且易于访问。
• LangGraph 映射： LangGraph 的图状态（Graph State）直接对应于智能体的工作记忆。它是一个在所有节点之间共享和传递的字典或 TypedDict。每个节点都可以读取当前状态，并返回一个更新后的状态。这种机制确保了智能体在整个任务执行过程中保持上下文，并积累信息。

在上面的示例中，AgentState 类就是我们的工作记忆，它存储了所有的 messages。Annotated[List[BaseMessage], operator.add] 的使用尤其巧妙，它模拟了工作记忆的追加特性：新的感知（消息）不断地被添加到现有记忆的末尾。

4.3 长时记忆：RAG 与专门的记忆节点

• 人类神经系统： 长时记忆存储着我们所有的知识和经验，包括语义记忆（事实知识）和情景记忆（个人事件）。它容量巨大，但检索需要特定的线索。
• LangGraph 映射： LangGraph 本身不直接提供长时记忆，但它允许我们通过专门的节点来集成检索增强生成（RAG）系统，从而模拟长时记忆。
  • RAG 节点： 智能体可以调用一个 RAG 节点，该节点负责查询一个向量数据库（模拟长时记忆库），检索相关文档或知识片段，然后将其作为上下文提供给 LLM 节点。
  • 记忆更新节点： 智能体也可以有一个节点，负责将新的重要信息或学习到的经验存储到长时记忆库中。

代码示例：RAG 增强的 LangGraph 代理

import operator
from typing import TypedDict, Annotated, List, Union
from langgraph.graph import StateGraph, END
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, ToolMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.documents import Document

# 1. 模拟长时记忆库 (FAISS Vector Store)
docs = [
    Document(page_content="LangGraph是一个用于构建LLM代理的图框架，支持循环和状态管理。"),
    Document(page_content="LangChain是一个用于开发由LLMs驱动的应用程序的框架。"),
    Document(page_content="Python是一种流行的编程语言，广泛应用于AI、Web开发等领域。"),
    Document(page_content="神经网络是机器学习模型的一种，模仿人脑结构。"),
]
text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=0)
splits = text_splitter.split_documents(docs)
vectorstore = FAISS.from_documents(documents=splits, embedding=OpenAIEmbeddings())
retriever = vectorstore.as_retriever()

# 2. 定义 RAG 工具
@tool
def retrieve_knowledge(query: str) -> str:
    """从长期知识库中检索相关信息。"""
    print(f"执行知识检索：{query}")
    results = retriever.invoke(query)
    return "n".join([doc.page_content for doc in results])

# 3. 定义图的共享状态
class AgentState(TypedDict):
    messages: Annotated[List[BaseMessage], operator.add]
    # 添加一个用于存储检索到的知识的字段，作为临时的上下文增强
    retrieved_context: Annotated[List[str], operator.add]

# 4. 定义LLM节点
llm = ChatOpenAI(model="gpt-4o", temperature=0)

def llm_agent_node_rag(state: AgentState):
    """代理人思考，并结合检索到的知识进行回答。"""
    current_messages = state["messages"]
    retrieved_context = "n".join(state.get("retrieved_context", [])) # 获取并组合上下文

    print(f"n--- LLM Agent 接收到消息：{current_messages[-1].content} ---")
    print(f"--- 结合检索到的上下文：n{retrieved_context} ---")

    # 将检索到的上下文添加到LLM的输入中
    messages_with_context = [
        HumanMessage(content=f"以下是额外的背景信息，请参考：n{retrieved_context}"),
        *current_messages
    ] if retrieved_context else current_messages

    llm_with_tools = llm.bind_tools([retrieve_knowledge])
    response = llm_with_tools.invoke(messages_with_context)

    return {"messages": [response]}

# 5. 定义工具调用节点 (处理 RAG 工具)
def tool_node_rag(state: AgentState):
    """根据LLM的决定执行工具的节点。"""
    print("n--- Tool Node 正在执行 ---")
    last_message = state["messages"][-1]

    if not last_message.tool_calls:
        raise ValueError("Tool node expects a message with tool_calls.")

    tool_outputs = []
    new_retrieved_context = []

    for tool_call in last_message.tool_calls:
        tool_name = tool_call["name"]
        tool_args = tool_call["args"]

        if tool_name == "retrieve_knowledge":
            output = retrieve_knowledge.invoke(tool_args["query"])
            tool_outputs.append(ToolMessage(content=output, tool_call_id=tool_call["id"]))
            new_retrieved_context.append(output) # 将检索结果也添加到工作记忆中，供LLM下次使用
        else:
            tool_outputs.append(ToolMessage(content=f"未知工具: {tool_name}", tool_call_id=tool_call["id"]))

    return {"messages": tool_outputs, "retrieved_context": new_retrieved_context} # 返回检索结果到状态中

# 6. 定义路由函数 (Executive Function: Decision Making)
def route_agent_rag(state: AgentState):
    """根据LLM的输出决定下一步是继续LLM思考还是执行工具。"""
    print("n--- Route Agent 正在决策 ---")
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        print("决策：需要执行工具。")
        return "call_tool"
    else:
        print("决策：LLM 思考完成。")
        return END

# 7. 构建图
workflow_rag = StateGraph(AgentState)

workflow_rag.add_node("llm_agent_rag", llm_agent_node_rag)
workflow_rag.add_node("tool_executor_rag", tool_node_rag)

workflow_rag.set_entry_point("llm_agent_rag")

workflow_rag.add_conditional_edges(
    "llm_agent_rag",
    route_agent_rag,
    {
        "call_tool": "tool_executor_rag",
        END: END,
    }
)

workflow_rag.add_edge("tool_executor_rag", "llm_agent_rag") # 工具执行后，将结果反馈给LLM进行进一步思考

app_rag = workflow_rag.compile()

# 运行图
initial_query_rag = "LangChain和LangGraph有什么关系？"
final_state_rag = app_rag.invoke({"messages": [HumanMessage(content=initial_query_rag)], "retrieved_context": []})

print("n--- 最终消息链 ---")
for msg in final_state_rag["messages"]:
    print(f"{msg.type}: {msg.content}")

输出示例 (部分)：

--- LLM Agent 接收到消息：LangChain和LangGraph有什么关系？ ---
--- 结合检索到的上下文：
 ---

--- Route Agent 正在决策 ---
决策：需要执行工具。

--- Tool Node 正在执行 ---
执行知识检索：LangChain和LangGraph

--- LLM Agent 接收到消息：LangChain是一个用于开发由LLMs驱动的应用程序的框架。
LangGraph是一个用于构建LLM代理的图框架，支持循环和状态管理。 ---
--- 结合检索到的上下文：
LangGraph是一个用于构建LLM代理的图框架，支持循环和状态管理。
LangChain是一个用于开发由LLMs驱动的应用程序的框架。 ---

--- Route Agent 正在决策 ---
决策：LLM 思考完成。

--- 最终消息链 ---
human: LangChain和LangGraph有什么关系？
ai_tool_call: name='retrieve_knowledge' args={'query': 'LangChain和LangGraph'} id='call_...'
tool: LangGraph是一个用于构建LLM代理的图框架，支持循环和状态管理。
LangChain是一个用于开发由LLMs驱动的应用程序的框架。
ai: LangChain是一个用于开发由LLMs驱动的应用程序的框架，而LangGraph是LangChain生态系统中的一个库，专门用于通过图结构构建复杂的、有状态的LLM代理，特别支持循环和更精细的控制流。可以理解为LangGraph是LangChain家族中用于编排高级代理的工具。

这个例子展示了智能体如何通过 retrieve_knowledge 工具访问其“长时记忆”，并将检索到的信息添加到其“工作记忆”（retrieved_context 字段）中，以便 LLM 在生成最终响应时能利用这些上下文。

4.4 执行功能：规划与决策（LLM Nodes & Conditional Edges）

• 人类神经系统： 前额叶皮层是执行功能的核心，负责规划、决策、问题解决、目标设定和行为抑制。它根据当前情境和长期目标来选择合适的行动路径。
• LangGraph 映射：
  • LLM 节点： 专门的 LLM 节点可以被设计为“规划者”或“决策者”。通过精心设计的提示，它们可以分析当前状态，生成行动计划，或评估多个选项并选择最佳路径。
  • 条件边（Conditional Edges）： 这是 LangGraph 模拟执行功能最直接的方式。路由函数（route_agent）根据 LLM 节点的输出或当前状态的特定属性，动态地决定下一步要激活哪个节点。这就像大脑根据当前的感知和内部评估，激活不同的神经通路。

代码示例：一个简单的规划-执行-评估循环

import operator
from typing import TypedDict, Annotated, List, Union
from langgraph.graph import StateGraph, END
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, ToolMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langchain_core.prompts import PromptTemplate

# 1. 定义工具 (简化)
@tool
def execute_task(task_description: str) -> str:
    """执行一个给定的任务，并返回其结果。"""
    print(f"执行任务：{task_description}")
    if "搜索" in task_description:
        return f"成功搜索到关于 '{task_description.replace('搜索', '').strip()}' 的信息。"
    elif "总结" in task_description:
        return f"成功总结了关于 '{task_description.replace('总结', '').strip()}' 的内容。"
    else:
        return f"任务 '{task_description}' 执行完成。"

# 2. 定义图的共享状态
class PlanningAgentState(TypedDict):
    messages: Annotated[List[BaseMessage], operator.add]
    plan: Annotated[str, operator.setitem] # 当前的计划
    task_output: Annotated[str, operator.setitem] # 任务执行结果
    iterations: int # 迭代次数，用于控制循环

# 3. 定义LLM
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# 4. 规划节点 (Executive Function: Planning)
def plan_node(state: PlanningAgentState):
    """LLM 代理人根据目标和当前状态制定或修正计划。"""
    current_messages = state["messages"]
    last_user_message = next((m.content for m in reversed(current_messages) if m.type == "human"), "")

    plan_prompt = PromptTemplate.from_template(
        """你是一个任务规划专家。根据以下用户请求和当前的任务执行情况（如果有），请制定或修正一个清晰的、一步步的计划。
        如果任务已经完成，请直接回答“完成”。

        用户请求: {user_request}
        当前计划 (如果存在): {current_plan}
        最新任务执行结果 (如果存在): {task_output}

        请提供下一步的具体行动计划，或者直接宣布“完成”。
        """
    )

    response = llm.invoke(plan_prompt.format(
        user_request=last_user_message,
        current_plan=state.get("plan", "无"),
        task_output=state.get("task_output", "无")
    ))

    new_plan = response.content
    print(f"n--- 规划节点：生成新计划 --- n{new_plan}")

    return {"plan": new_plan, "messages": [AIMessage(content=f"规划：{new_plan}")]}

# 5. 执行节点 (Action Execution)
def execute_node(state: PlanningAgentState):
    """LLM 代理人根据计划执行任务。"""
    current_plan = state["plan"]
    print(f"n--- 执行节点：正在执行计划 --- n{current_plan}")

    # 这里LLM可以解析计划，决定调用哪个工具
    # 为了简化，我们假设 execute_task 工具可以直接处理计划内容

    llm_with_tools = llm.bind_tools([execute_task])

    # 代理人思考如何将计划映射到工具调用
    tool_call_prompt = PromptTemplate.from_template(
        """你是一个任务执行者。根据以下计划，决定是否需要调用 'execute_task' 工具。
        如果需要，请生成相应的工具调用。如果不需要，或者计划已经完成，请直接回复“完成”。

        当前计划: {current_plan}

        如果计划是“完成”，请直接回复“完成”。
        """
    )

    response = llm_with_tools.invoke(tool_call_prompt.format(current_plan=current_plan))

    if response.tool_calls:
        tool_output_messages = []
        for tc in response.tool_calls:
            task_desc = tc["args"]["task_description"]
            output = execute_task.invoke(task_desc)
            tool_output_messages.append(ToolMessage(content=output, tool_call_id=tc["id"]))

        task_output = "n".join([m.content for m in tool_output_messages])
        return {"task_output": task_output, "messages": [response] + tool_output_messages}
    else:
        # 如果LLM没有决定调用工具，说明它认为任务已完成或不需要工具
        return {"task_output": "无", "messages": [response]}

# 6. 路由函数 (Executive Function: Decision Making/Control Flow)
def route_decision(state: PlanningAgentState):
    """根据计划和执行结果决定下一步。"""
    iterations = state.get("iterations", 0) + 1
    state["iterations"] = iterations # 更新迭代次数

    if iterations > 3: # 防止无限循环
        print("n--- 路由决策：达到最大迭代次数，终止。 ---")
        return END

    current_plan = state["plan"]
    last_message = state["messages"][-1]

    if current_plan == "完成":
        print("n--- 路由决策：计划已完成，终止。 ---")
        return END

    # 如果执行节点返回了工具输出，或者LLM自己生成了回复，说明需要重新规划或评估
    if state.get("task_output") or (isinstance(last_message, AIMessage) and not last_message.tool_calls):
        print("n--- 路由决策：任务执行或LLM回复，返回规划节点进行评估。 ---")
        return "plan"
    else:
        print("n--- 路由决策：继续执行任务。 ---")
        return "execute" # 继续执行

# 7. 构建图
workflow_planning = StateGraph(PlanningAgentState)

workflow_planning.add_node("plan", plan_node)
workflow_planning.add_node("execute", execute_node)

workflow_planning.set_entry_point("plan")

workflow_planning.add_conditional_edges(
    "plan",
    lambda state: "execute" if state["plan"] != "完成" else END, # 如果计划是完成，则结束，否则进入执行
    {"execute": "execute", END: END}
)

workflow_planning.add_conditional_edges(
    "execute",
    route_decision, # 执行完后，根据路由函数决定下一步
    {
        "plan": "plan",
        "execute": "execute", # 如果需要继续执行，则回到执行节点（例如，多步工具调用）
        END: END
    }
)

app_planning = workflow_planning.compile()

# 运行图
initial_query_planning = "请帮我搜索LangGraph的最新特性，并总结其中的关键点。"
final_state_planning = app_planning.invoke({"messages": [HumanMessage(content=initial_query_planning)], 
                                            "plan": "", "task_output": "", "iterations": 0})

print("n--- 最终状态消息链 ---")
for msg in final_state_planning["messages"]:
    print(f"{msg.type}: {msg.content}")

输出示例 (部分):

--- 规划节点：生成新计划 --- 
1. 搜索LangGraph的最新特性。
2. 总结搜索结果中的关键点。
3. 宣布任务完成。

--- 路由决策：任务执行或LLM回复，返回规划节点进行评估。 ---

--- 执行节点：正在执行计划 --- 
1. 搜索LangGraph的最新特性。
2. 总结搜索结果中的关键点。
3. 宣布任务完成。

执行任务：搜索LangGraph的最新特性

--- 路由决策：任务执行或LLM回复，返回规划节点进行评估。 ---

--- 规划节点：生成新计划 --- 
1. 搜索LangGraph的最新特性。（已完成）
2. 总结搜索结果中的关键点。
3. 宣布任务完成。

--- 路由决策：任务执行或LLM回复，返回规划节点进行评估。 ---

--- 执行节点：正在执行计划 --- 
1. 搜索LangGraph的最新特性。（已完成）
2. 总结搜索结果中的关键点。
3. 宣布任务完成。

执行任务：总结搜索结果中的关键点

--- 路由决策：任务执行或LLM回复，返回规划节点进行评估。 ---

--- 规划节点：生成新计划 --- 
完成

--- 路由决策：计划已完成，终止。 ---

--- 最终状态消息链 ---
human: 请帮我搜索LangGraph的最新特性，并总结其中的关键点。
ai: 规划：1. 搜索LangGraph的最新特性。
2. 总结搜索结果中的关键点。
3. 宣布任务完成。
ai_tool_call: name='execute_task' args={'task_description': '搜索LangGraph的最新特性'} id='call_...'
tool: 成功搜索到关于 'LangGraph的最新特性' 的信息。
ai: 规划：1. 搜索LangGraph的最新特性。（已完成）
2. 总结搜索结果中的关键点。
3. 宣布任务完成。
ai_tool_call: name='execute_task' args={'task_description': '总结搜索结果中的关键点'} id='call_...'
tool: 成功总结了关于 '搜索结果中的关键点' 的内容。
ai: 规划：完成

这个例子展示了一个典型的规划-执行-评估循环。plan_node 模拟了大脑的前额叶皮层，负责制定和修正计划。execute_node 模拟了运动皮层和工具使用，负责执行计划中的具体任务。route_decision 函数则是执行控制的关键，它根据任务的进展和结果，决定是继续执行，还是返回规划阶段进行调整，直到任务完成。这种循环结构是人类解决复杂问题时，迭代思考和试错的直接体现。

4.5 反思与自我修正：循环（Cycles）

• 人类神经系统： 我们在面对错误或不确定性时，会进行反思，评估过去的决策和行动，然后修正策略。这种内省和自我纠正能力是高级认知的重要组成部分。
• LangGraph 映射： LangGraph 的循环（Cycles）是实现反思和自我修正的关键。通过将节点和边配置成一个闭环，智能体可以在完成某个阶段的任务后，重新回到一个“反思”或“评估”节点。这个节点（通常是一个 LLM）会分析当前状态、之前的行动和结果，然后决定是继续当前路径、调整计划、请求更多信息，还是宣布完成。

上述的规划-执行-评估循环就是反思和自我修正的一个具体实例。每次 execute_node 完成后，路由都会将控制权交回 plan_node，让它有机会根据最新结果重新评估和调整整体计划，这正是“反思”的体现。

4.6 情感与动机系统：状态驱动的优先级与路由（State-driven Priority & Routing）

• 人类神经系统： 情感和动机（如恐惧、好奇、奖励追求）深刻影响我们的注意力和决策。它们为行为提供驱动力，并帮助我们评估情境的紧迫性和重要性。
• LangGraph 映射： 虽然 LangGraph 不直接模拟生物学意义上的情感，但我们可以通过在 AgentState 中引入表示动机或优先级的变量，并通过条件路由来模拟其影响。
  • 例如，可以有一个 urgency_score 或 satisfaction_level 字段。
  • 当 urgency_score 达到某个阈值时，路由函数可以强制智能体转向一个“紧急响应”节点，而不是继续常规规划。
  • 当 satisfaction_level 足够高时，智能体可能会决定终止任务。

代码示例：引入一个简单的“紧急性”状态

import operator
from typing import TypedDict, Annotated, List, Union
from langgraph.graph import StateGraph, END
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langchain_openai import ChatOpenAI
from langchain_core.prompts import PromptTemplate

# 1. 定义图的共享状态，加入紧急性评分
class UrgentAgentState(TypedDict):
    messages: Annotated[List[BaseMessage], operator.add]
    urgency_score: Annotated[int, operator.setitem] # 紧急性评分，0-100
    task_status: Annotated[str, operator.setitem] # 任务状态

# 2. 定义LLM
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# 3. 任务处理节点
def regular_task_node(state: UrgentAgentState):
    """处理常规任务的节点。"""
    print(f"n--- 常规任务节点：处理中，当前紧急性：{state['urgency_score']} ---")

    # 模拟任务处理，并可能根据紧急性调整行为
    response = llm.invoke(f"正在处理常规任务：{state['messages'][-1].content}。 紧急性评分为 {state['urgency_score']}。")

    # 假设常规任务会降低紧急性
    new_urgency = max(0, state['urgency_score'] - 10) 
    new_task_status = "处理中"

    return {"messages": [AIMessage(content=response.content)], 
            "urgency_score": new_urgency, 
            "task_status": new_task_status}

# 4. 紧急响应节点
def urgent_response_node(state: UrgentAgentState):
    """处理紧急情况的节点。"""
    print(f"n--- 紧急响应节点：紧急处理，当前紧急性：{state['urgency_score']} ---")

    response = llm.invoke(f"正在紧急处理请求：{state['messages'][-1].content}。 紧急性评分为 {state['urgency_score']}。")

    # 假设紧急响应会显著降低紧急性
    new_urgency = 0 
    new_task_status = "紧急已处理"

    return {"messages": [AIMessage(content=response.content)], 
            "urgency_score": new_urgency, 
            "task_status": new_task_status}

# 5. 初始评估节点 (Decision/Motivation)
def initial_assessment_node(state: UrgentAgentState):
    """评估用户请求的紧急性。"""
    print(f"n--- 初始评估节点：评估请求紧急性 ---")
    user_message = state["messages"][-1].content

    # LLM 评估紧急性
    urgency_prompt = PromptTemplate.from_template(
        """根据以下用户请求，请评估其紧急性，并给出一个0到100的评分。
        0表示完全不紧急，100表示非常紧急，需要立即处理。
        只回复一个整数。

        用户请求: {request}
        """
    )

    response = llm.invoke(urgency_prompt.format(request=user_message))
    try:
        urgency = int(response.content.strip())
    except ValueError:
        urgency = 50 # 默认值

    print(f"评估结果：紧急性评分 {urgency}")
    return {"urgency_score": urgency, "messages": [AIMessage(content=f"已评估紧急性：{urgency}")]}

# 6. 路由函数 (Motivational Control)
def route_by_urgency(state: UrgentAgentState):
    """根据紧急性评分决定流程。"""
    print(f"n--- 路由决策：根据紧急性 {state['urgency_score']} 决定 ---")
    if state["urgency_score"] >= 70: # 阈值
        print("决策：转入紧急响应。")
        return "urgent_response"
    elif state["urgency_score"] > 0:
        print("决策：转入常规任务处理。")
        return "regular_task"
    else:
        print("决策：任务完成或不紧急，结束。")
        return END

# 7. 构建图
workflow_urgent = StateGraph(UrgentAgentState)

workflow_urgent.add_node("initial_assessment", initial_assessment_node)
workflow_urgent.add_node("regular_task", regular_task_node)
workflow_urgent.add_node("urgent_response", urgent_response_node)

workflow_urgent.set_entry_point("initial_assessment")

workflow_urgent.add_conditional_edges(
    "initial_assessment",
    route_by_urgency,
    {
        "regular_task": "regular_task",
        "urgent_response": "urgent_response",
        END: END
    }
)

workflow_urgent.add_conditional_edges(
    "regular_task",
    route_by_urgency, # 常规任务处理后，重新评估是否仍需处理或结束
    {
        "regular_task": "regular_task",
        "urgent_response": "urgent_response", # 尽管在常规任务，紧急性可能又提高了
        END: END
    }
)

workflow_urgent.add_edge("urgent_response", END) # 紧急响应后直接结束

app_urgent = workflow_urgent.compile()

# 运行图
print("--- 场景一：低紧急性请求 ---")
final_state_low_urgency = app_urgent.invoke({
    "messages": [HumanMessage(content="请帮我整理一下今天的待办事项。")],
    "urgency_score": 0, "task_status": "开始"
})
print("n--- 最终状态消息链 (低紧急性) ---")
for msg in final_state_low_urgency["messages"]:
    print(f"{msg.type}: {msg.content}")

print("nn--- 场景二：高紧急性请求 ---")
final_state_high_urgency = app_urgent.invoke({
    "messages": [HumanMessage(content="系统出现严重错误，立即停止所有操作！")],
    "urgency_score": 0, "task_status": "开始"
})
print("n--- 最终状态消息链 (高紧急性) ---")
for msg in final_state_high_urgency["messages"]:
    print(f"{msg.type}: {msg.content}")

输出示例 (部分)：

--- 场景一：低紧急性请求 ---

--- 初始评估节点：评估请求紧急性 ---
评估结果：紧急性评分 30

--- 路由决策：根据紧急性 30 决定 ---
决策：转入常规任务处理。

--- 常规任务节点：处理中，当前紧急性：30 ---

--- 路由决策：根据紧急性 20 决定 ---
决策：转入常规任务处理。

--- 常规任务节点：处理中，当前紧急性：20 ---

--- 路由决策：根据紧急性 10 决定 ---
决策：转入常规任务处理。

--- 常规任务节点：处理中，当前紧急性：10 ---

--- 路由决策：根据紧急性 0 决定 ---
决策：任务完成或不紧急，结束。

--- 最终状态消息链 (低紧急性) ---
human: 请帮我整理一下今天的待办事项。
ai: 已评估紧急性：30
ai: 正在处理常规任务：请帮我整理一下今天的待办事项。 紧急性评分为 30。
ai: 正在处理常规任务：请帮我整理一下今天的待办事项。 紧急性评分为 20。
ai: 正在处理常规任务：请帮我整理一下今天的待办事项。 紧急性评分为 10。

--- 场景二：高紧急性请求 ---

--- 初始评估节点：评估请求紧急性 ---
评估结果：紧急性评分 95

--- 路由决策：根据紧急性 95 决定 ---
决策：转入紧急响应。

--- 紧急响应节点：紧急处理，当前紧急性：95 ---

--- 最终状态消息链 (高紧急性) ---
human: 系统出现严重错误，立即停止所有操作！
ai: 已评估紧急性：95
ai: 正在紧急处理请求：系统出现严重错误，立即停止所有操作！ 紧急性评分为 95。

通过在 AgentState 中引入 urgency_score 并在路由函数 route_by_urgency 中检查它，我们成功模拟了动机系统对行为的优先级排序和控制。这使得智能体能够根据其内部“感受”（紧急性）动态地调整其处理流程。

5. 综合实例：一个多模态研究代理

让我们将上述概念融会贯通，构建一个更复杂的、具有生物学启发性的研究代理。这个代理能够：

1. 接收用户查询。
2. 规划研究步骤。
3. 利用工具（模拟感知与行动）执行研究。
4. 根据研究结果进行反思和评估。
5. 在需要时检索长期知识。
6. 最终生成一份研究报告。

import operator
from typing import TypedDict, Annotated, List, Union
from langgraph.graph import StateGraph, END
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, ToolMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.documents import Document
from langchain_core.prompts import PromptTemplate

# --- 1. 模拟长时记忆库 (RAG Retriever) ---
knowledge_base_docs = [
    Document(page_content="LangGraph是一个用于构建LLM代理的图框架，支持循环和状态管理，由LangChain开发。"),
    Document(page_content="LangChain是一个用于开发由LLMs驱动的应用程序的框架，提供了许多模块和工具。"),
    Document(page_content="Python是一种流行的编程语言，广泛应用于AI、Web开发、数据科学等领域。"),
    Document(page_content="RAG (Retrieval Augmented Generation) 是一种结合检索系统和生成模型的技术，用于增强LLM的知识。"),
    Document(page_content="多模态AI是指能够处理和理解多种类型数据（如文本、图像、音频）的AI系统。"),
    Document(page_content="Agentic Workflow (代理工作流) 是指LLM能够自主规划、调用工具并根据反馈进行迭代的复杂任务处理方式。"),
]
text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=0)
splits = text_splitter.split_documents(knowledge_base_docs)
vectorstore = FAISS.from_documents(documents=splits, embedding=OpenAIEmbeddings())
long_term_memory_retriever = vectorstore.as_retriever()

# --- 2. 定义工具 (模拟感知与行动) ---
@tool
def search_internet(query: str) -> str:
    """在互联网上搜索最新信息。"""
    print(f"n[Tool Call] Searching internet for: '{query}'")
    # 模拟真实搜索结果
    if "LangGraph" in query and "latest features" in query:
        return "最新LangGraph特性包括：更强大的循环控制、异步支持、与LangServe的深度集成、新的调试工具。"
    elif "multi-modal AI" in query:
        return "多模态AI的最新进展包括：视觉-语言模型（VLMs）、跨模态生成、AI驱动的艺术创作。"
    else:
        return f"互联网搜索结果：未能找到关于 '{query}' 的特定最新信息，但找到了通用信息。"

@tool
def retrieve_from_memory(query: str) -> str:
    """从代理的长期记忆库中检索相关知识。"""
    print(f"n[Tool Call] Retrieving from long-term memory for: '{query}'")
    results = long_term_memory_retriever.invoke(query)
    if results:
        return "n".join([doc.page_content for doc in results])
    return "长期记忆库中未找到相关信息。"

# --- 3. 定义图的共享状态 (Working Memory) ---
class ResearchAgentState(TypedDict):
    user_query: str
    research_plan: Annotated[List[str], operator.add] # 逐步的计划
    research_findings: Annotated[List[str], operator.add] # 收集到的研究结果
    current_step: int # 当前计划执行到的步骤索引
    iterations: int # 循环迭代次数，防止无限循环
    final_report: str # 最终报告
    messages: Annotated[List[BaseMessage], operator.add] # 通用消息历史

# --- 4. 定义LLM ---
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# --- 5. 节点定义 ---

# 5.1 规划节点 (Executive Function: Planning)
def plan_research_node(state: ResearchAgentState):
    """根据用户查询和当前进展，规划或修正研究步骤。"""
    print("n--- [Node] Planning Research ---")
    user_query = state["user_query"]
    current_plan = "n".join(state["research_plan"]) if state["research_plan"] else "无"
    findings = "n".join(state["research_findings"]) if state["research_findings"] else "无"

    plan_prompt = PromptTemplate.from_template(
        """你是一个研究规划专家。根据以下用户查询和已经收集到的信息，请制定一个详细的、分步骤的研究计划。
        如果已有计划，请根据新信息进行修正。如果所有研究步骤都已完成且可以撰写最终报告，请在计划的最后一步写上 "撰写最终报告"。

        用户查询: {user_query}
        当前已收集信息: {findings}
        当前研究计划 (如果有): {current_plan}

        请输出一个编号列表作为研究计划。例如:
        1. 步骤一
        2. 步骤二
        ...
        """
    )

    response = llm.invoke(plan_prompt.format(user_query=user_query, findings=findings, current_plan=current_plan))
    new_plan_lines = response.content.strip().split('n')
    new_plan = [line.strip() for line in new_plan_lines if line.strip()]

    print(f"Generated Plan:n{response.content}")
    return {"research_plan": new_plan, "current_step": 0, "messages": [AIMessage(content=f"规划了研究计划。")]}

# 5.2 执行研究步骤节点 (Action Execution & Perception)
def execute_research_step_node(state: ResearchAgentState):
    """执行当前计划中的一个研究步骤，可能通过工具调用。"""
    print(f"n--- [Node] Executing Research Step {state['current_step'] + 1} ---")
    current_plan = state["research_plan"]
    step_index = state["current_step"]

    if step_index >= len(current_plan):
        # 应该通过路由避免这种情况，但作为安全措施
        print("错误：尝试执行超出计划范围的步骤。")
        return {"messages": [AIMessage(content="错误：尝试执行超出计划范围的步骤。")]}

    step_description = current_plan[step_index]
    print(f"Current Step: {step_description}")

    # LLM决定使用哪个工具来执行此步骤
    llm_with_tools = llm.bind_tools([search_internet, retrieve_from_memory])

    tool_prompt = PromptTemplate.from_template(
        """你是一个研究助手，负责执行研究计划中的一步。
        根据以下研究步骤，决定是使用 'search_internet' 工具来获取最新外部信息，
        还是使用 'retrieve_from_memory' 工具来查询内部长期记忆库。
        如果此步骤是“撰写最终报告”，请直接回复“FINAL_REPORT”。

        研究步骤: {step_description}
        """
    )

    response = llm_with_tools.invoke(tool_prompt.format(step_description=step_description))

    if "FINAL_REPORT" in response.content:
        print("识别到‘撰写最终报告’步骤。")
        return {"current_step": step_index + 1, "messages": [AIMessage(content="准备撰写最终报告。")]} # 标记为已处理此步骤

    if response.tool_calls:
        tool_output_messages = []
        findings = []
        for tc in response.tool_calls:
            tool_name = tc["name"]
            tool_args = tc["args"]

            output = ""
            if tool_name == "search_internet":
                output = search_internet.invoke(tool_args["query"])
            elif tool_name == "retrieve_from_memory":
                output = retrieve_from_memory.invoke(tool_args["query"])
            else:
                output = f"未知工具: {tool_name}"

            tool_output_messages.append(ToolMessage(content=output, tool_call_id=tc["id"]))
            findings.append(f"<{tool_name} result> {output}")

        print(f"Tool execution results: {findings}")
        return {"research_findings": findings, 
                "current_step": step_index + 1, # 步进到下一个计划
                "messages": [response] + tool_output_messages}
    else:
        # 如果LLM没有决定调用工具，可能直接生成了结果
        print(f"LLM directly responded for step: {response.content}")
        return {"research_findings": [response.content], 
                "current_step": step_index + 1,
                "messages": [response]}

# 5.3 撰写最终报告节点 (Output Generation)
def generate_final_report_node(state: ResearchAgentState):
    """根据所有收集到的研究发现，生成最终的研究报告。"""
    print("n--- [Node] Generating Final Report ---")
    user_query = state["user_query"]
    all_findings = "n".join(state["research_findings"])

    report_prompt = PromptTemplate.from_template(
        """你是一个资深研究员，请根据以下用户请求和收集到的研究发现，撰写一份结构清晰、内容详尽的研究报告。

        用户查询: {user_query}
        所有研究发现:
        {all_findings}

        请撰写最终报告。
        """
    )

    response = llm.invoke(report_prompt.format(user_query=user_query, all_findings=all_findings))
    print("Final Report Generated.")
    return {"final_report": response.content, "messages": [AIMessage(content="最终报告已生成。")]}

# 5.4 路由决策节点 (Executive Function: Control Flow & Reflection)
def route_research_flow(state: ResearchAgentState):
    """根据当前状态和计划进展，决定下一步流程。"""
    print(f"n--- [Node] Routing Research Flow (Iteration {state['iterations'] + 1}) ---")
    state["iterations"] += 1 # 增加迭代计数

    if state["iterations"] > 5: # 防止无限循环，设置最大迭代次数
        print("Max iterations reached. Ending.")
        return END

    current_plan = state["research_plan"]
    current_step_index = state["current_step"]

    # 检查是否已完成所有规划步骤，并且最后一步是“撰写最终报告”
    if current_step_index >= len(current_plan) and 
       any("撰写最终报告" in step for step in current_plan):
        print("All research steps executed. Proceeding to final report.")
        return "generate_report"
    elif current_step_index < len(current_plan):
        print("Next step in plan available. Continuing execution.")
        return "execute_step"
    else:
        # 如果没有更多计划步骤，但也没有明确的“撰写最终报告”指令，可能需要重新规划
        print("No more steps in current plan, or plan needs refinement. Re-planning.")
        return "plan_research"

# --- 6. 构建图 ---
workflow_research = StateGraph(ResearchAgentState)

workflow_research.add_node("plan_research", plan_research_node)
workflow_research.add_node("execute_step", execute_research_step_node)
workflow_research.add_node("generate_report", generate_final_report_node)

workflow_research.set_entry_point("plan_research")

# 规划后，总是进入路由决策，以便评估下一步是执行还是结束
workflow_research.add_edge("plan_research", "route_research_flow") 

# 执行步骤后，也进入路由决策
workflow_research.add_edge("execute_step", "route_research_flow")

# 路由决策根据结果导向
workflow_research.add_conditional_edges(
    "route_research_flow",
    route_research_flow, # 路由函数本身
    {
        "execute_step": "execute_step",
        "plan_research": "plan_research",
        "generate_report": "generate_report",
        END: END
    }
)

# 报告生成后结束
workflow_research.add_edge("generate_report", END)

app_research = workflow_research.compile()

# --- 7. 运行代理 ---
initial_research_query = "请研究多模态AI的最新进展，以及LangGraph如何应用于Agentic Workflow的设计。"
final_research_state = app_research.invoke({
    "user_query": initial_research_query,
    "research_plan": [],
    "research_findings": [],
    "current_step": 0,
    "iterations": 0,
    "final_report": "",
    "messages": [HumanMessage(content=initial_research_query)]
})

print("n--- 最终研究报告 ---")
print(final_research_state["final_report"])

print("n--- 最终消息链 (部分) ---")
for msg in final_research_state["messages"][-5:]: # 只打印最后5条消息
    print(f"{msg.type}: {msg.content}")

输出示例 (部分)：

--- [Node] Planning Research ---
Generated Plan:
1. 搜索“多模态AI最新进展”。
2. 搜索“LangGraph应用于Agentic Workflow”。
3. 结合搜索结果，分析多模态AI与LangGraph在Agentic Workflow中的融合潜力。
4. 撰写最终报告。

--- [Node] Routing Research Flow (Iteration 1) ---
Next step in plan available. Continuing execution.

--- [Node] Executing Research Step 1 ---
Current Step: 1. 搜索“多模态AI最新进展”。

[Tool Call] Searching internet for: '多模态AI最新进展'
Tool execution results: ['<search_internet result> 多模态AI的最新进展包括：视觉-语言模型（VLMs）、跨模态生成、AI驱动的艺术创作。']

--- [Node] Routing Research Flow (Iteration 2) ---
Next step in plan available. Continuing execution.

--- [Node] Executing Research Step 2 ---
Current Step: 2. 搜索“LangGraph应用于Agentic Workflow”。

[Tool Call] Searching internet for: 'LangGraph应用于Agentic Workflow'
Tool execution results: ["<search_internet result> 互联网搜索结果：未能找到关于 'LangGraph应用于Agentic Workflow' 的特定最新信息，但找到了通用信息。"]

--- [Node] Routing Research Flow (Iteration 3) ---
Next step in plan available. Continuing execution.

--- [Node] Executing Research Step 3 ---
Current Step: 3. 结合搜索结果，分析多模态AI与LangGraph在Agentic Workflow中的融合潜力。

LLM directly responded for step: 在多模态AI的最新进展中，视觉-语言模型（VLMs）和跨模态生成技术日益成熟。LangGraph作为构建复杂LLM代理的框架，其循环和状态管理能力使其非常适合编排多模态AI的Agentic Workflow。例如，一个LangGraph代理可以先通过视觉模块（工具）处理图像输入，然后将图像描述和用户指令传递给LLM进行规划，再调用生成模块（工具）生成跨模态内容，并根据用户反馈进行迭代优化。

--- [Node] Routing Research Flow (Iteration 4) ---
Next step in plan available. Continuing execution.

--- [Node] Executing Research Step 4 ---
Current Step: 4. 撰写最终报告。
识别到‘撰写最终报告’步骤。

--- [Node] Routing Research Flow (Iteration 5) ---
All research steps executed. Proceeding to final report.

--- [Node] Generating Final Report ---
Final Report Generated.

--- 最终研究报告 ---
## 多模态AI的最新进展与LangGraph在Agentic Workflow设计中的应用研究报告

### 摘要
本报告旨在探讨多模态人工智能的最新发展趋势，并分析LangGraph框架如何有效地应用于设计和实现基于大型语言模型（LLM）的Agentic Workflow，以融合多模态AI能力。研究发现，多模态AI正朝着视觉-语言模型（VLMs）和跨模态生成等方向深入发展，而LangGraph以其独特的图结构、状态管理和循环机制，为构建能够自主规划、执行、反思和利用多模态工具的智能代理提供了强大的架构支持。

### 1. 多模态AI的最新进展
多模态人工智能是当前AI领域的热点之一，其目标是使AI系统能够理解和处理来自不同模态（如文本、图像、音频、视频等）的信息。近期的主要进展包括：

*   **视觉-语言模型（VLMs）**: 这类模型能够同时处理图像和文本信息，实现图像描述生成、视觉问答、图像内容理解等任务。例如，GPT-4o等模型已经展示了强大的视觉理解能力。
*   **跨模态生成**: AI现在能够根据一种模态的输入生成另一种模态的内容，如文本生成图像（Text-to-Image，如DALL-E, Midjourney）、文本生成视频、图像生成文本描述等。
*   **AI驱动的艺术创作**: 多模态AI在艺术和设计领域的应用日益广泛，能够辅助甚至自主创作图像、音乐和故事。

这些进展使得AI系统能够更全面地感知和理解世界，并以更丰富的方式与人类互动。

### 2. LangGraph在Agentic Workflow设计中的应用
Agentic Workflow是指LLM能够自主地进行规划、决策、调用工具、执行任务，并根据反馈进行迭代和修正的复杂任务处理流程。LangGraph作为LangChain生态系统中的一个高级框架，其核心优势在于：

*   **图结构与模块化**: LangGraph允许开发者将复杂的代理行为分解为一系列独立的“节点”（Nodes），每个节点可以是一个LLM调用、一个工具执行、一个自定义函数或一个决策点。这与人类大脑的模块化处理机制高度相似，不同的脑区负责不同的认知功能。
*   **状态管理**: LangGraph维护一个共享的“状态”（State），它在所有节点之间传递上下文信息，如用户查询、当前计划、已收集的发现等。这类似于人类的“工作记忆”，确保了代理在整个任务执行过程中保持连贯性。
*   **循环与反馈机制**: LangGraph支持图中的循环（Cycles），这对于实现代理的规划-执行-反思（Plan-Execute-Reflect）循环至关重要。代理可以根据任务执行的结果，回到规划节点修正策略，或在遇到问题时进行自我纠正，这模拟了人类解决问题时的迭代思考过程。
*   **条件路由**: 通过条件边（Conditional Edges），代理可以根据当前状态或LLM的输出动态地选择下一步的行动路径，这体现了高级的执行功能和决策能力。

### 3. 多模态AI与LangGraph在Agentic Workflow中的融合潜力
LangGraph的架构与多模态AI的结合具有巨大的潜力：

*   **多模态感知代理**: 可以设计LangGraph节点来调用不同的多模态工具（如图像识别API、语音转文本API），将各种模态的输入转换为LLM可理解的文本描述。这些描述随后进入代理的工作记忆，供LLM进行高级推理。
*   **多模态内容生成代理**: 代理可以根据LLM的规划，调用多模态生成工具（如文本生成图像模型、文本生成语音模型），创造出结合多种模态的复杂输出。
*   **动态交互与反馈**: LangGraph的循环机制使得代理能够处理多模态的实时反馈。例如，一个代理可以根据用户对生成图像的口头反馈，重新规划并调整生成参数。
*   **复杂的跨模态任务**: 结合LangGraph的规划和执行能力，可以构建能够处理如“根据视频内容生成摘要并创建相关图片”或“分析产品图像和用户评论以生成营销文案”等复杂跨模态任务的代理。

### 结论
多模态AI的快速发展正在拓展人工智能的能力边界。而LangGraph以其灵活的图结构和强大的状态管理能力，提供了一个理想的框架来编排这些多模态能力，构建出更具生物学启发性的智能代理。通过将感知、记忆、规划、执行和反思等认知功能模块化并组织成动态工作流，我们能够设计出更健壮、更智能、更能适应复杂现实世界的AI系统。未来的研究将进一步探索如何自动化这些图结构的生成，以及如何让代理在运行时自主地学习和调整其内部架构，从而实现更深层次的生物学 plausible AI。

这个综合示例展示了一个具备规划、执行、感知（通过工具）、记忆（通过RAG）、反思（通过循环和路由）以及最终报告生成能力的智能体。它通过迭代的方式，逐步完成复杂的任务，这与人类解决问题时的认知过程高度一致。

6. 生物学启发式智能体设计的优势

采用 LangGraph 这种具有生物学启发性的架构来设计智能体，带来了诸多显著优势：

• 鲁棒性 (Robustness): 模块化设计使得系统更不容易因单一组件失效而崩溃。当某个工具调用失败时，智能体可以通过反思节点重新规划，而不是直接失败。
• 可解释性 (Interpretability): 图结构清晰地展示了信息流和决策路径，使得我们更容易理解智能体“为什么”会做出某个决策或采取某个行动，这对于调试和信任至关重要。
• 灵活性与可扩展性 (Flexibility & Scalability): 轻松添加新的工具（感知/行动能力）、新的LLM节点（认知功能）或修改现有节点的逻辑，而无需重构整个系统。
• 涌现智能 (Emergent Intelligence): 简单的模块通过复杂的交互和循环，能够产生出远超单个组件能力的复杂、智能的行为，这与大脑中神经元网络的涌现现象异曲同工。
• 减少幻觉与提高可靠性 (Reduced Hallucination & Increased Reliability): 通过将任务分解、引入工具进行事实核查、以及建立反思循环，智能体可以更有效地验证信息，减少LLM的“幻觉”，提高输出的可靠性。
• 更接近人类认知模式 (Closer to Human Cognition): 这种设计范式使得AI智能体在处理复杂任务时，其行为模式更接近人类的思考过程，从而提升用户体验和人机协作效率。

7. 挑战与未来方向

尽管 LangGraph 提供了一个令人兴奋的范式，但在构建生物学启发式智能体时，我们仍面临挑战：

• 设计复杂性： 随着节点和循环的增多，图的设计和调试会变得复杂。如何自动化或半自动化图的构建是一个研究方向。
• 状态一致性与管理： 确保 AgentState 在复杂流转中保持一致性，以及如何有效地管理和压缩工作记忆，以适应LLM的上下文窗口限制。
• “硬编码”生物学： 当前我们仍然在手动设计这些“神经通路”和“脑区”的逻辑。未来的目标是让智能体能够自主地学习和演化其内部架构。
• 跨模态能力的深度融合： 虽然可以通过工具集成多模态能力，但更深层次的、在LLM内部进行多模态推理和生成，仍是前沿研究课题。
• 长期记忆的动态更新与遗忘： 如何模拟人类记忆的动态特性，包括知识的持续学习、更新、以及有选择的遗忘。

未来的研究将继续探索如何将认知科学的最新发现融入AI agent的设计中，例如，更复杂的注意力机制、强化学习驱动的决策优化、以及更精细的情感和动机模型。随着技术的发展，LangGraph这类框架将不断演进，为我们提供更强大的工具，去构建真正智能和适应性强的AI系统。

结语

今天，我们共同探索了 LangGraph 架构如何与人类神经系统的信息处理原理深度契合。从模块化的节点到动态的条件边，从共享的工作记忆到迭代的反馈循环，LangGraph 为我们构建具备规划、执行、反思和学习能力的智能体，提供了一条充满希望的路径。这不仅仅是技术上的创新，更是我们理解智能本质，并将其应用于人工智能领域的深刻尝试。生物学的启发，正引领我们走向更智能、更具生命力的AI未来。