本文是Deep Agents with LangGraph
系列教程的学习记录。
第一部分是定义了一个计算工具，来让ai调用，并返回最终结果。这个使用了一些过程和方法，记录一下自己的心得。

代码仓库

notebook内容

第一部分主要以notebook这个教程的内容来组织。

前面的一些文档，可以自己搜索，注册，并获取这些apikey并保存到.env文件中。

加载环境

首先，我们要给出.env文件。系列教程中，需要使用下面着这些相关工具的apikey。第一个部分没有使用TVLY_API_KEY（进行网络搜索的代理）

DEEPSEEK_API_KEY =
TVLY_API_KEY=
LANGSMITH_API_KEY=
LANGSMITH_TRACING=true
LANGSMITH_PROJECT=deep-agents-from-scratch

自动查找 …/.env 文件。若找到该文件，将从其中加载环境变量，并覆盖当前Shell中的任何变量。若未找到，则使用当前Shell中已有的变量。

import os

from dotenv import load_dotenv

load_dotenv(os.path.join("..", ".env"), override=True)

创建 React Agent - 预构建版本

在本课程中，你将构建一个 深度智能体。我们将基于 LangGraph 的“预构建”智能体抽象层进行开发，这将大幅简化代码。本节课你将学习预构建的 ReAct 智能体，内容包括：

• 什么是 ReAct 智能体
• 我们实现的功能特性及更多学习资源
  • 使用工具构建智能体
  • 图结构、状态和消息机制
  • 通过工具访问和修改状态
  • 🪝 钩子函数与结构化响应

什么是 ReAct 智能体

Reason and Act

你将使用 LangGraph 开源的 create_react_agent（参见此处）抽象层。ReAct 智能体是一种采用“推理与行动”框架的AI智能体，它将思维链推理与外部工具使用相结合。该框架因论文《ReAct: 语言模型中推理与行动的协同作用》而广受欢迎。

该智能体包含三个核心组件：大型语言模型、可使用的工具集、以及提供指令的提示模板。

LLM 在循环中运行。每次迭代时，它会分析上下文（包含可用工具列表），决定是否需要调用工具。若需要，则选择具体工具并生成工具调用指令。该指令会被发送至工具节点执行，工具节点执行后将结果（观察值）返回给LLM。LLM 根据观察结果决定下一步行动。此循环将持续直到满足停止条件——通常是当智能体认为不再需要调用更多工具时。

注意：create_react_agent 即将在V1版本中升级！它将并入 LangChain 并更名为简化的 create_agent。

LangGraph 预构建 create_react_agent 的核心功能

create_react_agent 抽象层具备以下特性。虽然本课程不会全部使用这些功能，但它们充分展示了使用 create_react_agent 构建项目的优势（了解更多请点击这里）：

• 记忆集成：原生支持_短期_（基于会话）和_长期_（跨会话持久化）记忆，使聊天机器人和助手具备状态保持能力
• 人工干预控制：执行可无限期暂停_以等待人工反馈——不同于受限于实时交互的WebSocket方案。这允许在工作流任意节点进行异步审批、修正或干预
• 流式支持：实时流式传输智能体状态、模型令牌、工具输出或混合数据流
• 部署工具链：包含无基础设施依赖的部署工具。LangGraph Platform 支持测试、调试和部署
  • Studio：用于检查和调试工作流的可视化IDE
  • LangSmith：追踪与评估工具
  • 支持多种生产环境部署方案

create-react-agent 功能完善，支持多种输入格式，允许高度定制。既可作为简单的智能体-工具循环，也可通过前后置钩子和/或结构化输出进行定制化扩展。

本课程将使用基础模式，我认为现在的dify和coze应该是实现了对这些步骤的封装。

# 导入工具需要使用的包，部分包可能在之后就会更新
from typing import Annotated, List, Literal, Union
from langchain_core.messages import ToolMessage
from langchain_core.tools import InjectedToolCallId, tool
from langgraph.prebuilt import InjectedState
from langgraph.types import Command



from langchain_openai import ChatOpenAI  # 导入ChatOpenAI
from IPython.display import Image, display
from langgraph.prebuilt import create_react_agent

#定义工具
@tool
def calculator(
    operation: Literal["add","subtract","multiply","divide"],
    a: Union[int, float],
    b: Union[int, float],
) -> Union[int, float]:
    """Define a two-input calculator tool.

    Arg:
        operation (str): The operation to perform ('add', 'subtract', 'multiply', 'divide').
        a (float or int): The first number.
        b (float or int): The second number.
        
    Returns:
        result (float or int): the result of the operation
    Example
        Divide: result   = a / b
        Subtract: result = a - b
    """
    if operation == 'divide' and b == 0:
        return {"error": "Division by zero is not allowed."}

    # Perform calculation
    if operation == 'add':
        result = a + b
    elif operation == 'subtract':
        result = a - b
    elif operation == 'multiply':
        result = a * b
    elif operation == 'divide':
        result = a / b
    else: 
        result = "unknown operation"
    return result



# Create agent using create_react_agent directly

SYSTEM_PROMPT = "You are a helpful arithmetic assistant who is an expert at using a calculator."

# 使用ChatOpenAI配置DeepSeek API
model = ChatOpenAI(
    model="deepseek-chat",  # DeepSeek模型名称
    temperature=0.0,
    openai_api_key=os.getenv("DEEPSEEK_API_KEY"),  # 替换为你的DeepSeek API密钥
    openai_api_base="https://api.deepseek.com/v1",  # DeepSeek API基础URL
    max_tokens=2048
)

#定义工具集
tools = [calculator]

# Create agent
agent = create_react_agent(
    model,
    tools,
    prompt=SYSTEM_PROMPT,
    #state_schema=AgentState,  # default
).with_config({"recursion_limit": 20})  #recursion_limit limits the number of steps the agent will run
#限制智能体执行的步骤。

# Show the agent
display(Image(agent.get_graph(xray=True).draw_mermaid_png()))

执行后，会得到如下的结构

type(agent)
langgraph.graph.state.CompiledStateGraph

图结构、状态与消息机制

稍后你将运行这个智能体，现在让我们深入了解一下图结构。

如果你想查看实现细节，可以在这里查看 create_react_agent 的源代码。如果想尝试自己构建简化版本，可以参考基础课程：LangGraph入门，模块1，第6课，智能体。

定义智能体：如之前所示，定义智能体时需要提供：模型、一个或多个工具、"系统"提示词，以及状态模式（默认使用
AgentState
，其主要内容是一个消息列表）。（调用详情参见此处）

在底层，这会定义并编译上面展示的LangGraph图结构。一个重要细节是，工具节点是另一个预构建组件——ToolNode，详见此处。工具节点会执行LLM消息中识别的所有工具并返回结果。

调用智能体：
现在让我们调用智能体，看看会得到什么结果！
代码额外用到了utils。

"""Utility functions for displaying messages and prompts in Jupyter notebooks."""

import json

from rich.console import Console
from rich.panel import Panel
from rich.text import Text

console = Console()


def format_message_content(message):
    """Convert message content to displayable string."""
    parts = []
    tool_calls_processed = False

    # Handle main content
    if isinstance(message.content, str):
        parts.append(message.content)
    elif isinstance(message.content, list):
        # Handle complex content like tool calls (Anthropic format)
        for item in message.content:
            if item.get("type") == "text":
                parts.append(item["text"])
            elif item.get("type") == "tool_use":
                parts.append(f"\n🔧 Tool Call: {item['name']}")
                parts.append(f"   Args: {json.dumps(item['input'], indent=2, ensure_ascii=False)}")
                parts.append(f"   ID: {item.get('id', 'N/A')}")
                tool_calls_processed = True
    else:
        parts.append(str(message.content))

    # Handle tool calls attached to the message (OpenAI format) - only if not already processed
    if (
        not tool_calls_processed
        and hasattr(message, "tool_calls")
        and message.tool_calls
    ):
        for tool_call in message.tool_calls:
            parts.append(f"\n🔧 Tool Call: {tool_call['name']}")
            parts.append(f"   Args: {json.dumps(tool_call['args'], indent=2, ensure_ascii=False)}")
            parts.append(f"   ID: {tool_call['id']}")

    return "\n".join(parts)


def format_messages(messages):
    """Format and display a list of messages with Rich formatting."""
    for m in messages:
        msg_type = m.__class__.__name__.replace("Message", "")
        content = format_message_content(m)

        if msg_type == "Human":
            console.print(Panel(content, title="🧑 Human", border_style="blue"))
        elif msg_type == "Ai":
            console.print(Panel(content, title="🤖 Assistant", border_style="green"))
        elif msg_type == "Tool":
            console.print(Panel(content, title="🔧 Tool Output", border_style="yellow"))
        else:
            console.print(Panel(content, title=f"📝 {msg_type}", border_style="white"))


def format_message(messages):
    """Alias for format_messages for backward compatibility."""
    return format_messages(messages)


def show_prompt(prompt_text: str, title: str = "Prompt", border_style: str = "blue"):
    """Display a prompt with rich formatting and XML tag highlighting.

    Args:
        prompt_text: The prompt string to display
        title: Title for the panel (default: "Prompt")
        border_style: Border color style (default: "blue")
    """
    # Create a formatted display of the prompt
    formatted_text = Text(prompt_text)
    formatted_text.highlight_regex(r"<[^>]+>", style="bold blue")  # Highlight XML tags
    formatted_text.highlight_regex(
        r"##[^#\n]+", style="bold magenta"
    )  # Highlight headers
    formatted_text.highlight_regex(
        r"###[^#\n]+", style="bold cyan"
    )  # Highlight sub-headers

    # Display in a panel for better presentation
    console.print(
        Panel(
            formatted_text,
            title=f"[bold green]{title}[/bold green]",
            border_style=border_style,
            padding=(1, 2),
        )
    )

# more expressive runner
async def stream_agent(agent, query, config=None):
    async for graph_name, stream_mode, event in agent.astream(
        query,
        stream_mode=["updates", "values"], 
        subgraphs=True,
        config=config
    ):
        if stream_mode == "updates":
            print(f'Graph: {graph_name if len(graph_name) > 0 else "root"}')
            
            node, result = list(event.items())[0]
            print(f'Node: {node}')
            
            for key in result.keys():
                if "messages" in key:
                    # print(f"Messages key: {key}")
                    format_messages(result[key])
                    break
        elif stream_mode == "values":
            current_state = event

    return current_state

from utils import format_messages

result1 = agent.invoke(
    {
        "messages": [
            {
                "role": "user",
                "content": "What is 3.1 * 4.2?",
            }
        ],
    }
)

format_messages(result1["messages"])

这个调用，可以显示，ai调用工具，工具输出的整个过程。

result = agent.invoke(
    {
        "messages": [
            {
                "role": "user",
                "content": "What is 55+98*100?",
            }
        ],
    }
)

format_messages(result["messages"])

我进行两个步骤 的运算。

补充部分

感谢你提供的背景信息！这很好地解释了 LangGraph 中智能体的内部工作机制。让我基于这些信息来详细说明整个流程：

智能体定义的核心组件

根据你提到的源码，create_react_agent 主要包含：

1. 状态定义 (AgentState)

# 从源码中可以看到的简化版本
class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    # 可能包含其他字段如 sender, ...

2. 图结构构建

• LLM 节点: 处理推理和决策
• 工具节点 (ToolNode): 执行工具调用
• 条件边: 根据 LLM 输出决定下一步

调用过程的详细分解

当你调用 agent.invoke() 时：

步骤 1: 状态初始化

result = agent.invoke({
    "messages": [
        {
            "role": "user", 
            "content": "What is 55+98*100?"
        }
    ],
})

这个字典被映射到 AgentState，创建初始状态。

步骤 2: 图执行流程

# 伪代码展示 ToolNode 的工作方式（基于源码）
class ToolNode:
    def __call__(self, state: AgentState):
        # 1. 从最新消息中提取工具调用
        last_message = state["messages"][-1]
        tool_calls = extract_tool_calls(last_message)
        
        # 2. 执行所有工具
        results = []
        for tool_call in tool_calls:
            tool = self.tools[tool_call["name"]]
            result = tool.invoke(tool_call["args"])
            results.append(result)
        
        # 3. 返回工具执行结果作为新消息
        return {"messages": [ToolMessage(content=result) for result in results]}

步骤 3: ReAct 循环模式

用户问题 → LLM思考 → 工具调用 → 工具结果 → LLM思考 → ... → 最终答案

实际执行示例

对于数学问题 “What is 55+98*100?”：

1. 初始状态: [HumanMessage("What is 55+98*100?")]
2. LLM 分析: 识别需要计算工具
3. 工具调用: 调用计算器执行 98*100 然后 55+9800
4. 返回结果: [HumanMessage(...), AIMessage(思考), ToolMessage(9855), AIMessage(最终答案)]

关键设计优势

1. 模块化: ToolNode 统一处理所有工具调用
2. 状态管理: add_messages 确保消息正确累积
3. 灵活性: 可以轻松添加新工具而不改变核心逻辑
4. 可观测性: 完整的消息历史便于调试和追踪

在工具中访问和修改状态

状态管理

LangGraph 的一个优秀特性就是状态管理。图结构拥有一个类型化数据结构，该结构在图执行期间对每个节点可用，并能持久化存储到长期存储中。你可以利用这个特性在节点间共享信息、调试图执行过程，或者将长时间运行的图重置到早期状态。

定义图状态时，需要指定数据类型和一个"归约器"函数。归约器描述了如何向该元素添加信息。当任务被分配到多个并行执行且同时更新状态的节点时，这个功能特别有用。

本示例使用了默认的 AgentState，其定义位于 langgraph.prebuilt.chat_agent_executor。

    class AgentState(TypedDict):
        """智能体的状态。"""
        messages: Annotated[Sequence[BaseMessage], add_messages]
        remaining_steps: NotRequired[RemainingSteps]

• messages 是一个 BaseMessage 列表（定义在 langchain_core），包含与LLM交互的输入输出消息
  • typing.Annotated 允许你为类型提示附加任意元数据。语法：Annotated[类型, 元数据1, 元数据2, …]
• add_messages 归约器会将新消息追加到消息列表末尾
• remaining_steps 用于跟踪图中的步骤数。你会看到它被初始化为 recursion_limit，但由图内部跟踪，对用户不可见

过程说明和分析：

原教程没有说明白
我问一下具体过程。
这是一个很好的问题，涉及到 LangGraph 中智能体的状态管理机制。让我详细解释一下：

AgentState 的作用

AgentState 定义了智能体运行时的状态容器：

class AgentState(TypedDict):
    """智能体的状态。"""
    messages: Annotated[Sequence[BaseMessage], add_messages]  # 消息序列
    remaining_steps: NotRequired[RemainingSteps]  # 可选：剩余步骤计数

• messages: 存储对话历史，使用 add_messages 处理器自动追加新消息
• remaining_steps: 可选字段，用于限制最大执行步骤

为什么能调用到智能体

1. 状态传递机制

当你调用 agent.invoke() 时，传入的字典会成为初始状态：

result = agent.invoke({
    "messages": [  # 这个会映射到 AgentState 的 messages 字段
        {
            "role": "user",
            "content": "What is 55+98*100?",
        }
    ],
})

2. 图执行流程

智能体实际上是一个 LangGraph，执行过程如下：

# 伪代码展示执行逻辑
def graph_execution(initial_state: AgentState):
    state = initial_state  # 从传入的字典创建状态
    
    while not should_stop(state):
        # 根据当前状态选择下一个节点（工具或LLM）
        next_node = decide_next_step(state)
        
        # 执行节点，更新状态
        new_messages = next_node.execute(state["messages"])
        state["messages"] = add_messages(state["messages"], new_messages)
    
    return state

3. 类型映射

LangGraph 会自动将你的输入字典映射到 AgentState 类型：

• {"messages": [...]} → AgentState(messages=...)
• 使用 add_messages 处理器确保消息正确追加而不是替换

完整示例

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, Sequence
from langchain_core.messages import BaseMessage, HumanMessage
from langgraph.checkpoint.sqlite import SqliteSaver

# 定义状态
class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], add_messages]

# 创建图
graph_builder = StateGraph(AgentState)

def llm_node(state: AgentState):
    # 处理逻辑...
    return {"messages": [new_message]}

graph_builder.add_node("llm", llm_node)
graph_builder.set_entry_point("llm")
graph = graph_builder.compile()

# 调用 - 输入自动映射到 AgentState
result = graph.invoke({
    "messages": [
        HumanMessage(content="What is 55+98*100?")
    ]
})

如下图这个特征。

关键点

1. 类型安全: AgentState 确保状态结构一致
2. 自动处理: add_messages 处理器简化了消息管理
3. 状态持久化: 整个执行过程的状态变化都被跟踪
4. 灵活性: 可以添加其他状态字段如 remaining_steps 来控制执行

这就是为什么简单的字典输入能够正确调用到智能体执行流程的原因。

对话内容

让我们快速了解一下这个机制。

from IPython.display import JSON
from langchain_core.messages import messages_to_dict

JSON({"messages": messages_to_dict(result1["messages"])})

{
  "messages": [
    {
      "type": "human",
      "data": {
        "content": "What is 3.1 * 4.2?",
        "additional_kwargs": {},
        "response_metadata": {},
        "type": "human",
        "name": null,
        "id": "d245d49d-01ec-4b43-aaa8-91386bccfddb",
        "example": false
      }
    },
    {
      "type": "ai",
      "data": {
        "content": "I'll calculate 3.1 multiplied by 4.2 for you.",
        "additional_kwargs": {
          "tool_calls": [
            {
              "id": "call_00_3sxPMydOIP3gRWRMVg3AQTge",
              "function": {
                "arguments": "{\"operation\": \"multiply\", \"a\": 3.1, \"b\": 4.2}",
                "name": "calculator"
              },
              "type": "function",
              "index": 0
            }
          ],
          "refusal": null
        },
        "response_metadata": {
          "token_usage": {
            "completion_tokens": 45,
            "prompt_tokens": 321,
            "total_tokens": 366,
            "completion_tokens_details": null,
            "prompt_tokens_details": {
              "audio_tokens": null,
              "cached_tokens": 0
            },
            "prompt_cache_hit_tokens": 0,
            "prompt_cache_miss_tokens": 321
          },
          "model_name": "deepseek-chat",
          "system_fingerprint": "fp_ffc7281d48_prod0820_fp8_kvcache",
          "id": "d5624aa6-99ef-42e9-bfc9-e03e1afb34a7",
          "service_tier": null,
          "finish_reason": "tool_calls",
          "logprobs": null
        },
        "type": "ai",
        "name": null,
        "id": "run--1e9ffbf2-07e8-4500-af84-82006f67fdb3-0",
        "example": false,
        "tool_calls": [
          {
            "name": "calculator",
            "args": {
              "operation": "multiply",
              "a": 3.1,
              "b": 4.2
            },
            "id": "call_00_3sxPMydOIP3gRWRMVg3AQTge",
            "type": "tool_call"
          }
        ],
        "invalid_tool_calls": [],
        "usage_metadata": {
          "input_tokens": 321,
          "output_tokens": 45,
          "total_tokens": 366,
          "input_token_details": {
            "cache_read": 0
          },
          "output_token_details": {}
        }
      }
    },
    {
      "type": "tool",
      "data": {
        "content": "13.020000000000001",
        "additional_kwargs": {},
        "response_metadata": {},
        "type": "tool",
        "name": "calculator",
        "id": "9b986c84-50f4-49b5-ab7a-ac27045af104",
        "tool_call_id": "call_00_3sxPMydOIP3gRWRMVg3AQTge",
        "artifact": null,
        "status": "success"
      }
    },
    {
      "type": "ai",
      "data": {
        "content": "3.1 × 4.2 = 13.02\n\nThe result is approximately 13.02 (the slight decimal extension is due to floating-point precision in the calculation).",
        "additional_kwargs": {
          "refusal": null
        },
        "response_metadata": {
          "token_usage": {
            "completion_tokens": 37,
            "prompt_tokens": 376,
            "total_tokens": 413,
            "completion_tokens_details": null,
            "prompt_tokens_details": {
              "audio_tokens": null,
              "cached_tokens": 320
            },
            "prompt_cache_hit_tokens": 320,
            "prompt_cache_miss_tokens": 56
          },
          "model_name": "deepseek-chat",
          "system_fingerprint": "fp_ffc7281d48_prod0820_fp8_kvcache",
          "id": "7acefa82-116d-4b93-965a-d7ec8ee7261a",
          "service_tier": null,
          "finish_reason": "stop",
          "logprobs": null
        },
        "type": "ai",
        "name": null,
        "id": "run--5a321057-378d-4a2c-8876-2b46531b8a88-0",
        "example": false,
        "tool_calls": [],
        "invalid_tool_calls": [],
        "usage_metadata": {
          "input_tokens": 376,
          "output_tokens": 37,
          "total_tokens": 413,
          "input_token_details": {
            "cache_read": 320
          },
          "output_token_details": {}
        }
      }
    }
  ]
}

教程文本

在工具中访问和修改状态

状态管理

LangGraph 的一项优秀特性是状态管理。图结构在运行期间拥有一个类型化的数据结构，该结构可供每个节点使用，并能持久化存储到长期存储器中。您可以利用这个特性在节点间共享信息、调试图执行过程，或是将长时间运行的图重置到更早的状态。

当您为图定义状态时，需要定义数据类型和一个"归约器"函数。归约器描述了如何向该状态元素添加信息。这在任务被并行执行的多个节点映射时尤其有用，因为这些节点会同时更新状态。

本示例中使用的是默认的 AgentState，其定义位于 langgraph.prebuilt.chat_agent_executor。

    class AgentState(TypedDict):
        """智能体的状态"""
        messages: Annotated[Sequence[BaseMessage], add_messages]
        remaining_steps: NotRequired[RemainingSteps]

• messages 是一个 BaseMessage 列表（定义于 langchain_core），包含与LLM交互的输入输出消息。
  • typing.Annotated 允许您为类型提示附加任意元数据。语法为：Annotated[类型, 元数据1, 元数据2, …]
• add_messages 归约器会将新消息追加到消息列表的末尾
• remaining_steps 用于追踪图中的执行步数。您会看到它被初始化为 recursion_limit，但该值由图内部管理，对用户不可见
  让我们快速了解一下这个机制。

自定义状态

让我们扩展计算器功能，使其能够记录所有执行过的运算操作。这需要在状态中添加一个列表，并创建一个归约器函数来将操作添加到列表中。这样可以安全处理列表为空或操作为空的情况。

from typing import Annotated, Sequence, TypedDict
from langchain_core.messages import BaseMessage
import operator

# 定义自定义状态
class CalculatorState(TypedDict):
    """计算器的状态"""
    messages: Annotated[Sequence[BaseMessage], add_messages]
    operations: Annotated[list, add_operation]  # 新增：记录所有运算操作的列表
    remaining_steps: int

# 定义操作归约器函数
def add_operation(current_operations: list, new_operation: dict) -> list:
    """将新操作添加到操作列表中"""
    if current_operations is None:
        current_operations = []
    
    if new_operation is not None:
        current_operations.append(new_operation)
    
    return current_operations

# 或者使用更简洁的lambda表达式
add_operation = lambda current, new: (current or []) + ([new] if new else [])

现在，当我们在工具中执行运算时，可以同时更新操作记录：

def calculate_operation(state: CalculatorState, operation: str, numbers: list):
    """执行计算并记录操作"""
    result = perform_calculation(operation, numbers)
    
    # 创建操作记录
    operation_record = {
        "operation": operation,
        "numbers": numbers,
        "result": result,
        "timestamp": datetime.now().isoformat()
    }
    
    # 更新状态
    state["operations"].append(operation_record)
    
    return result

这样，每次执行运算时都会自动记录操作详情，便于后续查询和分析运算历史。

两个工具定义的代码对比

原来的加法器

@tool
def calculator(
    operation: Literal["add","subtract","multiply","divide"],
    a: Union[int, float],
    b: Union[int, float],
) -> Union[int, float]:
    """Define a two-input calculator tool.

    Arg:
        operation (str): The operation to perform ('add', 'subtract', 'multiply', 'divide').
        a (float or int): The first number.
        b (float or int): The second number.
        
    Returns:
        result (float or int): the result of the operation
    Example
        Divide: result   = a / b
        Subtract: result = a - b
    """
    if operation == 'divide' and b == 0:
        return {"error": "Division by zero is not allowed."}

    # Perform calculation
    if operation == 'add':
        result = a + b
    elif operation == 'subtract':
        result = a - b
    elif operation == 'multiply':
        result = a * b
    elif operation == 'divide':
        result = a / b
    else: 
        result = "unknown operation"
    return result

带记录的加法器

from langgraph.prebuilt.chat_agent_executor import AgentState


def reduce_list(left: list | None, right: list | None) -> list:
    """Safely combine two lists, handling cases where either or both inputs might be None.

    Args:
        left (list | None): The first list to combine, or None.
        right (list | None): The second list to combine, or None.

    Returns:
        list: A new list containing all elements from both input lists.
               If an input is None, it's treated as an empty list.
    """
    if not left:
        left = []
    if not right:
        right = []
    return left + right

class CalcState(AgentState):
    """Graph State."""
    ops: Annotated[List[str], reduce_list]

@tool
def calculator_wstate(
    operation: Literal["add","subtract","multiply","divide"],
    a: Union[int, float],
    b: Union[int, float],
    state: Annotated[CalcState, InjectedState],   # not sent to LLM
    tool_call_id: Annotated[str, InjectedToolCallId] # not sent to LLM
) -> Union[int, float]:
    """Define a two-input calculator tool.

    Arg:
        operation (str): The operation to perform ('add', 'subtract', 'multiply', 'divide').
        a (float or int): The first number.
        b (float or int): The second number.
        
    Returns:
        result (float or int): the result of the operation
    Example
        Divide: result   = a / b
        Subtract: result = a - b
    """
    if operation == 'divide' and b == 0:
        return {"error": "Division by zero is not allowed."}

    # Perform calculation
    if operation == 'add':
        result = a + b
    elif operation == 'subtract':
        result = a - b
    elif operation == 'multiply':
        result = a * b
    elif operation == 'divide':
        result = a / b
    else: 
        result = "unknown operation"
    ops = [f"({operation}, {a}, {b})," ]
    return Command(
        update={
            "ops": ops,
            "messages": [
                ToolMessage(f"{result}", tool_call_id=tool_call_id)
            ],
        }
    )

结果对比是多了ops的记录段。

结果对比

实测对比，确实一个可以记录过程，一个无法记录，只能提供结果。

ai优化条理清晰版

太好了 👍 那我来帮你把这篇关于 LangGraph 深度智能体（ReAct Agent）代码解析与精华提炼 的内容排版成一篇可以直接在 CSDN 发表的技术博文。

这篇文章会：

• 保留你和 Claude 的代码逻辑要点
• 去掉重复、结构不清的地方
• 加上更适合技术博客的标题层级、目录、说明图示、总结和延伸方向
• 让整体风格像你在系统学习 LangGraph 时的学习日志 / 教程

---

🚀 LangGraph 深度智能体实战：从基础 ReAct Agent 到带状态管理的智能体（Claude 代码精华解析）

此处的示例代码，是上面的进行了一些缩减。

💡 本文通过一个计算器示例，从零讲解如何使用 LangGraph 创建一个支持工具调用和状态管理的 ReAct 智能体（Reason + Act），并详细解析 Claude 代码中的设计精华，帮助你快速掌握 LangGraph 的智能体机制。

---

🧭 一、背景与学习目标

LangGraph 是一种以「图结构驱动智能体行为」的框架。
本文目标是：

• 理解 ReAct 模型的核心思想：推理（Reason） + 行动（Act）循环
• 学会使用 LangGraph 的 create_react_agent 创建可调用外部工具的智能体
• 掌握状态管理机制（AgentState、InjectedState、Command）的使用方式

---

📦 二、项目环境配置

我们首先使用 .env 文件管理多模型 API Key：

OPENAI_API_KEY=sk-xxxxx
DEEPSEEK_API_KEY=sk-deepseekxxxxx
ANTHROPIC_API_KEY=sk-claudexxxxx

Python 加载：

from dotenv import load_dotenv
load_dotenv()

💬 LangGraph 会自动从环境变量中读取对应 Key。
当你在同一项目中混用多个模型（如 DeepSeek + OpenAI）时尤其方便。

---

🧠 三、ReAct Agent 概念速览

1️⃣ 什么是 ReAct 模式？

ReAct（Reason + Act） 是一种智能体架构，强调「推理」与「行动」的交替循环。

User Query
  ↓
[LLM Reasoning] → 判断是否调用工具
  ↓（是）
[Tool Executes] → 返回结果
  ↓
[LLM Reflects] → 综合回答或继续行动

LangGraph 的 create_react_agent 正是基于这个框架的高阶封装。

---

🧮 四、第一阶段：基础 ReAct Agent（无状态）

1. 定义工具

from langchain.tools import tool
from typing import Literal

@tool
def calculator(operation: Literal["add", "subtract", "multiply", "divide"], a: float, b: float) -> float:
    """简单四则运算工具"""
    if operation == "add": return a + b
    if operation == "subtract": return a - b
    if operation == "multiply": return a * b
    if operation == "divide": return a / b if b != 0 else float("inf")

2. 构建 Agent

from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent

model = ChatOpenAI(model="deepseek-chat", temperature=0)
agent = create_react_agent(model, [calculator], prompt="You are a helpful math assistant.")

3. 执行智能体任务

result = agent.invoke({
    "messages": [{"role": "user", "content": "What is 3.1 * 4.2?"}]
})

运行输出：

Tool call: calculator(operation="multiply", a=3.1, b=4.2)
Result: 13.02
Final Answer: The result is 13.02

✅ 特点：

• 能自动判断何时调用工具
• 每次调用相互独立，没有状态记忆

---

🔄 五、第二阶段：带状态管理的 ReAct Agent

1️⃣ 定义自定义状态

from langgraph.prebuilt.chat_agent_executor import AgentState
from typing_extensions import Annotated

def reduce_list(left: list | None, right: list | None) -> list:
    """安全地合并状态中的操作列表"""
    return (left or []) + (right or [])

class CalcState(AgentState):
    ops: Annotated[list[str], reduce_list]  # 记录操作历史

2️⃣ 定义带状态更新的工具

from langgraph.types import Command, ToolMessage
from langgraph.prebuilt import InjectedState, InjectedToolCallId

@tool
def calculator_wstate(
    operation: Literal["add", "subtract", "multiply", "divide"],
    a: float, b: float,
    state: Annotated[CalcState, InjectedState],
    tool_call_id: Annotated[str, InjectedToolCallId],
):
    """带状态记录的计算器"""
    result = eval(f"{a} {'+' if operation=='add' else '-' if operation=='subtract' else '*' if operation=='multiply' else '/'} {b}")

    return Command(
        update={
            "ops": [f"({operation}, {a}, {b})"],  # 操作记录
            "messages": [ToolMessage(content=f"Result: {result}", tool_call_id=tool_call_id)]
        }
    )

⚙️ InjectedState 与 InjectedToolCallId 是 LangGraph 的 隐式注入机制，不会发送给 LLM。

---

3️⃣ 构建带状态的 Agent

model = ChatOpenAI(model="deepseek-chat", temperature=0)
agent = create_react_agent(model, [calculator_wstate], system="You are a math assistant with memory.")

4️⃣ 执行过程与状态更新

result = agent.invoke({
    "messages": [{"role": "user", "content": "Calculate 3 + 5, then multiply by 2."}]
})

执行流程：

1. Agent 决定调用 calculator_wstate

2. 工具执行 add(3, 5) → 状态更新 ops = ["(add, 3, 5)"]

3. Agent 推理继续 → 调用 multiply(8, 2) → 状态更新
   ops = ["(add, 3, 5)", "(multiply, 8, 2)"]

4. 最终回答：

   The result is 16.
   

---

🧩 六、关键概念总结表

名称	含义	用途
@tool	工具装饰器	将 Python 函数暴露给 LLM 调用
AgentState	智能体状态基类	管理 messages 与自定义变量
InjectedState	隐式注入	工具可访问状态但不暴露给 LLM
InjectedToolCallId	调用标识符	用于生成响应消息
Command	状态更新命令	控制 LangGraph 状态流
reduce_list	合并函数	解决状态并发冲突

---

🧠 七、设计亮点与 Claude 代码精华提炼

1️⃣ Command 模式更新状态
通过返回 Command(update=...) 实现图状态的同步与可追溯。

2️⃣ Injected 参数机制
注入内部上下文（如状态、ID），让工具函数既能“感知外部”，又不影响 LLM 提示语。

3️⃣ ReAct 循环逻辑显性化
LangGraph 将 Reason/Act 循环抽象为有向图节点，每一步都可追踪与可视化。

4️⃣ 状态合并与函数式设计
reduce_list 的函数式写法体现 LangGraph 的「声明式状态管理哲学」。

---

📈 八、拓展方向

• 🤝 多工具协作：例如同时调用搜索与数据库
• 🧩 自定义记忆模块：让智能体具备上下文记忆
• 🔍 可视化执行流程：使用 .get_graph().draw_mermaid_png()
• ⚙️ 异步执行：结合 stream_agent 实现流式输出

---

🏁 九、总结

通过本文示例，我们从 基础 ReAct Agent 入门，逐步理解了：

1. 工具调用机制
2. 状态管理设计
3. Command 模式的可扩展性

🔮 LangGraph 的核心优势在于：图式思维 + 状态驱动的智能体设计。
当你理解这些机制后，可以轻松扩展到搜索、任务分解、数据库操作等复杂场景。

---

🪜 十、下一步学习建议

• 学习 LangGraph Studio 可视化执行图
• 尝试多模型混合（DeepSeek + GPT-4）
• 实现记忆型 Agent（保存对话历史）
• 阅读 LangGraph 官方教程中 AgentExecutor 源码

---