LangGraph 1.x · Agent评估

背景

LangGraph 是 LangChain 团队推出的一个用于构建有状态、多参与者应用程序的框架，它通过图结构来编排和协调 AI 代理的工作流程。2025 年，LangGraph 1.0 的正式发布标志着它成为了构建 AI Agent 的事实标准。

为什么选择 LangGraph？

• 图结构编排：LangGraph 使用图（Graph）来定义 Agent 的工作流，使得复杂的多步骤任务变得清晰可控
• 状态管理：内置的状态管理机制，让 Agent 能够在执行过程中保持上下文信息
• 灵活的控制流：支持条件分支、循环、并行执行等多种控制流模式
• 工具调用集成：与 LangChain 生态无缝集成，轻松实现函数调用和工具使用
• 生产就绪：提供了完善的监控、调试和部署工具

本文目的

本文将通过构建一个简单的计算器 Agent 来验证 LangGraph 1.0 的核心功能，包括：

1. 如何定义状态和节点
2. 如何实现工具调用
3. 如何构建条件控制流
4. 如何在实际场景中使用 LangGraph 构建 Agent

验证

安装依赖

uv init
uv add "langchain>=1.1.3" "langchain-openai>=1.0.0" "langgraph>=1.0.4" "langgraph-cli[inmem]>=0.4.9" python-dotenv
uv sync

main.py

import os
import operator
from typing import Literal
from dotenv import load_dotenv

from langchain.tools import tool
from langchain.chat_models import init_chat_model
from langchain.messages import AnyMessage, SystemMessage, ToolMessage, HumanMessage
from typing_extensions import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END

# 加载 .env 文件中的环境变量
load_dotenv()

# 配置自建模型（必须从 .env 文件中读取）
BASE_URL = os.getenv("OPENAI_BASE_URL")
API_KEY = os.getenv("OPENAI_API_KEY")
MODEL_NAME = os.getenv("OPENAI_MODEL")

# 检查必需的配置项是否存在
if not BASE_URL:
    raise ValueError("OPENAI_BASE_URL 未在 .env 文件中配置")
if not API_KEY:
    raise ValueError("OPENAI_API_KEY 未在 .env 文件中配置")
if not MODEL_NAME:
    raise ValueError("OPENAI_MODEL 未在 .env 文件中配置")

model = init_chat_model(
    model=MODEL_NAME,  # 自建模型直接使用模型名称，不需要 openai: 前缀
    model_provider="openai",
    temperature=0.7,
    base_url=BASE_URL,
    api_key=API_KEY,
)


# 定义工具
@tool
def multiply(a: int, b: int) -> int:
    """Multiply `a` and `b`.

    Args:
        a: First int
        b: Second int
    """
    return a * b


@tool
def add(a: int, b: int) -> int:
    """Adds `a` and `b`.

    Args:
        a: First int
        b: Second int
    """
    return a + b


@tool
def divide(a: int, b: int) -> float:
    """Divide `a` and `b`.

    Args:
        a: First int
        b: Second int
    """
    return a / b


# 为LLM添加工具
tools = [add, multiply, divide]
tools_by_name = {tool.name: tool for tool in tools}
model_with_tools = model.bind_tools(tools)


# 步骤2：定义状态
class MessagesState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    llm_calls: int


# 步骤3：定义模型节点
def llm_call(state: dict):
    """LLM decides whether to call a tool or not"""
    return {
        "messages": [
            model_with_tools.invoke(
                [
                    SystemMessage(
                        content="You are a helpful assistant tasked with performing arithmetic on a set of inputs."
                    )
                ]
                + state["messages"]
            )
        ],
        "llm_calls": state.get('llm_calls', 0) + 1
    }


# 步骤4：定义工具节点
def tool_node(state: dict):
    """Performs the tool call"""
    result = []
    for tool_call in state["messages"][-1].tool_calls:
        tool = tools_by_name[tool_call["name"]]
        observation = tool.invoke(tool_call["args"])
        result.append(ToolMessage(content=str(observation), tool_call_id=tool_call["id"]))
    return {"messages": result}


# 步骤5：定义结束逻辑
def should_continue(state: MessagesState) -> Literal["tool_node", END]:
    """Decide if we should continue the loop or stop based upon whether the LLM made a tool call"""
    messages = state["messages"]
    last_message = messages[-1]

    # If the LLM makes a tool call, then perform an action
    if last_message.tool_calls:
        return "tool_node"

    # Otherwise, we stop (reply to the user)
    return END


# 步骤6：构建并编译代理
# 构建工作流
agent_builder = StateGraph(MessagesState)

# 添加节点
agent_builder.add_node("llm_call", llm_call)
agent_builder.add_node("tool_node", tool_node)

# 添加边以连接节点
agent_builder.add_edge(START, "llm_call")
agent_builder.add_conditional_edges(
    "llm_call",
    should_continue,
    ["tool_node", END]
)
agent_builder.add_edge("tool_node", "llm_call")

# 编译代理
agent = agent_builder.compile()


# 示例用法
if __name__ == "__main__":
    print("=== LangGraph Calculator Agent Demo ===\n")
    
    # 测试用例1：加法
    print("Test 1: Add 3 and 4")
    messages = [HumanMessage(content="Add 3 and 4.")]
    result = agent.invoke({"messages": messages})
    print(f"LLM calls: {result.get('llm_calls', 0)}")
    for m in result["messages"]:
        print(f"- {m.__class__.__name__}: {m.content}")
    print("\n")
    
    # 测试用例2：乘法
    print("Test 2: Multiply 5 and 6")
    messages = [HumanMessage(content="Multiply 5 and 6.")]
    result = agent.invoke({"messages": messages})
    print(f"LLM calls: {result.get('llm_calls', 0)}")
    for m in result["messages"]:
        print(f"- {m.__class__.__name__}: {m.content}")
    print("\n")
    
    # 测试用例3：除法
    print("Test 3: Divide 20 by 4")
    messages = [HumanMessage(content="Divide 20 by 4.")]
    result = agent.invoke({"messages": messages})
    print(f"LLM calls: {result.get('llm_calls', 0)}")
    for m in result["messages"]:
        print(f"- {m.__class__.__name__}: {m.content}")
    print("\n")
    
    # 测试用例4：复杂计算
    print("Test 4: Add 10 and 5, then multiply the result by 3")
    messages = [HumanMessage(content="Add 10 and 5, then multiply the result by 3.")]
    result = agent.invoke({"messages": messages})
    print(f"LLM calls: {result.get('llm_calls', 0)}")
    for m in result["messages"]:
        print(f"- {m.__class__.__name__}: {m.content}")

langgraph.json

{
  "dependencies": [
    "."
  ],
  "graphs": {
    "chat": "main:agent"
  },
  "env": ".env",
  "http": {
    "cors": {
      "allow_origins": ["*"],
      "allow_methods": ["*"],
      "allow_headers": ["*"],
      "allow_credentials": false
    }
  }
}

.env

LANGSMITH_API_KEY=XXX

# OpenAI 配置
OPENAI_BASE_URL=XXX
OPENAI_API_KEY=XXX
OPENAI_MODEL=XXX

重要步骤详解

1. 环境配置与模型初始化（第 12-34 行）

load_dotenv()  # 加载 .env 文件中的环境变量
BASE_URL = os.getenv("OPENAI_BASE_URL")
API_KEY = os.getenv("OPENAI_API_KEY")
MODEL_NAME = os.getenv("OPENAI_MODEL")

• 从 .env 读取配置（API 地址、密钥、模型名）
• 检查必需配置是否存在
• 使用 init_chat_model 初始化聊天模型，并绑定到自定义 API 端点

---

2. 工具定义（第 37-68 行）

@tool
def multiply(a: int, b: int) -> int:
    """Multiply `a` and `b`."""
    return a * b

• 使用 @tool 装饰器定义三个工具：add、multiply、divide
• 工具描述用于让 LLM 理解何时调用
• 第 72-74 行：将工具绑定到模型，使 LLM 可调用

---

3. 状态定义（第 77-81 行）

class MessagesState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    llm_calls: int

• messages：消息列表，使用 operator.add 实现追加
• llm_calls：记录 LLM 调用次数
• 状态在节点间传递，保存对话历史

---

4. LLM 调用节点（第 84-98 行）

def llm_call(state: dict):
    return {
        "messages": [
            model_with_tools.invoke([
                SystemMessage(content="...")] + state["messages"]
            )
        ],
        "llm_calls": state.get('llm_calls', 0) + 1
    }

• 接收当前状态，调用模型
• 添加系统消息（角色设定）
• 返回模型响应（可能包含工具调用）
• 更新 llm_calls 计数

---

5. 工具执行节点（第 102-109 行）

def tool_node(state: dict):
    result = []
    for tool_call in state["messages"][-1].tool_calls:
        tool = tools_by_name[tool_call["name"]]
        observation = tool.invoke(tool_call["args"])
        result.append(ToolMessage(...))
    return {"messages": result}

• 从最后一条消息提取工具调用
• 按名称查找工具并执行
• 将结果封装为 ToolMessage 返回
• 结果会加入消息历史，供后续 LLM 使用

---

6. 条件路由逻辑（第 113-123 行）

def should_continue(state: MessagesState) -> Literal["tool_node", END]:
    last_message = messages[-1]
    if last_message.tool_calls:
        return "tool_node"  # 有工具调用，去执行工具
    return END  # 没有工具调用，结束流程

• 判断最后一条消息是否包含工具调用
• 有则路由到 tool_node，无则结束

---

7. 图构建（第 128-144 行）

agent_builder = StateGraph(MessagesState)
agent_builder.add_node("llm_call", llm_call)
agent_builder.add_node("tool_node", tool_node)
agent_builder.add_edge(START, "llm_call")
agent_builder.add_conditional_edges("llm_call", should_continue, ["tool_node", END])
agent_builder.add_edge("tool_node", "llm_call")
agent = agent_builder.compile()

• 创建状态图，添加两个节点
• 边连接：
  • START → llm_call：入口
  • llm_call → tool_node/END：条件路由
  • tool_node → llm_call：工具执行后返回 LLM

工作流循环：

START → llm_call → [有工具调用?] → tool_node → llm_call → [有工具调用?] → ... → END

---

8. 测试用例（第 148-185 行）

• 测试用例 1-3：简单计算（加法、乘法、除法）
• 测试用例 4：多步计算（先加后乘）

每个测试：

1. 创建用户消息
2. 调用 agent.invoke() 执行
3. 打印 LLM 调用次数和所有消息

---

核心设计理念

1. 状态管理：通过 MessagesState 维护对话历史
2. 工具调用：LLM 决定何时调用工具
3. 循环执行：工具执行后返回 LLM，可继续调用工具
4. 条件路由：根据 LLM 输出决定下一步

---

执行流程示例

假设用户输入：”Add 10 and 5, then multiply the result by 3”

1. START → llm_call：LLM 分析请求，决定调用 add(10, 5)
2. llm_call → tool_node：执行 add(10, 5)，返回 15
3. tool_node → llm_call：LLM 看到结果 15，决定调用 multiply(15, 3)
4. llm_call → tool_node：执行 multiply(15, 3)，返回 45
5. tool_node → llm_call：LLM 看到结果 45，生成最终回复
6. llm_call → END：流程结束

这是一个典型的 ReAct（Reasoning + Acting）模式实现，LLM 可以推理并执行多步操作。

如需进一步解释某个部分，请告诉我。

load_dotenv()  # 加载 .env 文件中的环境变量
BASE_URL = os.getenv("OPENAI_BASE_URL")
API_KEY = os.getenv("OPENAI_API_KEY")
MODEL_NAME = os.getenv("OPENAI_MODEL")

@tool
def multiply(a: int, b: int) -> int:
    """Multiply `a` and `b`."""
    return a * b

class MessagesState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    llm_calls: int

def llm_call(state: dict):
    return {
        "messages": [
            model_with_tools.invoke([
                SystemMessage(content="...")] + state["messages"]
            )
        ],
        "llm_calls": state.get('llm_calls', 0) + 1
    }

def tool_node(state: dict):
    result = []
    for tool_call in state["messages"][-1].tool_calls:
        tool = tools_by_name[tool_call["name"]]
        observation = tool.invoke(tool_call["args"])
        result.append(ToolMessage(...))
    return {"messages": result}

def should_continue(state: MessagesState) -> Literal["tool_node", END]:
    last_message = messages[-1]
    if last_message.tool_calls:
        return "tool_node"  # 有工具调用，去执行工具
    return END  # 没有工具调用，结束流程

agent_builder = StateGraph(MessagesState)
agent_builder.add_node("llm_call", llm_call)
agent_builder.add_node("tool_node", tool_node)
agent_builder.add_edge(START, "llm_call")
agent_builder.add_conditional_edges("llm_call", should_continue, ["tool_node", END])
agent_builder.add_edge("tool_node", "llm_call")
agent = agent_builder.compile()

START → llm_call → [有工具调用?] → tool_node → llm_call → [有工具调用?] → ... → END

在 WebUI 上测试

uv run langgraph dev