LangGraph实战精通：从零开始构建智能体应用的状态机编排指南

# 使用uv安装LangGraph和相关依赖
uv pip install -U langgraph langchain python-dotenv typing-extensions

# .env文件内容
DEEPSEEK_API_KEY=your_deepseek_api_key_here

from typing import Annotated

from langchain.chat_models import init_chat_model
from typing_extensions import TypedDict

from langgraph.graph import StateGraph, START
from langgraph.graph.message import add_messages

import os
from dotenv import load_dotenv

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

class State(TypedDict):
    messages: Annotated[list, add_messages]


graph_builder = StateGraph(State)


llm = init_chat_model(
    "deepseek-chat",  # 使用DeepSeek模型
    api_key=os.environ.get("DEEPSEEK_API_KEY")
)


def chatbot(state: State):
    return {"messages": [llm.invoke(state["messages"])]}


# 添加节点和边
graph_builder.add_node("chatbot", chatbot)
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!")
            break
        stream_graph_updates(user_input)
    except KeyboardInterrupt:
        print("\nGoodbye!")
        break

from typing import Annotated
from langchain.chat_models import init_chat_model
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START
from langgraph.graph.message import add_messages
import os
from dotenv import load_dotenv

load_dotenv()

class State(TypedDict):
    messages: Annotated[list, add_messages]

graph_builder = StateGraph(State)

llm = init_chat_model(
    "deepseek-chat",  # 使用DeepSeek模型
    api_key=os.environ.get("DEEPSEEK_API_KEY")
)

def chatbot(state: State):
    return {"messages": [llm.invoke(state["messages"])]}

graph_builder.add_node("chatbot", chatbot)
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!")
            break
        stream_graph_updates(user_input)
    except KeyboardInterrupt:
        print("\nGoodbye!")
        break

uv run 1-build-basic-chatbot.py

User: who are you?
Assistant: I’m DeepSeek Chat, your AI assistant created by **DeepSeek**! 🤖✨ I’m here to help answer your questions, provide insights, and assist with anything from general knowledge to coding, writing, and more.

Think of me as a friendly, knowledgeable companion—always ready to chat, research, or brainstorm with you! 💡

What can I help you with today? 😊

User: exit
Goodbye!

print(graph.get_graph().draw_mermaid())

---
config:
  flowchart:
    curve: linear
---
graph TD;
	__start__([<p>__start__</p>]):::first
	chatbot(chatbot)
	__end__([<p>__end__</p>]):::last
	__start__ --> chatbot;
	chatbot --> __end__;
	classDef default fill:#f2f0ff,line-height:1.2
	classDef first fill-opacity:0
	classDef last fill:#bfb6fc

uv pip install -U langchain-tavily httpx

DEEPSEEK_API_KEY=your_deepseek_api_key
TAVILY_API_KEY=your_tavily_api_key

from typing import Annotated

from langchain.chat_models import init_chat_model
from langchain_tavily import TavilySearch
from langchain_core.messages import BaseMessage
from typing_extensions import TypedDict

from langgraph.graph import StateGraph
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition

import os
from dotenv import load_dotenv

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

class State(TypedDict):
    messages: Annotated[list, add_messages]


graph_builder = StateGraph(State)

llm = init_chat_model(
    "deepseek-chat",  # 使用DeepSeek模型
    api_key=os.environ.get("DEEPSEEK_API_KEY")
)

# 初始化Tavily搜索工具
tool = TavilySearch(max_results=2)
tools = [tool]

# 将工具绑定到LLM
llm_with_tools = llm.bind_tools(tools)

def chatbot(state: State):
    return {"messages": [llm_with_tools.invoke(state["messages"])]}

# 添加聊天机器人节点
graph_builder.add_node("chatbot", chatbot)

# 添加工具节点
tool_node = ToolNode(tools=[tool])
graph_builder.add_node("tools", tool_node)

# 添加条件边
graph_builder.add_conditional_edges(
    "chatbot",
    tools_condition,
)
# 工具调用完成后，返回到聊天机器人节点
graph_builder.add_edge("tools", "chatbot")
graph_builder.set_entry_point("chatbot")
graph = graph_builder.compile()

# 打印图结构
print(graph.get_graph().draw_mermaid())

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!")
            break
        stream_graph_updates(user_input)
    except KeyboardInterrupt:
        print("\nGoodbye!")
        break

from langchain_tavily import TavilySearch
from langgraph.prebuilt import ToolNode, tools_condition

tool = TavilySearch(max_results=2)
tools = [tool]

llm_with_tools = llm.bind_tools(tools)

tool_node = ToolNode(tools=[tool])
graph_builder.add_node("tools", tool_node)

graph_builder.add_conditional_edges(
    "chatbot",
    tools_condition,
)

graph_builder.add_edge("tools", "chatbot")
graph_builder.set_entry_point("chatbot")
graph = graph_builder.compile()

print(graph.get_graph().draw_mermaid())

uv run 2-add-tools.py

---
config:
  flowchart:
    curve: linear
---
graph TD;
	__start__([<p>__start__</p>]):::first
	chatbot(chatbot)
	tools(tools)
	__end__([<p>__end__</p>]):::last
	__start__ --> chatbot;
	chatbot -.-> __end__;
	chatbot -.-> tools;
	tools --> chatbot;
	classDef default fill:#f2f0ff,line-height:1.2
	classDef first fill-opacity:0
	classDef last fill:#bfb6f

User: 什么是LangGraph的核心组件？

Assistant:
Assistant: {"query": "LangGraph 核心组件", "follow_up_questions": null, "answer": null, "images": [], "results": [{"url": "https://cloud.tencent.com/developer/news/2180955", "title": "Agent快速构建框架的langGraph到底是什么及案例 - 腾讯云", "content": "LangGraph是一个使用 LLM 构建有状态、多参与者应用程序的库，用于创建代理和多代理工作流。请在此处查看入门教程。\n\nLangGraph 的灵感来自Pregel和Apache Beam。公共接口灵感来自NetworkX。LangGraph 由 LangChain 的创建者 LangChain Inc 构建，但可以在没有 LangChain 的情况下使用。\n\nLangGraph核心组件\n\nLangGraph的核心组件包括Graphs、State、Nodes、Edges、Send、Checkpointer。LangGraph有哪些优点：可控性、持久性、Human-in-the-loop、ReactAgent。\n\nstate\n\nfrom langgraph.graph import StateGraph\n\nfrom typing import TypedDict, List, Annotated\n\nimport Operator\n\nclass State(TypedDict):\n\n  input: str", "score": 0.9403148, "raw_content": null}, {"url": "https://blog.csdn.net/TaloyerG/article/details/142546057", "title": "Langgraph 的Planning Agent 深度解析原创 - CSDN博客", "content": "Langgraph 的核心组件包括语言理解模块、知识库和规划模块。 1.2.1 语言理解模块. 语言理解模块负责解析用户的自然语言输入，将其转化为可供计算机处理的", "score": 0.9348309, "raw_content": null}], "response_time": 1.87}
Assistant: LangGraph的核心组件主要包括以下几个方面：

1. **Graphs（图）**：用于构建和管理工作流的结构。
2. **State（状态）**：用于跟踪和管理应用程序的状态。
3. **Nodes（节点）**：表示工作流中的单个步骤或操作。
4. **Edges（边）**：定义节点之间的关系和流程控制。
5. **Send（发送）**：用于在节点之间传递数据或消息。
6. **Checkpointer（检查点）**：用于持久化和恢复工作流的状态。

此外，LangGraph还支持可控性、持久性、Human-in-the-loop（人工干预）和ReactAgent（反应式代理）等功能特性。

这些组件共同构成了LangGraph的基础，使其能够高效地构建和管理有状态的多参与者应用程序或代理工作流。

User: exit
Goodbye!

"""LangGraph 教程: 添加记忆功能的聊天机器人

本示例展示了如何使用 LangGraph 的检查点功能为聊天机器人添加记忆功能，
使其能够记住对话历史并在多轮对话中保持上下文。
"""

from typing import Annotated

from langchain.chat_models import init_chat_model
from langchain_tavily import TavilySearch
from langchain_core.messages import BaseMessage
from typing_extensions import TypedDict

# 导入 MemorySaver 用于实现记忆功能
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition

import os
from dotenv import load_dotenv

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

# 定义状态类型，使用 add_messages 注解来自动合并消息列表
class State(TypedDict):
    messages: Annotated[list, add_messages]  # 消息列表将使用 add_messages reducer 自动合并


# 初始化 DeepSeek 聊天模型
llm = init_chat_model(
    "deepseek-chat",  # 使用DeepSeek模型
    api_key=os.environ.get("DEEPSEEK_API_KEY")  # 从环境变量中获取API密钥
)

# 创建状态图构建器
graph_builder = StateGraph(State)

# 初始化Tavily搜索工具
print("\n初始化Tavily搜索工具...")
tool = TavilySearch(max_results=2)  # 设置最多返回2个搜索结果
tools = [tool]

# 将工具绑定到LLM
llm_with_tools = llm.bind_tools(tools)

# 定义聊天机器人节点函数
def chatbot(state: State):
    """LLM节点函数，处理用户输入并生成响应"""
    return {"messages": [llm_with_tools.invoke(state["messages"])]}

# 添加聊天机器人节点
graph_builder.add_node("chatbot", chatbot)

# 添加工具节点
tool_node = ToolNode(tools=[tool])
graph_builder.add_node("tools", tool_node)

# 添加条件边
graph_builder.add_conditional_edges(
    "chatbot",
    tools_condition,
)
# 工具调用完成后，返回到聊天机器人节点
graph_builder.add_edge("tools", "chatbot")
graph_builder.set_entry_point("chatbot")

print("\n构建图并添加记忆功能...")

# 创建内存保存器
print("\n创建 MemorySaver 实例作为检查点保存器...")
memory = MemorySaver()  # 在内存中保存状态，适用于开发和测试

# 使用内存保存器编译图
print("使用检查点保存器编译图...")
graph = graph_builder.compile(checkpointer=memory)  # 将内存保存器传递给图

# 打印图结构
print("\n图结构如下：")
print(graph.get_graph().draw_mermaid())

# 定义对话线程ID
print("\n设置对话线程 ID = '1'...")
config = {"configurable": {"thread_id": "1"}}  # 使用线程ID来标识和区分不同的对话

# 示例 1: 第一次对话
print("\n示例 1: 第一次对话 - 用户介绍自己")
user_input = "Hi there! My name is Will."
print(f"\n用户输入: '{user_input}'")

# 注意: config 是 stream() 函数的第二个参数!
print("使用线程 ID '1' 调用图...")
events = graph.stream(
    {"messages": [{"role": "user", "content": user_input}]},
    config,  # 传递包含 thread_id 的配置
    stream_mode="values",
)

print("\n助理回应:")
for event in events:
    event["messages"][-1].pretty_print()  # 打印助理的回应

# 示例 2: 测试记忆功能
print("\n\n示例 2: 第二次对话 - 测试记忆功能")
user_input = "Remember my name?"
print(f"\n用户输入: '{user_input}'")

# 使用相同的线程ID再次调用图
print("使用相同的线程 ID '1' 再次调用图...")
events = graph.stream(
    {"messages": [{"role": "user", "content": user_input}]},
    config,  # 使用相同的配置，图将加载之前保存的状态
    stream_mode="values",
)

print("\n助理回应 (应该记得用户名字):")
for event in events:
    event["messages"][-1].pretty_print()

# 示例 3: 新对话线程
print("\n\n示例 3: 新对话线程 - 测试线程隔离")
print("创建新的线程 ID = '2'...")

# 使用不同的线程ID
print("使用新的线程 ID '2' 调用图...")
events = graph.stream(
    {"messages": [{"role": "user", "content": user_input}]},
    {"configurable": {"thread_id": "2"}},  # 使用新的线程ID
    stream_mode="values",
)

print("\n助理回应 (不应该记得用户名字):")
for event in events:
    event["messages"][-1].pretty_print()

# 示例 4: 返回第一个线程
print("\n\n示例 4: 返回第一个线程 - 验证记忆持久性")
print(f"\n用户输入: '{user_input}'")

# 再次使用第一个线程ID
print("再次使用线程 ID '1' 调用图...")
events = graph.stream(
    {"messages": [{"role": "user", "content": user_input}]},
    config,  # 使用原始线程ID
    stream_mode="values",
)

print("\n助理回应 (应该仍然记得用户名字):")
for event in events:
    event["messages"][-1].pretty_print()

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()
graph = graph_builder.compile(checkpointer=memory)

config = {"configurable": {"thread_id": "1"}}

def stream_graph_updates(user_input: str):
    # 注意：config是stream()的第二个位置参数
    for event in graph.stream(
        {"messages": [{"role": "user", "content": user_input}]},
        config,
        stream_mode="values",
    ):
        for value in event.values():
            print("Assistant:", value["messages"][-1].content)

uv run 3-add-memory.py

---
config:
  flowchart:
    curve: linear
---
graph TD;
	__start__([<p>__start__</p>]):::first
	chatbot(chatbot)
	tools(tools)
	__end__([<p>__end__</p>]):::last
	__start__ --> chatbot;
	chatbot -.-> __end__;
	chatbot -.-> tools;
	tools --> chatbot;
	classDef default fill:#f2f0ff,line-height:1.2
	classDef first fill-opacity:0
	classDef last fill:#bfb6fc

设置对话线程 ID = '1'...

示例 1: 第一次对话 - 用户介绍自己

用户输入: 'Hi there! My name is Will.'
使用线程 ID '1' 调用图...

助理回应:
================================ Human Message =================================

Hi there! My name is Will.
================================== Ai Message ==================================

Hi Will! It's great to meet you. How can I assist you today?


示例 2: 第二次对话 - 测试记忆功能

用户输入: 'Remember my name?'
使用相同的线程 ID '1' 再次调用图...

助理回应 (应该记得用户名字):
================================ Human Message =================================

Remember my name?
================================== Ai Message ==================================

Of course, Will! I’ll remember your name for the rest of our conversation. How can I help you today? 😊


示例 3: 新对话线程 - 测试线程隔离
创建新的线程 ID = '2'...
使用新的线程 ID '2' 调用图...

助理回应 (不应该记得用户名字):
================================ Human Message =================================

Remember my name?
================================== Ai Message ==================================

I don’t have the ability to remember personal details like names between interactions. However, you can tell me your name, and I’ll happily use it for the rest of our current conversation! What should I call you?


示例 4: 返回第一个线程 - 验证记忆持久性

用户输入: 'Remember my name?'
再次使用线程 ID '1' 调用图...

助理回应 (应该仍然记得用户名字):
================================ Human Message =================================

Remember my name?
================================== Ai Message ==================================

Got it, Will! I’ll keep your name in mind while we chat. Let me know how I can assist you! 😊