实战案例：创建一个基础聊天机器人

目录

2.3.1 创建调用deepseek-v3的聊天机器人

2.3.2 案例代码解析

1. 导入必要的库

2. 加载环境变量

3. 定义状态（State）

4. 创建图构建器

5. 初始化模型

6. 定义节点函数（Node Function）

7. 构建图（Graph）

8. 实现流式响应

9. 主循环（Main Loop）

10. 总结

2.3.3 运行聊天机器人

基于LangGraph搭建状态机框架，定义对话状态与节点，对接阿里云百炼提供的DeepSeek模型构建基础对话流程。

2.3.1 创建调用deepseek-v3的聊天机器人

现在，让我们开始实现一个基础的聊天机器人。首先创建一个名为basic-chatbot.py的文件，并添加以下代码。

【示例2.2】基础聊天机器人实现代码（basic-chatbot.py）。

from typing import Annotated

from langchain_openai import ChatOpenAI

from typing_extensions import TypedDict

from langgraph.graph import StateGraph, START

from langgraph.graph.message import add_messages

#启用dotenv读取环境变量

import os

from dotenv import load_dotenv

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

load_dotenv()

class State(TypedDict):

messages: Annotated[list, add_messages]

graph_builder = StateGraph(State)

#从环境变量读取API Key

llm = ChatOpenAI(

model="deepseek-v3",

api_key=os.getenv("DASHSCOPE_API_KEY"), #从.env读取

base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",

temperature=0

)

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

运行输出（可能与此处给出的结果存在区别）：

User:请问南果梨的产地与水果特色？

Assistant:南果梨是一种具有独特风味和地域特色的水果，以下是关于其产地和特色的详细介绍：

### **产地**

1. **核心产区**

- **辽宁省鞍山市**：尤其是海城市、千山区（原旧堡区）及周边地区，是南果梨最著名的产地，这里的气候和土壤条件（偏酸性棕壤土）非常适宜其生长。

- **其他地区**：辽宁省的辽阳、营口等地也有种植，但品质和风味以鞍山产的最为突出。

2. **生长环境**

-南果梨喜昼夜温差大的温带气候，鞍山地区秋季光照充足、温差显著，利于糖分积累，形成独特香气。

### **水果特色**

1. **外观与口感**

- **外形**：果实较小（单果约50-100克），成熟时果皮黄绿色带红晕，表面有细小果点。

- **质地**：刚采摘时脆硬，后熟后果肉变软糯，细腻多汁，入口即化。

- **风味**：酸甜适中，具有浓郁的复合果香（类似香蕉、玫瑰的香气），甜度可达14%-16%。

2. **独特后熟特性**

-需常温放置3-7天后熟，果肉由硬变软，香气充分释放，此时为最佳食用期。

3. **营养价值**

-富含维生素C、花青素、矿物质（如钙、铁），有抗氧化作用，有助于促进消化。

4. **文化地位**

-被列为中国四大名梨之一（与库尔勒香梨、莱阳梨等并列），2005年成为国家地理标志产品，是鞍山的农业名片。

### **其他信息**

- **采收与保鲜**：9月上旬成熟，常温保存约1-2周，冷藏可延长至1个月，但后熟后需尽快食用。

- **深加工**：常用于制作果汁、果酒、果脯等产品，进一步拓展其经济价值。

南果梨以其“香气袭人、口感独特”著称，是兼具地域性和品质优势的特色水果，适合喜欢酸甜风味和馥郁果香的人群品尝。

User:

2.3.2 案例代码解析

1. 导入必要的库

这部分代码的作用是引入程序运行所依赖的所有工具和框架。

from typing import Annotated

from langchain_openai import ChatOpenAI

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

（1）from typing import Annotated：这是Python 3.9+提供的类型提示增强功能。它允许为变量或函数参数添加元数据（注解）。在LangGraph中，它被用来为状态中的某些字段（如messages）附加特殊的行为。

（2）from langchain_openai import ChatOpenAI：这是LangChain库中用于初始化大模型的统一入口函数。它可以根据提供的模型名称（如"deepseek-v3"）自动选择并配置对应的模型封装。

（3）from typing_extensions import TypedDict：用于创建强类型的字典。在LangGraph中，它是定义工作流状态（State）结构的标准方式，能让状态的每个字段都有明确的类型定义，提高代码可读性和健壮性。

（4）from langgraph.graph import StateGraph,START：StateGraph是LangGraph的核心类，用于构建一个由节点和边组成的有状态工作流（图）。START是一个特殊的常量，代表图中的起始节点。

（5）from langgraph.graph.message import add_messages：这是一个LangGraph提供的状态更新策略。它定义了当新的消息产生时，应该如何更新状态中的messages列表。具体来说，它会将新消息追加到列表末尾，并自动处理ai和human角色的消息封装。

（6）import os：Python内置的用于与操作系统交互的库。这里主要用来获取环境变量。

（7）from dotenv import load_dotenv：一个第三方库，用于从.env文件中加载环境变量到程序的运行环境中。这是管理API密钥等敏感信息的最佳实践。

2. 加载环境变量

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

load_dotenv()

这行代码会查找当前目录下名为.env的文件，并将文件中定义的键值对（如DASHSCOPE_API_KEY=sk-xxx）加载到Python的os.environ字典中，目的是安全地管理敏感信息（如API密钥），避免将它们硬编码在代码中。

3.定义状态（State）

class State(TypedDict):

messages: Annotated[list, add_messages]

这是整个工作流的核心数据结构，它定义了在图的各个节点之间传递和修改的数据。

（1）class State(TypedDict)语句定义了一个名为State的类，它继承自TypedDict。这意味着State的实例是一个字典，但它的键（messages）和对应的值类型（list）是固定的。

（2）messages:Annotated[list, add_messages]语句中，messages是状态中唯一的字段，它将存储一个消息列表，每个消息都是一个字典（如{"role":"user","content":"你好"}）。Annotated[list, add_messages]在这里发挥了关键作用，它告诉LangGraph，messages字段是一个列表，当有新的消息需要更新到这个字段时，应该使用add_messages策略。add_messages会智能地将新消息追加到现有列表的末尾，确保对话历史的连续性。

4. 创建图构建器

graph_builder = StateGraph(State)

这行代码创建了一个StateGraph的实例，命名为graph_builder。在创建时，我们将刚刚定义的State类作为参数传入。这告诉图构建器，整个工作流将围绕State这种数据结构来展开。graph_builder将负责管理节点、边以及状态在它们之间的流转。

5. 初始化模型

llm = ChatOpenAI(

model="deepseek-v3",

api_key=os.getenv("DASHSCOPE_API_KEY"), #从.env读取

base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",

temperature=0

)

这行代码初始化了一个DeepSeek模型（deepseek-v3模型）。

（1）ChatOpenAI("deepseek-v3",...)：ChatOpenAI函数根据"deepseek-v3"这个字符串，自动识别并调用LangChain中与阿里云百炼提供的DeepSeek模型对应的封装。

（2）api_key=os.getenv("DASHSCOPE_API_KEY")：从环境变量中获取DASHSCOPE的API密钥，并传递给模型。如果环境变量未设置，os.getenv()会返回None，可能导致模型初始化失败。

6.定义节点函数（Node Function）

def chatbot(state: State):

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

这是一个节点函数，它定义了图中一个具体的处理步骤。当工作流执行到这个节点时，这个函数就会被调用。

（1）chatbot(state: State)：函数接收一个state参数，它的类型是我们之前定义的State。这个state包含了当前对话的所有信息（主要是messages列表）。

llm.invoke(state["messages"]):#这是函数的核心逻辑

它将state中的messages列表（即完整的对话历史）传递给大模型llm。模型会根据这些历史生成一个新的回复。

（2）return {"messages":[...]}：函数的返回值是一个字典。这个字典的结构必须与State类型相匹配。它告诉图构建器graph_builder：“请用我返回的这个新字典来更新全局状态”。

• llm.invoke(...)返回的是一个ChatMessage对象。
• [llm.invoke(...)]将这个对象放入一个列表中。
• {"messages":[...]}表示：“我要更新State中的messages字段，新的值是一个包含AI回复的列表”。

由于我们在State定义中使用了add_messages注解，LangGraph会自动将这个新列表中的消息追加到原有的messages列表后面，而不是替换它。

7.构建图（Graph）

#添加节点和边

graph_builder.add_node("chatbot", chatbot)

graph_builder.add_edge(START, "chatbot")

graph = graph_builder.compile()

这部分代码负责将节点和它们之间的连接关系组装成一个可执行的图。

• graph_builder.add_node("chatbot",chatbot)：向图中添加一个名为"chatbot"的节点，这个节点的执行逻辑由我们定义的chatbot函数来实现。
• graph_builder.add_edge(START,"chatbot")：定义了一条从START节点到"chatbot"节点的边。这告诉图执行引擎：“工作流开始时，请首先执行'chatbot'节点”。
• graph = graph_builder.compile()：这是最后一步，也是关键的一步。它将graph_builder对象编译成一个最终可执行的graph对象。在这个阶段，LangGraph会进行一些内部优化和验证，确保图的结构是正确的。

8. 实现流式响应

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)

这个函数封装了如何与已编译好的graph进行交互，以实现流式输出（即模型边生成、程序边打印）。

• graph.stream(...)：调用graph对象的stream方法来执行图。与invoke方法一次性获取所有结果不同，stream会以流式生成的方式逐步返回执行过程中的更新。
• {"messages":[{"role":"user","content":user_input}]}：这是启动图时传入的初始状态。它是一个State类型的字典，包含用户的最新输入。
• for event in graph.stream(...)：遍历流式返回的事件（event）。每个event代表图执行过程中的一个状态更新。
• for value in event.values()：每个event可能包含多个节点的更新，但在我们这个简单的图中，每次更新只来自"chatbot"节点。
• print("Assistant:",value["messages"][-1].content)：

• value：是更新后的部分状态（一个State字典）。
• value：["messages"]是更新后的消息列表。
• [-1]：取列表中的最后一个元素，也就是大模型刚刚生成的最新一条消息。
• .content：获取该消息的内容并打印。

9.主循环（Main Loop）

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

这是程序的入口点，一个标准的无限循环，用于持续与用户交互。

• while True：启动一个无限循环。
• user_input = input("User: ")：在控制台打印"User:"提示符，并等待用户输入。
• if user_input.lower() in ["quit","exit","q"]:break：如果用户输入了退出（exit）指令，打印告别信息并跳出循环，程序结束。
• stream_graph_updates(user_input)：如果用户输入了正常内容，则调用我们定义的流式响应函数，将用户输入传递给graph并处理和打印AI的回复。
• except KeyboardInterrupt：捕获用户按下Ctrl+C键的中断信号，同样打印告别信息并优雅地退出程序。

10. 总结

这段代码完整地演示了如何使用LangGraph构建一个最简单的、具有状态记忆（对话历史）的聊天机器人。

• 核心思想：将对话流程建模为一个状态（State）在图（Graph）中流转的过程。
• 状态（State）：State类定义了对话的全部数据（messages）。
• 节点（Node）：chatbot函数定义了具体的处理逻辑（调用LLM）。
• 图（Graph）：StateGraph将状态和节点组织起来，形成一个可执行的工作流。
• 执行：通过graph.stream()触发整个流程，并以流式方式获取结果。

2.3.3 运行聊天机器人

现在可以运行这个简单的聊天机器人了。确保你已经在.env文件中设置了有效的DeepSeek API密钥，然后在Miniconda Prompt命令行窗口中运行：

python basic-chatbot.py

或者直接在PyCharm中运行此代码，你将看到一个交互式的命令行界面，通过此界面可以与聊天机器人对话：

User:请问南果梨南果梨的产地与水果特色？

Assistant:当然！南果梨南果梨是中国非常有特色的一种水果，以其独特的香气和口感而闻名。以下是关于它的产地和水果特色的详细介绍：

一、核心产地

南果梨南果梨最著名、最核心的产地是辽宁省鞍山市的千山区（特别是大屯镇和接文镇）以及海城市。

...

User: exit

Goodbye!