LangGraph 的基础概念与实战应用。LangGraph 是基于 LangChain 的状态机库，利用有向无环图协调多个 LLM 或状态。文章详细讲解了环境安装、工具定义、Agent 状态设计（TypedDict）、节点函数编写、图结构构建（条件边与普通边）以及编译运行流程。通过智普 AI 模型实现了天气查询等工具调用的完整示例，并补充了执行流程解析、最佳实践及注意事项，帮助开发者快速掌握基于 LangGraph 构建复杂 Agent 的方法。
LangGraph 是 LangChain 框架基础上的一个库，它是 LangChain Expression Language (LCEL) 的扩展。LangGraph 能够利用有向无环图（DAG）的方式，去协调多个大语言模型（LLM）或状态信息。相比于 LCEL，LangGraph 的使用逻辑更为复杂，但能够处理循环、记忆管理等更复杂的流程，使得业务逻辑更加清晰。
LangGraph 相当于一种高级的状态机语言，非常适合构建需要多步决策、工具调用和状态持久化的 Agent 应用。本文将详细介绍如何安装 LangGraph，并基于智普 AI 的大模型实现一个可以调用工具的 Agent。
LangGraph 是一个独立的库，默认情况下不会随 LangChain 一起安装。在使用前，需要先通过 pip 进行安装。
pip install langgraph
此外，还需要确保安装了 LangChain 及其相关依赖。由于 OpenAI 接口在国内访问可能受限，本教程统一使用智普 AI（Zhipu AI）的大模型进行实践。智普 AI 的接口与 OpenAI 高度兼容，可以直接复用大部分 OpenAI 的 Tool 接口代码。
在 LangGraph 中，工具（Tools）是 Agent 执行具体任务的手段。我们可以使用 @tool 装饰器来定义工具。工具函数需要描述其功能、参数 schema 以及返回值。
以下示例定义了三个工具：
tagging: 分析句子的情感极性和语言。overview: 提供文本摘要、语言和关键词。get_current_temperature: 根据经纬度获取当前天气温度。import datetime
import operator
from typing import TypedDict, Annotated, Union, Optional, Type, List
import requests
from langchain.agents import create_openai_tools_agent
from langchain.pydantic_v1 import BaseModel, Field
from langchain.tools import BaseTool, tool
from langchain_core.agents import AgentAction, AgentFinish
from langchain_core.messages import BaseMessage
from langgraph.graph import END, StateGraph
from langgraph.prebuilt.tool_executor import ToolExecutor
from zhipu_llm import ChatZhipuAI
# 配置 API Key 和模型
zhipuai_api_key = "YOUR_API_KEY_HERE"
glm3 = "glm-3-turbo"
glm4 = "glm-4"
chat_zhipu = ChatZhipuAI(
temperature=0.8,
api_key=zhipuai_api_key,
model=glm3
)
# 定义 Tagging 工具的输入模式
class Tagging(BaseModel):
sentiment: str = Field(description="sentiment of text, should be `pos`, `neg`, or `neutral`")
language: str = Field(description="language of text (should be ISO 639-1 code)")
# 定义 Overview 工具的输入模式
class Overview(BaseModel):
summary: str = Field(description="Provide a concise summary of the content.")
language: str = Field(description="Provide the language that the content is written in.")
keywords: str = Field(description="Provide keywords related to the content.")
# 定义 tagging 工具
@tool("tagging", args_schema=Tagging)
def tagging(s1: str, s2: str):
"""分析句子的情感极性，并输出句子对应的语言"""
return "The sentiment is {a}, the language is {b}".format(a=s1, b=s2)
# 定义 overview 工具
@tool("overview", args_schema=Overview)
def overview(summary: str, language: str, keywords: str):
"""Overview of a section of text."""
return "Summary: {a}\nLanguage: {b}\nKeywords: {c}".format(a=summary, b=language, c=keywords)
# 定义 get_current_temperature 工具
@tool
def get_current_temperature(latitude: float, longitude: float):
"""Fetch current temperature for given coordinates."""
BASE_URL = "https://api.open-meteo.com/v1/forecast"
params = {
'latitude': latitude,
'longitude': longitude,
'hourly': 'temperature_2m',
'forecast_days': 1,
}
response = requests.get(BASE_URL, params=params)
if response.status_code == 200:
results = response.json()
else:
raise Exception(f"API Request failed with status code: {response.status_code}")
current_utc_time = datetime.datetime.utcnow()
time_list = [datetime.datetime.fromisoformat(time_str.replace('Z', '+00:00')) for time_str in results['hourly']['time']]
temperature_list = results['hourly']['temperature_2m']
closest_time_index = min(range(len(time_list)), key=lambda i: abs(time_list[i] - current_utc_time))
current_temperature = temperature_list[closest_time_index]
return f'The current temperature is {current_temperature}°C'
tools = [tagging, overview, get_current_temperature]
LangGraph 中最基础的类型是 StatefulGraph，这种图会在每一个 Node（节点）之间传递不同的状态信息。每个节点会根据自己定义的逻辑更新这个状态。
具体来说，我们需要继承 TypedDict 类来定义状态结构。下图展示了我们定义的四个关键变量：
input: 输入字符串，代表用户的主要请求。chat_history: 之前的对话信息列表，作为上下文传入。agent_outcome: 来自代理的响应，可以是 AgentAction（需要调用工具）或 AgentFinish（任务完成）。如果是 AgentFinish，AgentExecutor 结束；否则调用请求的工具。intermediate_steps: 代理在一段时间内采取的行动和相应观察结果的列表。每次迭代都会更新。这里特别需要注意的是 intermediate_steps 使用了 operator.add 注解。这表示对该状态的修改操作是追加到现有值上，而不是覆盖它，这对于维护对话历史至关重要。
class AgentState(TypedDict):
# The input string
input: str
# The list of previous messages in the conversation
chat_history: list[BaseMessage]
# The outcome of a given call to the agent
# Needs `None` as a valid type, since this is what this will start as
agent_outcome: Union[AgentAction, AgentFinish, None]
# List of actions and corresponding observations
# Here we annotate this with `operator.add` to indicate that operations to
# this state should be ADDED to the existing values (not overwrite it)
intermediate_steps: Annotated[list[tuple[AgentAction, str]], operator.add]
在 LangGraph 中，节点通常是一个函数或者 LangChain 中的 Runnable 对象。
我们这里定义两个核心节点：
此外，还需要定义节点之间的连接，也就是边。包括条件判断的边（决定走向）和普通边（固定流向）。
# This a helper class we have that is useful for running tools
# It takes in an agent action and calls that tool and returns the result
tool_executor = ToolExecutor(tools)
# Define the agent node function
def run_agent(data):
agent_outcome = agent_runnable.invoke(data)
return {"agent_outcome": agent_outcome}
# Define the function to execute tools
def execute_tools(data):
# Get the most recent agent_outcome - this is the key added in the `agent` above
agent_action = data["agent_outcome"]
print("agent action:{}".format(agent_action))
output = tool_executor.invoke(agent_action[-1])
return {"intermediate_steps": [(agent_action[-1], str(output))]}
# Define logic that will be used to determine which conditional edge to go down
def should_continue(data):
# If the agent outcome is an AgentFinish, then we return `exit` string
# This will be used when setting up the graph to define the flow
if isinstance(data["agent_outcome"], AgentFinish):
return "end"
# Otherwise, an AgentAction is returned
# Here we return `continue` string
# This will be used when setting up the graph to define the flow
else:
return "continue"
定义完节点和逻辑后，就可以构建整个图了。值得注意的是，条件判断的边和普通边的添加方式是不一样的。
StateGraph 实例。should_continue 函数的返回值决定下一步。# Get the prompt to use - you can modify this!
prompt = hub.pull("hwchase17/openai-tools-agent")
# Construct the OpenAI Functions agent
agent_runnable = create_openai_tools_agent(chat_zhipu, tools, prompt)
# Define a new graph
workflow = StateGraph(AgentState)
# Define the two nodes we will cycle between
workflow.add_node("agent", run_agent)
workflow.add_node("action", execute_tools)
# Set the entrypoint as `agent`
# This means that this node is the first one called
workflow.set_entry_point("agent")
# We now add a conditional edge
workflow.add_conditional_edges(
# First, we define the start node. We use `agent`.
"agent",
# Next, we pass in the function that will determine which node is called next.
should_continue,
# Finally we pass in a mapping.
# The keys are strings, and the values are other nodes.
# END is a special node marking that the graph should finish.
{
# If `tools`, then we call the tool node.
"continue": "action",
# Otherwise we finish.
"end": END,
},
)
# We now add a normal edge from `tools` to `agent`.
# This means that after `tools` is called, `agent` node is called next.
workflow.add_edge("action", "agent")
# Finally, we compile it!
# This compiles it into a LangChain Runnable,
# meaning you can use it as you would any other runnable
app = workflow.compile()
最后，我们需要准备输入数据并运行应用。输入应包含用户的初始请求和空的对话历史。
inputs = {"input": "what is the weather in NewYork", "chat_history": []}
result = app.invoke(inputs)
print(result["agent_outcome"].messages[0].content)
执行流程解析：
agent 节点。run_agent 函数被调用，LLM 分析输入 "what is the weather in NewYork"。AgentAction 指向 get_current_temperature 工具。should_continue 检测到不是 AgentFinish，返回 "continue"。action 节点，execute_tools 被调用，实际请求天气 API。intermediate_steps 状态。agent 节点。run_agent 再次被调用，LLM 结合历史步骤生成最终回答。AgentFinish，should_continue 返回 "end"。在实际开发 LangGraph 应用时，建议遵循以下最佳实践：
execute_tools 等节点中加入 try-except 块，防止因工具调用失败导致整个图崩溃。可以将错误信息记录到 intermediate_steps 中供 LLM 参考。app.get_graph().draw_mermaid_png() 来生成流程图，帮助理解复杂的节点流转关系。StateGraph 和 Runnable 接口的变动。本文详细介绍了 LangGraph 的核心概念及实战应用。通过定义状态、节点和边，我们成功构建了一个具备工具调用能力的 Agent。LangGraph 的强大之处在于其灵活的状态管理和循环处理能力，这使得它能够胜任比简单链式调用更复杂的任务场景。掌握 LangGraph 将为构建企业级 AI 应用打下坚实基础。
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
使用加密算法（如AES、TripleDES、Rabbit或RC4）加密和解密文本明文。 在线工具，加密/解密文本在线工具，online
生成新的随机RSA私钥和公钥pem证书。 在线工具，RSA密钥对生成器在线工具，online
基于 Mermaid.js 实时预览流程图、时序图等图表，支持源码编辑与即时渲染。 在线工具，Mermaid 预览与可视化编辑在线工具，online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online