使用 LangGraph 构建 AI Agent：基本聊天机器人

打开Python，用几行代码构建一个能记住对话、按流程执行任务的智能体，效率提升10倍不止

“帮我写封邮件”、“这段代码怎么优化”——如果你对AI的使用还停留在这类简单问答上，那可能错过了LLM应用的真正潜力。现在，基于大型语言模型的智能体（AI Agent）已经能够自主规划、调用工具、处理复杂任务，而实现这一切的关键技术就是LangGraph。

今天，我们将从零开始，用Python构建两个LangGraph应用：一个能计算最大公约数的工作流，和一个具备记忆功能的智能聊天机器人。

01 走进LangGraph的世界：从LangChain到智能体

在深入代码之前，让我们先搞清楚LangGraph到底是什么。

LangChain是一个框架，它能让你轻松使用GPT、Claude等基础模型，并为它们装备工具，或让它们在简单的ReAct工作流中完成推理、规划和执行。

而LangGraph则让你对工作流程有完全的控制权。你可以创建自定义状态、节点、边、条件、中间件等等。如果ReAct智能体做的简单规划对你来说不够用，你想要构建一个复杂的、生产就绪的智能体，那就应该使用LangGraph。

LangGraph允许你强制让执行（或其部分）按照预定义的顺序发生，同时不放弃任何功能。

可以把LangChain看作是构建智能体的乐高积木，而LangGraph则是详细的建筑图纸和施工流程。

一个核心区别在于：LangChain适合简单的线性工作流，需要手动设置内存、持久化等；而LangGraph提供了更复杂的能力，支持状态管理、循环工作流和人工监督的开箱即用功能。

LangGraph的图形化架构使用有向图表示智能体工作流，支持循环和条件分支，能够维护所有组件间的共享状态，实现无缝通信。

LangChain 环境的最后一部分是 LangSmith——它是一个平台，允许你使用实时生产数据进行持续测试和改进，并在一个友好的用户界面中观察、评估和部署你的代理。但是，本课程不会使用 LangSmith，因为它并非学习基础知识的必要条件，而且包含它只会使文章篇幅过长。

本文是关于使用 LangGraph 和 LangChain 构建复杂智能体的——所以我们将平衡使用这两者。

首先，你需要一个 Python 解释器（版本 3.13 或更高版本）以及以下库：

bs4>=0.0.2 
ddgs>=9.6.1 
dotenv>=0.9.9 
langchain>=1.0.0 
langchain-community>=0.4 
langchain-huggingface>=1.0.0 
langchain-openai>=1.0.0 
langchain-text-splitters>=1.0.0 
langgraph>=1.0.0 
sentence-transformers>=5.1.1

如果你使用 pip，可以使用 pip install 命令安装它们pip install [library]。不过，我们建议你使用uv为你的项目创建一个独立的虚拟环境——在本例中，一个虚拟环境就足以满足整个课程的需求，因为上述依赖项列表适用于所有部分。安装 uv 后，你可以使用 pip install uv 在你的文件夹中创建虚拟环境uv venv，然后uv init使用 pip install uv 添加依赖项uv add [library]。你将使用 pip run 运行你的脚本uv run filename.py。

02 第一块积木：构建你的首个LangGraph工作流

让我们从一个简单的例子开始：计算两个数的最大公约数（GCD）。这个算法虽然简单，但能完美展示LangGraph的核心概念。

a = int( input ( "a = " )) 
b = int( input ( "b = " )) 
while b != 0 : 
  a, b = b, a % b 
  print( "GCD = " , a)

状态（State）是你的工作流记忆

在LangGraph中，状态就像程序的全局内存，所有节点都能访问和修改它。对于GCD计算，我们需要跟踪两个数字：

from typing import TypedDict

classState(TypedDict):
    a: int
    b: int

节点（Nodes）是执行具体任务的单元

节点就像函数，接收当前状态副本，返回状态的修改部分（不是完整的新状态）。

节点会获取当前状态的副本，并返回对其的修改。这与普通函数不同——你不会返回整个新状态，而只会返回修改后的状态。例如，如果你的状态是， State(a=3, b=2)并且你执行return {"a": 1}了操作，图中的状态会变为State(a=1, b=2)，但注意b本身并没有改变。因此，如果你完全不想改变状态，你可以执行操作return {}。

如何将执行过程划分为节点通常是任意的——整个最大公约数算法完全可以只用一个节点来表示。但为了演示如何构建图，我们将创建不同的节点：一个用于获取用户输入，一个用于计算 a 和 b 的下一个值，一个用于输出结果。

defget_user_data(_: State) -> State:
"""获取用户输入的数字"""
    a = int(input("a = "))
    b = int(input("b = "))
return {"a": a, "b": b}

defmodify(state: State) -> State:
"""执行欧几里得算法的一步"""
    a, b = state["a"], state["b"]
    a, b = b, a % b  # 关键计算步骤
return {"a": a, "b": b}

defwrite(state: State) -> State:
"""输出最终结果"""
    print("GCD = ", state["a"])
return {}  # 不修改状态

注意：每个节点都接收一个State 实例（当前状态的副本）作为输入，并返回State修改后的对象。你可以像访问字典一样访问 State 中的变量state[“a”]，state[“b”]也可以返回字典。你也可以直接返回 StateState(a=…,b=…)，但这通常不是常用的做法，因为 StateState 继承自 State``TypedDict是有原因的。此外，如果你在某个节点中不使用状态，可以用下划线替换它，但必须将其放在函数声明中。

边（Edges）连接节点，决定执行顺序

边定义了节点间的流向，包括条件边用于实现分支逻辑。

需要连接各个节点来创建一个图。首先，你需要初始化图并添加节点，同时为节点命名。

from langgraph.graph import StateGraph, START, END

# 创建图
graph = StateGraph(State)

# 添加节点
graph.add_node("get_user_data", get_user_data)
graph.add_node("modify", modify)
graph.add_node("write", write)

边定义了节点的执行顺序——从节点 A 到节点 B 的边表示节点 A 在节点 B 之前运行。此外，还有一些常量节点START，END需要添加到 LangGraph 中才能确定执行顺序。在我们的例子中，我们首先获取用户数据：

# 添加边：START → 获取数据 → 循环条件检查
graph.add_edge(START, “get_user_data”)

然而，在修改  a和b之前，我们需要检查循环条件。在 LangGraph 中如何实现呢？可以使用条件边。从给定节点出发，检查一个条件来确定接下来要执行哪个节点。但是，在检查循环迭代之后，我们不想再返回到获取用户数据的步骤。我们只需要执行一次——因此我们需要一个“虚拟节点”来标识循环的起点，以便返回。虽然并非总是必需的，但在本例中却非常重要：

defloop_condition( _: State ) -> State:
return {} 


graph.add_node("loop_condition" , loop_condition) 

graph.add_edge("get_user_data" , loop_condition)

现在，我们可以创建一个带有条件边的循环：

# 条件边：根据b是否为0决定下一步
graph.add_conditional_edges(
"get_user_data",
lambda state: state["b"] != 0,  # 条件函数
    {
True: "modify",   # b≠0时继续计算
False: "write",   # b=0时输出结果
    },
)

# 循环边：计算后返回检查条件
graph.add_edge("modify", "get_user_data")
graph.add_edge("write", END)

首先，你指定它从哪个节点出发——在本例中是loop_condition。接着，你传递一个函数，该函数接收当前状态作为输入并返回某个值——这里函数非常简单，因此我们直接用lambda实现。最后但同样重要的是，你传递一个字典，将函数结果映射到下一个目标节点的名称。在此处，若 b != 0，条件返回 True 则转至修改节点；否则转至写入节点输出结果。从写入节点开始，工作流最终在 END 节点结束。

编译并运行你的工作流

# 编译图
workflow = graph.compile()

# 可视化图结构（可选）
with open("gcd_graph.png", "wb") as f:
    f.write(workflow.get_graph().draw_mermaid_png())

# 运行工作流
workflow.invoke({})

现在，我们的图表如下所示：

要运行工作流，只需使用一些初始状态调用它即可。在我们的例子中，我们实际上不需要在开始时传递任何内容，因为我们稍后会a从b用户那里获取数据，所以我们只需使用一个空字典来调用它：

workflow.invoke({})

我们的程序已经基本完成，你可以立即运行它，但你可能会注意到它在处理高复杂度数据时无法正常工作a——b这是因为 LangGraph 限制了循环的递归深度。你需要在调用程序时通过配置参数来增加递归深度限制：

workflow.invoke({}, {“recursion_limit”: 100 })

尝试运行这个程序，输入a = 64和b = 240，你会得到GCD = 16。

这个简单的例子展示了LangGraph的核心：状态管理、节点函数和条件边。接下来，让我们加入大语言模型，构建真正的智能体。

03 智能升级：打造有记忆的聊天机器人

现在，让我们用LangGraph和OpenAI的GPT-4o构建一个智能聊天机器人。这个机器人不仅能回答问题，还能记住整个对话历史。

第一步：环境设置

from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, AIMessage
from langchain.agents import AgentState
from dotenv import load_dotenv
import os

# 加载环境变量（确保你的.env文件中有OPENAI_API_KEY）
load_dotenv()

第二步：定义智能状态

这次我们使用AgentState，它有一个特殊属性：messages字段会自动追加新消息，而不是覆盖：

classState(AgentState):
    iteration: int  # 跟踪对话轮次

第三步：创建核心对话节点

这个节点负责与LLM交互并维护对话历史：

# 设置迭代限制和初始化模型
ITERATION_LIMIT = 5
model = init_chat_model("openai:gpt-4o")

defask_llm(state: State) -> State:
"""与LLM交互的核心节点"""
# 获取用户输入
    user_query = input("query: ")
    user_message = HumanMessage(user_query)

# 调用模型，传入完整的对话历史
    answer_message: AIMessage = model.invoke(state["messages"] + [user_message])
    print("answer: ", answer_message.content)

# 更新状态：追加新消息，增加迭代计数
return {
"messages": [user_message, answer_message],  # 自动追加到历史
"iteration": state["iteration"] + 1,
    }

第四步：构建智能对话图

# 创建状态图
graph = StateGraph(State)

# 添加对话节点
graph.add_node("ask_llm", ask_llm)

# 设置初始边和条件边
graph.add_edge(START, "ask_llm")
graph.add_conditional_edges(
"ask_llm",
lambda state: state["iteration"] < ITERATION_LIMIT,  # 检查是否达到轮次限制
    {
True: "ask_llm",  # 继续对话
False: END,       # 结束对话
    },
)

# 编译工作流
workflow = graph.compile()

第五步：运行你的智能聊天机器人

# 初始化状态并运行
initial_state = {"iteration": 0, "messages": []}
workflow.invoke(initial_state, {"recursion_limit": 100})

现在，你可以和这个机器人进行真正的多轮对话了：

query: 你好，我是张三
answer: 你好张三！很高兴认识你。有什么我可以帮助你的吗？
query: 你还记得我的名字吗？
answer: 当然记得，你刚刚告诉我你叫张三。今天有什么需要我协助的呢？
query: 2+2等于多少？
answer: 2+2等于4。还有其他数学问题吗？
...

如果没有对话历史记录——每次查询都是独立的。此外，循环是无限的，这不是一个好的做法。现在，我们将添加对话记忆功能并限制循环迭代次数。

消息历史记录

由于我们希望消息列表能在多个节点执行过程中被重复使用，因此将其添加到状态中。为此，你需要继承AgentState类而非TypedDict类：

from langchain.agents import AgentState 

classState(AgentState ):
pass

AgentState实施过程从以下方式开始：

classAgentState(TypedDict, Generic [ResponseT]):
"""代理的状态模式。"""

     messages: Required[Annotated[list[AnyMessage], add_messages]

使用 add_messages 会改变更新的处理方式。当你返回{"messages": [HumanMessage(…), HumanMessage(…)]}时，返回的消息会被追加到 messages 中，而不是覆盖它。现在节点应呈现如下形式：

defask_llm( state: State ) -> State:
    user_query = input ( "query: " ) 
    user_message = HumanMessage(user_query) 
    answer_message: AIMessage = model.invoke(state[ "messages" ] + [user_message]) 
print ( "answer: " , answer_message.content) 

return { 
"messages" : [user_message, answer_message], 
    }

现在，你将整个消息历史记录传递给模型，以便它记住对话内容。节点会使用新的用户消息和模型中的回复更新状态，并将更新后的状态追加到现有状态messages（而不是覆盖现有状态）。

迭代次数限制

由于我们想要限制迭代次数，因此需要跟踪已经执行了多少次迭代。

classState( AgentState ):
    iteration: int  # 添加此行

由于只有一个节点，我们将递增iteration该节点的值。我们需要通过返回更新来改变节点状态：

defask_llm(state: State) -> State:
    user_query = input("query: ")
    user_message = HumanMessage(user_query)
    answer_message: AIMessage = model.invoke(state["messages"] + [user_message])
    print("answer: ", answer_message.content)

return {
"messages": [user_message, answer_message],
    }

最后一步是更新图，添加一条条件边来检查迭代次数。如果超过迭代次数，则转到下一个步骤END。

ITERATION_LIMIT = 5

 graph = StateGraph(State) 

graph.add_node( "ask_llm" , ask_llm) 

graph.add_edge(START, "ask_llm" ) 
graph.add_conditional_edges( 
"ask_llm" , 
lambda state: state[ "iteration" ] < ITERATION_LIMIT, 
    { 
True : "ask_llm" , 
False : END 
    } 
)

这次调用时，我们需要将其设置iteration为 0。

关键进步：与简单调用ChatGPT API不同，这个机器人通过state["messages"]维护了完整的对话历史，实现了真正的上下文感知。同时，通过iteration计数，我们避免了无限循环，这在生产环境中至关重要。

04 理解LangGraph的核心优势

通过这两个例子，你应该已经感受到了LangGraph的强大。让我们总结一下它的关键优势：

循环处理能力：传统的LangChain链是线性的，而LangGraph支持复杂的循环和条件分支，这正好符合人类对话的自然流程——多轮次、有条件的交互。

内置状态管理：AgentState和状态归约器（reducer）自动处理状态更新。在我们的聊天机器人中，messages字段使用add_messages归约器，确保新消息被追加而不是覆盖。

可视化与可调试性：workflow.get_graph().draw_mermaid_png()可以生成工作流的可视化图表，这让复杂的智能体逻辑一目了然，极大简化了调试过程。

灵活的节点设计：每个节点可以专注于单一职责。在实际的复杂智能体中，你可以有专门的节点处理工具调用、条件判断、结果验证等不同任务。

生产就绪特性：LangGraph支持持久化检查点（checkpoint），这意味着即使程序中断，智能体也能从上次停止的地方恢复执行，这对长时间运行的任务至关重要。

05 从简单到复杂：LangGraph的进阶应用

LangGraph的真正威力在复杂智能体系统中才完全展现。在实际应用中，你可以构建：

多智能体协作系统：不同特长的智能体组成团队，协作完成复杂任务。例如，一个研究助手系统可能包含网络搜索智能体、文档分析智能体和报告撰写智能体，它们通过LangGraph协调工作。

工具调用智能体：智能体可以调用外部工具，如搜索引擎、数据库查询、API调用等。通过条件边检查tool_calls属性，智能体可以自主决定何时调用工具。

人机协作流程：在关键决策点引入人工审核，这在金融、医疗等高风险领域特别有用。LangGraph的interrupt机制允许工作流暂停，等待人工输入。

自适应RAG系统：根据查询复杂度动态调整检索策略。简单查询直接回答，复杂查询先检索资料再生成答案。

企业级审批流程：将传统工作流自动化，每个审批节点对应一个LangGraph节点，实现智能路由和状态跟踪。

写在最后

LangGraph将LLM应用开发从简单的问答模式提升到了真正的智能体系统层面。通过状态图、节点和边的概念，开发者可以构建出能够自主规划、使用工具、处理复杂工作流的智能应用。

我们的最大公约数计算器和智能聊天机器人只是LangGraph的入门示例。在实际项目中，你可以将这些概念扩展到更复杂的场景：添加工具调用让智能体能搜索网页或查询数据库；实现多智能体协作让多个专业智能体共同解决问题；设计条件工作流处理不同的用户意图。

LangGraph的学习曲线可能比直接调用API要陡峭，但它的回报是巨大的：更可控、更可靠、更强大的LLM应用。

你已经迈出了构建智能体的第一步。接下来，可以探索工具调用、多智能体系统等高级主题，将你的聊天机器人升级为真正的AI助手。

你在实际项目中尝试过LangGraph或类似的智能体框架吗？遇到了哪些挑战？有什么有趣的应用场景想要分享吗？欢迎在评论区留言，一起探讨智能体开发的无限可能！