大模型的开发应用（十六）：Agent 与 LangGraph基础

1 什么是 Agent

Agent 在早几年被翻译成代理，但这与它的实际功能并不贴切，因此这两年称呼其为 “智能体” 比较多，它是指能自主感知环境、做出决策并执行动作以达成目标的实体，通过利用大语言模型的推理能力，能够自主对复杂的人类任务进行目标规划、任务拆解、工具调用、过程迭代，并在没有人类干预的情况下完成任务。

Agent 的核心在于其自主性：无需人类全程操控，能主动解决问题。

1.1 Agent 的核心特征

1. 自主性（Autonomy）

   • 能独立运行，无需人类直接干预。
   • 例如：自动驾驶汽车根据路况自动调整路线。

2. 感知能力（Perception）

   • 通过传感器或数据输入感知环境（如摄像头、文本输入、API 数据等）。

3. 决策与推理（Decision-making）

   • 根据目标分析信息，制定行动策略（可能涉及任务拆分、规划、学习、逻辑推理等）。

4. 执行能力（Action）

   • 通过物理动作（如机械臂）或数字操作（如调用 API、发送消息）影响环境。

5. 目标导向（Goal-oriented）

   • 行为围绕特定目标展开（如“解答用户问题”或“优化能源消耗”）。

6. 适应性（Adaptivity）

   • 学习经验或适应环境变化（如推荐系统根据用户反馈调整推送策略）。

1.2 AI Agent 的常见类型

1. 简单反射型（Reflex Agents）

   • 基于预设规则直接响应（如：温度传感器触发空调开关）。

2. 基于模型的（Model-based Agents）

   • 通过内部模型追踪环境状态变化（如：预测交通的导航系统）。

3. 目标导向型（Goal-based Agents）

   • 通过规划路径达成目标（如：物流机器人规划最优配送路径）。

4. 基于效用的（Utility-based Agents）

   • 在多个目标间权衡，选择最优解（如：投资系统平衡风险与收益）。

5. 学习型（Learning Agents）

   • 通过数据反馈改进行为（如：AlphaGo 通过自我对弈提升棋力）。

我们这个系列的文章介绍的是大模型，因此本文接下来介绍的 Agent 都是基于大语言模型的。

1.3 现代 AI Agent 的关键技术

• 大语言模型（LLM）驱动：
  通过自然语言理解任务并生成推理步骤（如：ChatGPT 充当客服助手）。
• 工具调用（Tool Use）：
  结合外部工具扩展能力（如：调用计算器、数据库、搜索API）。
• 记忆机制（Memory）：
  存储短期交互记录或长期知识库（如：记录用户偏好实现个性化回复）。
• 多智能体系统（Multi-Agent Systems）：
  多个 Agent 协作（如：供应链管理中库存、物流、销售 Agent 协同决策）。

1.4 Agent vs. 传统程序

特性	Agent（智能体）	传统程序
自主性	✅ 主动决策与执行	❌ 需人工触发
环境交互	✅ 实时感知与动态响应	❌ 固定输入输出
目标复杂度	✅ 处理开放性问题（如客户咨询）	❌ 处理预设任务（如数据排序）
适应性	✅ 可学习新数据优化行为	❌ 需人工更新逻辑

---

1.5 实际应用场景

• 个人助理
  （如：GPT-4 结合日历/邮件 API 安排会议）。
• 工业自动化
  （如：工厂中监控设备状态的 Agent 预测故障）。
• 游戏 NPC
  （如：《我的世界》中由 LLM 驱动的角色生成个性化剧情）。
• 科研探索
  （如：AutoGPT 自动检索文献、设计实验并撰写报告）。

1.6 Agent 的核心价值

将 AI 从“被动工具”变为“主动协作者”：用户只需提出目标（如：“制定减肥计划”），Agent 会自主分解任务、调用工具（查食谱/运动库）、生成个性化方案并持续跟踪进度。

随着多模态模型（图像/语音）和具身智能（机器人）的发展，Agent 正在成为人与数字世界交互的核心入口。

2 LangGraph 简介

LangGraph 是一个由 LangChain 团队开发的 Python 框架，专门用于构建复杂、有状态的多步骤工作流（尤其是面向智能体（Agent）的应用）。它通过图（Graph）结构来建模任务流程，让开发者能够灵活设计包含循环、分支和状态管理的 AI 系统。

2.1 核心概念

1. 图（Graph）结构

   • 用节点（Nodes） 表示任务步骤（例如：调用 LLM、执行代码、检索知识库）。
   • 用边（Edges） 定义步骤间的流转逻辑（例如：根据上一步的结果决定下一步操作）。

2. 状态（State）

   • 在整个工作流中共享的动态数据容器（例如：用户问题、中间结果、历史记录）。
   • 每个节点接收状态、修改状态，并传递给下一个节点。

3. 循环（Cycles）

   • 我们人去调大模型，如果大模型的回复没达到要求，那么我们换一种方式重新让模型生成，这样一直循环下去，直到模型生成的内容符合要求，或者超过调用次数，这就是循环性。在 Agent 中，是 Agent 的大脑去调用大模型（这里假设被调用的大模型为模型A），然后大脑会去评估模型的回复，若不符合要求则重新调用，这里大脑也是一个大模型（假设为模型B），一般情况下，模型B的智商远高于模型A，模型A被当成了工具。
   • 支持工作流循环执行（例如：反复思考直到答案满足条件）。
   • 这是其区别于传统链（Chain）的核心能力！

2.2 关键能力

✅ 构建复杂 Agent 系统

• 实现多工具调用、自我反思、动态规划（如：ReAct, Plan-and-Execute 等 Agent 架构）。
  ✅ 处理长时间对话
• 通过状态管理维护会话历史和上下文。
  ✅ 协调多个子任务
• 例如：先检索知识 → 分析 → 验证 → 生成最终结果。
  ✅ 错误恢复与重试
• 定义异常处理分支（如：当 API 调用失败时切换备用方案）。

2.3 与 LangChain 的关系

• 定位：LangGraph 是 LangChain 生态中的工作流协调层（类似 Airflow 但专为 AI 设计）。
• 互补性：
  • LangChain 提供基础组件（LLM 调用、工具封装等）。
  • LangGraph 用图结构将这些组件组合成复杂逻辑。

2.4 典型应用场景

1. 多轮问答 Agent
   • 循环执行：问题分解 → 搜索 → 整合 → 验证，直到答案达标。
2. 自动化决策系统
   • 根据实时数据动态选择执行路径（如：金融风控场景）。
3. 游戏 NPC 逻辑引擎
   • 控制角色行为树，结合 LLM 生成动态剧情。
4. 复杂任务编排
   • 例如：自动写代码 → 测试 → 调试 → 生成报告。

2.5 优势总结

特点	说明
灵活循环控制	支持 while 式逻辑（传统 Chain 难以实现）
状态共享	跨节点传递结构化数据
可视化调试	支持工作流执行过程可视化
与 LangChain 无缝集成	复用现有组件（Tools, LLMs 等）

⚡️ 一句话总结：LangGraph 是 AI 工作流的“操作系统”，专为需要状态追踪 + 循环决策的复杂 Agent 设计。适合构建超越单次问答的下一代 AI 应用！

2.6 安装与入门教程

安装就一条命令：

pip install -U langgraph

LangGraph 的官方入门教程链接（这套教程对中国用户不友好，因为需要花钱买各种工具和模型的API Key，有些工具和模型，不对中国用户开放）。

3 LangGraph 的基本使用

3.1 图结构组件

LangGraph 的核心是将代理工作流建模为图，你可以使用三个关键组件来定义 Agent 的行为：

3.1.1 状态

状态是所有节点共享的数据结构，就像一个"数字笔记本"，记录和跟踪 AI 系统处理的所有重要信息。它可以是任何 Python 类型，但通常是 TypedDict 或 Pydantic BaseModel。

下面是一个非常典型的状态类定义

from typing import TypedDict, List
from langchain_core.messages import BaseMessageclass GraphState(TypedDict):messages: List[BaseMessage]		# 用于记录每个节点做了什么user_info: dict					# 存储当前用户的个人信息和偏好设置tools_used: List[str]			# 记录在当前会话中已经使用过的工具列表step_count: int					# 跟踪当前对话或任务执行的步骤数

3.1.2 节点

节点一般是编码 Agent 逻辑的 Python 函数，函数的第一个参数是状态，第二个参数（可选）一般是 “配置”，包含可选的可配置参数。节点接收当前状态作为输入，执行一些计算，并返回一个更新的状态。可以使用 StateGraph 对象（StateGraph类稍后会介绍）的 add_node 方法将这些节点添加到图形中。

START 与 END：START 节点是一个特殊节点，它代表将用户输入发送到图形的节点，引用此节点的主要目的是确定哪些节点应该首先被调用；END 节点也是一个特殊节点，它代表一个终端节点，当你想要指定哪些边在完成操作后没有动作时，可以引用此节点。这两个节点已经创建好了，不需要我们自己编写函数。

END节点可以没有，但START节点必须有。

3.1.3 边

边定义了逻辑如何路由以及图形如何决定停止。有一些关键类型的边Python 函数，根据当前状态确定要执行的下一个节点，它们可以是条件分支（条件边）或固定转换（普通边）。可以使用 StateGraph 对象的 add_edge 或 add_conditional_edges 添加边。

一个节点可以有多个输出边。如果一个节点有多个输出边，则所有这些目标节点将在下一个超级步骤中并行执行。

3.2 StateGraph类

StateGraph 类是使用的主要图类，关于这个类，有三点需要掌握：

1. 它在创建的时候，需要把状态类给传进去。

2. 有了 StateGraph 类对象，才能添加节点和边。

3. 图构建好之后，需要进行编译，编译图的目的是对图的结构进行一些基本检查（没有孤立的节点等等），如果通过编译，说明没问题。

3.3 简单实现

我们做一个简单的示例：

from typing import TypedDict, List
from langgraph.graph import START, StateGraph 
from langchain_core.messages import BaseMessage# 定义状态
class GraphState(TypedDict):messages: List[BaseMessage]		# 用于记录每个节点做了什么# 创建 StateGraph 对象，使用 GraphState 类型作为状态类型 
workflow = StateGraph(dict) # 定义一个节点函数 my_node，接收状态和配置，返回新的状态 
def my_node(state, config): return {"x": state["x"] + 1,"y": state["y"] + 2} # 向构建器中添加节点 my_node，节点名称将自动设置为'my_node'，运行到这个节点的时候将调用 my_node 函数 
workflow.add_node(my_node) # 添加一条边，从 START 到'my_node'节点 
workflow.add_edge(START, "my_node") # 编译状态图，生成可执行的图 
graph = workflow.compile() # 调用编译后的图，传入初始状态{"x": 1} 
print(graph.invoke({"x": 1,"y":2}))

输出：

{'x': 2, 'y': 4}

上面的步骤可以归纳为：

1. 定义状态；
2. 创建 StateGraph 对象；
3. 往 StateGraph 对象添加节点和边；
4. 编译图

使用 LangGraph 构建 Agent 时，无论多复杂的 Agent，都是上面的步骤。

4 使用 LangGraph 构建聊天机器人

我们现在使用 LangGraph 来创建一个基本聊天机器人，这个聊天机器人将直接回复用户的消息，这个简单的案例用于加深 LangGraph 构建 Agent 的步骤。

4.1 模型选型：商用模型 VS 本地模型

要构建 Agent，那么大模型必须足够强，这里我们对比一下商用模型（ GLM-4-Plus）和本地模型（Qwen2.5-4B-Instruct）调用工具的能力。

先跑一下智谱的模型：

from langgraph.prebuilt import create_react_agent
from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage# 配置智谱模型
import os
os.environ["ZHIPUAI_API_KEY"] = "XXXXX"		# 智谱的API Key
from langchain_community.chat_models import ChatZhipuAIzhipu_model = ChatZhipuAI(model="glm-4-plus",temperature=0.5,
)# 也可以是下面这样调用
# from langchain_openai import ChatOpenAI
# zhipu_model = ChatOpenAI(
#    openai_api_base="https://open.bigmodel.cn/api/paas/v4",
#    model_name="glm-4-plus",
#    openai_api_key="XXXXX",
#    max_tokens=400
# )# 查询天气的模拟工具
def get_weather(city: str) -> str:"""Get weather for a given city."""return f"It's sunny in {city}!"# 构建 Agent
agent = create_react_agent(model=zhipu_model,tools=[get_weather],		# 把模拟查询天气的函数作为工具传进去prompt="You are a helpful assistant"
)# Run the agent
result = agent.invoke({"messages": [{"role": "user", "content": "what is the weather in Beijing"}]})# 打印对话过程（遍历消息列表）
for message in result['messages']:if isinstance(message, HumanMessage):role = "用户"if isinstance(message, AIMessage):role = "AI"if isinstance(message, SystemMessage):role = "System"if isinstance(message, ToolMessage):role = "Tool"if len(message.content) == 0 and role == 'AI':print(f"{role:8}: tool_call ...")else:print(f"{role:8}: {message.content}")

输出：

用户      : what is the weather in Beijing
AI      : tool_call ...
Tool    : It's sunny in Beijing!
AI      : It's sunny in Beijing! If you need more detailed weather information, such as the current temperature, humidity, or forecasts for the next few days, feel free to ask!

我们在创建智能体的时候，把模拟查询天气的函数作为工具传进去了，从打印的结果来看，在用户询问天气的时候，智能体（AI）先是调用工具，然后工具进行查询，然后智能体把查询得到的结果返回给用户，然后又附上了一段贴心的文字：If you need more detailed weather information, such as the current temperature, humidity, or forecasts for the next few days, feel free to ask!

接下来我们试一下本地模型，在调用本地模型之前，先用 LMDeploy 把本地的 Qwen2.5-4B-Instruct 跑起来：

lmdeploy serve api_server /data/coding/models/Qwen/Qwen1.5-4B-Chat

现在我们用本地模型完成上述任务：

from langgraph.prebuilt import create_react_agent
from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage# 配置模型
from langchain_openai import ChatOpenAIlocal_model = ChatOpenAI(openai_api_base="http://0.0.0.0:23333/v1",model_name="/data/coding/models/Qwen/Qwen1.5-4B-Chat",openai_api_key="empty",max_tokens=400
)# 查询天气的模拟工具
def get_weather(city: str) -> str:"""Get weather for a given city."""return f"It's sunny in {city}!"# 构建 Agent
agent = create_react_agent(model=local_model,tools=[get_weather],		# 把模拟查询天气的函数作为工具传进去prompt="You are a helpful assistant"
)# Run the agent
result = agent.invoke({"messages": [{"role": "user", "content": "what is the weather in Beijing"}]})# 打印对话过程（遍历消息列表）
for message in result['messages']:if isinstance(message, HumanMessage):role = "用户"if isinstance(message, AIMessage):role = "AI"if isinstance(message, SystemMessage):role = "System"if isinstance(message, ToolMessage):role = "Tool"if len(message.content) == 0 and role == 'AI':print(f"{role:8}: tool_call ...")else:print(f"{role:8}: {message.content}")

输出：

用户      : what is the weather in Beijing
AI      : I'm sorry, as an AI language model, I don't have access to real-time weather information. However, you can check the weather in Beijing by searching "Beijing weather" on a search engine or by using a weather app on your device.

可以看到，Qwen2.5-4B-Instruct 不会调用工具，说明这个模型智商不够。

当然，这里进行对比的目的，是介绍如何使用 LangChain 配置本地模型和商用模型，直接用 langchain_openai.ChatOpenAI 就行。

模型确定后，就可以加载进来了：

from langchain_openai import ChatOpenAI
llm = ChatOpenAI(openai_api_base="https://open.bigmodel.cn/api/paas/v4",model_name="glm-4-plus",openai_api_key="XXXXX",max_tokens=400
)

4.2 定义状态

首先定义状态：

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph.message import add_messagesclass State(TypedDict):# Messages have the type "list". The `add_messages` function# in the annotation defines how this state key should be updated# (in this case, it appends messages to the list, rather than overwriting them)messages: Annotated[list, add_messages]

定义图时，第一步是定义其状态，状态包括图的模式和处理状态更新的 reducer 函数（这个例子中是add_messages）。在我们的示例中，状态是一个具有一个键：messages 的 TypedDict。 add_messages 函数用于将新消息追加到列表中，而不是覆盖它。没有 reducer 注解的键将覆盖先前的值。

4.3 创建 StateGraph 对象

StateGraph 对象将我们的聊天机器人结构定义为“状态机”：

from langgraph.graph import StateGraph, STARTgraph_builder = StateGraph(State)

我们将添加节点来表示 LLM 和聊天机器人可以调用的函数，并添加边来指定机器人应如何在这些函数之间进行转换。

我们的图现在可以处理两个关键任务：

1. 每个节点都可以接收当前 状态 作为输入，并输出状态的更新。
2. 对消息的更新将追加到现有列表而不是覆盖它，这得益于与 Annotated 语法一起使用的预构建 add_messages 函数

4.4 添加一个节点

接下来，添加一个“chatbot”节点，节点表示工作单元，通常是普通的 Python 函数。我们让这个节点来回复用户，可以将聊天模型集成到一个简单的节点中：

def chatbot(state: State):return {"messages": [llm.invoke(state["messages"])]}# The first argument is the unique node name
# The second argument is the function or object that will be called whenever
# the node is used.
graph_builder.add_node("chatbot", chatbot)

注意 chatbot 节点函数将当前状态作为输入，并返回一个包含更新的消息列表的字典，键为“messages”。这是所有 LangGraph 节点函数的基本模式。状态中的 add_messages 函数会将 LLM 的响应消息追加到状态中已有的消息之后。

4.5 添加入口（添加边）

添加一个 入口 点，以告诉图每次运行时从何处开始工作

graph_builder.add_edge(START, "chatbot")

4.6 编译图

在运行图之前，我们需要对其进行编译。我们可以通过在图构建器上调用 compile() 来完成。这将创建一个 CompiledGraph，我们可以在我们的状态上调用它。

graph = graph_builder.compile()

4.7 运行聊天机器人

现在运行聊天机器人：

def stream_graph_updates(user_input: str):for event in graph.stream({"messages": [{"role": "user", "content": user_input}]}):for value in event.values():print("Assistant:", value["messages"][-1].content)while True:try:user_input = input("User: ")if user_input.lower() in ["quit", "exit", "q"]:print("Goodbye!")breakstream_graph_updates(user_input)except:# fallback if input() is not availableuser_input = "What do you know about LangGraph?"print("User: " + user_input)stream_graph_updates(user_input)break

在Notebook中，这段代码存在一些问题，导致直接走异常处理流程，可以运行稍后展示的完整的代码来测试结果。

4.8 完整代码

from typing import Annotated
from typing_extensions import TypedDictfrom langgraph.graph import StateGraph, START
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAIclass State(TypedDict):# Messages have the type "list". The `add_messages` function# in the annotation defines how this state key should be updated# (in this case, it appends messages to the list, rather than overwriting them)messages: Annotated[list, add_messages]# 创建一个 StateGraph
graph_builder = StateGraph(State)# 配置模型
llm = ChatOpenAI(openai_api_base="https://open.bigmodel.cn/api/paas/v4",model_name="glm-4-plus",openai_api_key="XXXXX",max_tokens=400
)# 将模型集成到节点中
def chatbot(state: State):return {"messages": [llm.invoke(state["messages"])]}# 添加节点
graph_builder.add_node("chatbot", chatbot)
# The first argument is the unique node name
# The second argument is the function or object that will be called whenever
# the node is used.# 添加一个入口点
graph_builder.add_edge(START, "chatbot")# 编译图
graph = graph_builder.compile()# 从流式输出中打印信息
def stream_graph_updates(user_input: str):for event in graph.stream({"messages": [{"role": "user", "content": user_input}]}):for value in event.values():print("Assistant:", value["messages"][-1].content)# 对话
while True:try:user_input = input("User: ")if user_input.lower() in ["quit", "exit", "q"]:print("Goodbye!")breakstream_graph_updates(user_input)except:# fallback if input() is not availableuser_input = "What do you know about LangGraph?"print("User: " + user_input)stream_graph_updates(user_input)break

输出：

User: 你知道LangGraph吗？
Assistant: 截至我的知识更新日期（2023年4月），"LangGraph" 不是一个广为人知的术语或特定的技术产品。它可能是一个特定领域或某个项目中的专有名词，或者是某种新技术的名称，但在我的知识库中没有具体的信息。如果 "LangGraph" 是指某种与语言处理、图形表示或相关技术结合的新工具或概念，那么它可能是在我知识更新之后出现的。为了获取最新和最准确的信息，建议直接查阅相关的技术文档、学术论文或联系相关领域的专家。如果你能提供更多的上下文或详细信息，我或许能够提供更具体的帮助或猜测其可能的含义。
User: