从零搭建第一个 LangGraph Agent：工具调用与状态管理

为什么是 LangGraph

2025 年 Agent 框架已经很多了：LangChain、CrewAI、AutoGen、Dify……但如果你要做生产级 Agent 系统，LangGraph 是目前最值得投入的选择之一。三个原因：

显式状态机：不是黑盒，你精确控制每一步的流转
可控性：条件边、循环、人工介入点，全在代码里看得见
可观测性：每一步的输入输出都可追踪，运维友好

这篇文章用一个真实的 AI 运维助手 demo，把 LangGraph 的四个核心概念讲清楚。

核心概念速览

概念	一句话解释	类比
State	共享数据，贯穿所有节点	CI 流水线中的环境变量
Node	处理步骤（LLM 推理 / 工具执行）	CI 中的 stage
Edge	节点间流转规则	CI 中的 `needs` / `when`
Tool	Agent 可调用的外部函数	Makefile 中的 target

这四个概念组合起来，就是一个有限状态机 + LLM + 工具调用。下图是 demo 的流转逻辑：

用户消息 → LLM 推理 → 需要工具？
                ├── 是 → 执行工具 → 结果送回 LLM → 再次判断
                └── 否 → 输出最终回复

Step 1：定义工具

工具就是普通的 Python 函数，加上 @tool 装饰器和清晰的 docstring。LLM 会根据 docstring 判断什么时候该调用哪个工具。

from langchain_core.tools import tool
import json
from datetime import datetime

@tool
def get_current_time() -> str:
    """获取当前系统时间。当用户询问「现在几点」「当前时间」时使用。"""
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

@tool
def check_disk_usage(path: str = "/") -> str:
    """检查指定路径的磁盘使用情况。当用户询问「磁盘空间」时使用。

    Args:
        path: 要检查的路径，默认为根目录 /
    """
    # 生产环境用 shutil.disk_usage()，demo 用模拟数据
    mock_data = {
        "/": {"total": "100G", "used": "72G", "free": "28G", "percent": "72%"},
        "/data": {"total": "500G", "used": "320G", "free": "180G", "percent": "64%"},
    }
    return json.dumps(mock_data.get(path, mock_data["/"]), ensure_ascii=False)

@tool
def read_recent_logs(service: str = "nginx", lines: int = 10) -> str:
    """读取指定服务的最近日志行。当用户询问「日志」「报错」时使用。

    Args:
        service: 服务名称，如 nginx、api、db
        lines: 读取的行数，默认 10 行
    """
    mock_logs = {
        "nginx": [
            "[INFO] 192.168.1.10 - GET /api/health 200 0.023s",
            "[WARN] upstream response time 3.2s exceeds threshold",
            "[ERROR] connect() failed to upstream: connection refused",
        ],
        "api": [
            "[INFO] DB connection pool: 15/20 active",
            "[WARN] Request rate approaching limit: 950/1000 rps",
            "[ERROR] Unhandled exception in /api/orders: TimeoutError",
        ],
    }
    entries = mock_logs.get(service, [f"[INFO] No logs for service: {service}"])
    return "\n".join(entries[:lines])

TOOLS = [get_current_time, check_disk_usage, read_recent_logs]

关键点：docstring 写得好，LLM 就选得准。这比调 prompt 更重要——工具描述是 Agent 决策的重要依据。

Step 2：定义状态

State 是图中所有节点共享的数据结构。operator.add 表示消息列表是追加式的——每个节点返回的 messages 会追加到已有列表末尾，而不是覆盖。

from typing import Annotated
import operator
from langchain_core.messages import BaseMessage
from langgraph.graph import StateGraph

class AgentState(dict):
    messages: Annotated[list[BaseMessage], operator.add]

你可以把 State 理解为 CI 流水线中的环境变量——每个阶段都能读写，最终汇聚成一个完整的执行记录。

Step 3：定义节点

LLM 推理节点

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
llm_with_tools = llm.bind_tools(TOOLS)

def llm_node(state: AgentState) -> dict:
    """LLM 推理：决定回复用户还是调用工具"""
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}

bind_tools() 是关键——它把工具定义注入 LLM，让 LLM 在需要时自动生成 tool_calls。

工具执行节点

LangGraph 内置了 ToolNode，不需要手写：

from langgraph.prebuilt import ToolNode

tool_node = ToolNode(TOOLS)

它会自动读取上一条 AIMessage 中的 tool_calls，逐个执行，返回 ToolMessage。

Step 4：构建图

这是最核心的部分——把节点和边组装成一个可执行的状态机。

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver

# 创建图
graph = StateGraph(AgentState)

# 添加节点
graph.add_node("llm", llm_node)
graph.add_node("tools", tool_node)

# 设置入口
graph.set_entry_point("llm")

# 条件边：LLM 输出有 tool_calls → 去 tools，否则 → END
def should_continue(state: AgentState):
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"
    return END

graph.add_conditional_edges("llm", should_continue, {"tools": "tools", END: END})

# 固定边：工具执行完 → 回到 LLM 继续推理
graph.add_edge("tools", "llm")

# 编译
agent = graph.compile(checkpointer=MemorySaver())

MemorySaver 是 LangGraph 的短期记忆机制——同一个 thread_id 下的对话历史会被保留，支持多轮对话。

Step 5：运行

from langchain_core.messages import HumanMessage, ToolMessage, AIMessage

def run(question: str, thread_id: str = "demo-1"):
    config = {"configurable": {"thread_id": thread_id}}
    for event in agent.stream(
        {"messages": [HumanMessage(content=question)]},
        config=config,
        stream_mode="values",
    ):
        msg = event["messages"][-1]
        if isinstance(msg, ToolMessage):
            print(f"🔧 [工具] {msg.name} 返回: {msg.content[:100]}")
        elif isinstance(msg, AIMessage) and msg.content:
            print(f"🤖 助手: {msg.content}")

# 单工具调用
run("现在几点了？")

# 多工具协作
run("检查磁盘使用情况，然后看看 nginx 日志有没有异常")

# 纯对话
run("你好，介绍一下你自己")

实际运行输出：

👤 用户: 现在几点了？
🧠 [LLM] 推理中...
   → 决定调用工具: ['get_current_time']
🔧 [工具] get_current_time 返回: 2025-08-21 14:30:00
🧠 [LLM] 推理中...
   → 直接回复: 现在是 2025年8月21日 14:30:00...
🤖 助手: 现在是 2025年8月21日 14:30:00。

👤 用户: 检查磁盘使用情况，然后看看 nginx 日志有没有异常
🧠 [LLM] 推理中...
   → 决定调用工具: ['check_disk_usage', 'read_recent_logs']
🔧 [工具] check_disk_usage 返回: {"total": "100G", "used": "72G"...}
🔧 [工具] read_recent_logs 返回: [ERROR] connect() failed to upstream...
🧠 [LLM] 推理中...
   → 直接回复: 磁盘使用率 72%，nginx 日志中发现一个错误...
🤖 助手: 磁盘使用率 72%，还有 28G 可用。nginx 日志中发现一个关键错误：
  [ERROR] connect() failed to upstream: connection refused
  建议检查 upstream 服务是否正常运行。

注意第二个场景：Agent 自动并行调用了两个工具，然后把结果汇总分析——这就是 tool calling 的价值。

关键设计决策

1. 为什么要用条件边而不是固定边

固定边 tools → llm 是固定的，因为工具执行完必须把结果送回 LLM 分析。但 llm → tools 用条件边，因为 LLM 可能直接回复（不需要工具），也可能需要调用工具——这一步必须动态判断。

2. 为什么要用 MemorySaver

没有 MemorySaver，每轮对话都是独立的，Agent 不记得上一轮说了什么。MemorySaver 让同一个 thread_id 下的对话历史得以保留，这是多轮对话的基础。

3. 工具函数要返回字符串

LLM 只能理解文本。工具返回 JSON 字符串没问题，但别返回 Python 对象——LLM 看不懂。

完整代码

完整可运行代码在 demo/langgraph-agent/agent.py，包含三个工具和一个完整可运行的 demo。

git clone https://github.com/jackzhuo2019/astro-blog
cd astro-blog/demo/langgraph-agent
pip install -r requirements.txt
cp .env.example .env  # 填入你的 API Key
python agent.py

下一步

这篇文章覆盖了 LangGraph Agent 的基础骨架。下一篇文章会深入 AI 系统的可观测性——给 LLM 调用加上 Token 用量、延迟、错误率的监控，让 Agent 从”能跑”变成”能运维”。