1. 环境准备与基础概念回顾在开始构建有状态智能体之前我们需要先确保开发环境就绪。我推荐使用Python 3.10版本这个版本在稳定性和新特性支持上达到了很好的平衡。安装LangGraph非常简单只需要一条命令pip install langgraph langchain-openai如果你像我一样喜欢用conda管理环境可以先创建一个独立环境conda create -n langgraph-demo python3.10 conda activate langgraph-demo这里有个小技巧我习惯在项目根目录下放一个requirements.txt文件记录所有依赖包及其版本号。这样下次重装环境时直接pip install -r requirements.txt就能一键恢复。这个习惯帮我省去了很多为什么在我的机器上能运行的尴尬时刻。说到有状态智能体它最迷人的特点就是能记住对话历史。想象你在和一个健谈的朋友聊天如果对方每句话都像初次见面一样问你怎么称呼那该多扫兴。传统无状态Agent就像这个健忘的朋友而有状态智能体则能记住上下文让对话自然流畅。2. 构建第一个对话节点我们先从最简单的打招呼节点开始。这个节点会根据是否首次见面给出不同的问候语。下面是代码实现from langgraph.graph import StateGraph, END from typing import Dict, TypedDict class AgentState(TypedDict): is_first_time: bool user_name: str def greeting_node(state: AgentState): if state.get(is_first_time, True): return {response: 你好我是你的AI助手怎么称呼你} else: return {response: f{state[user_name]}又见面啦今天想聊点什么} workflow StateGraph(AgentState) workflow.add_node(greeting, greeting_node)这里有几个关键点需要注意我们定义了一个AgentState类型明确声明了状态中包含哪些字段greeting_node会根据is_first_time状态值返回不同的问候语使用StateGraph创建图结构时需要指定状态类型我第一次写这段代码时犯了个错误忘记处理user_name可能不存在的情况。结果当新用户第一次打招呼时程序就直接报错了。所以现在养成了习惯所有状态字段访问都用.get()方法并设置合理的默认值。3. 添加状态更新逻辑现在我们的智能体会打招呼了但还不会记住用户信息。让我们添加一个状态更新节点def update_user_info_node(state: AgentState): # 假设用户回复包含在state[user_input]中 if state.get(is_first_time, True): return { is_first_time: False, user_name: extract_name(state[user_input]) } return state def extract_name(text: str) - str: # 简单实现取第一个非空行的前两个字符作为称呼 lines [line.strip() for line in text.split(\n) if line.strip()] if not lines: return 朋友 return lines[0][:2] 先生/女士 workflow.add_node(update_info, update_user_info_node)这里我设计了一个简单的姓名提取函数。在实际项目中你可能会用更复杂的方法比如调用大模型提取姓名实体结合用户账户系统获取正式名称使用正则匹配常见称呼模式记得添加边来连接这些节点workflow.add_edge(greeting, update_info) workflow.add_edge(update_info, END) workflow.set_entry_point(greeting) app workflow.compile()4. 实现多轮对话循环真正的对话不会一次就结束。我们需要改造图结构支持循环对话from langgraph.graph import START def user_input_node(state: AgentState): # 模拟用户输入 user_input input(你) return {user_input: user_input} def should_continue(state: AgentState): # 简单的结束条件用户输入再见 if 再见 in state.get(user_input, ): return end return continue workflow.add_node(get_input, user_input_node) workflow.add_conditional_edges( update_info, should_continue, {continue: get_input, end: END} ) workflow.add_edge(get_input, greeting)现在对话流程变成了打招呼 → 更新信息 → 获取输入 → 判断是否继续。我在测试时发现如果用户直接说再见流程会异常终止。所以又在greeting_node里加了保护逻辑def greeting_node(state: AgentState): if user_input in state and 再见 in state[user_input]: return {response: 好的下次见} # 原有逻辑...5. 添加记忆功能让我们给智能体增加些记忆力让它能记住对话历史class AgentState(TypedDict): is_first_time: bool user_name: str user_input: str conversation_history: list[str] def log_conversation_node(state: AgentState): history state.get(conversation_history, []) history.append(f用户{state[user_input]}) history.append(f助手{state[response]}) return {conversation_history: history} workflow.add_node(log_conversation, log_conversation_node)然后调整边连接workflow.add_edge(greeting, log_conversation) workflow.add_edge(log_conversation, update_info)现在每次对话都会被记录下来。我们可以进一步利用这些历史信息比如在问候时提及上次的话题def greeting_node(state: AgentState): if state.get(is_first_time, True): return {response: 你好我是你的AI助手怎么称呼你} else: last_topic extract_last_topic(state.get(conversation_history, [])) return { response: f{state[user_name]}我们上次聊到了{last_topic}。要继续这个话题吗 }6. 集成大语言模型现在让我们引入真正的AI能力 - 集成OpenAI的聊天模型from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-3.5-turbo) def generate_response_node(state: AgentState): history state.get(conversation_history, []) prompt \n.join(history[-6:]) # 取最近3轮对话 response llm.invoke(prompt) return {response: response.content} workflow.insert_node_before(greeting, generate_response, generate_response_node)这里有几个优化点限制历史对话长度避免提示词过长使用最新的gpt-3.5-turbo模型性价比高将生成节点放在问候节点之前形成完整流程记得设置环境变量export OPENAI_API_KEY你的API密钥7. 完整示例与调试技巧把所有代码整合起来我们的智能体现在具备记忆用户信息记录对话历史调用大语言模型支持多轮对话完整运行示例# 初始化状态 state {is_first_time: True} # 运行对话循环 for _ in range(5): # 最多5轮对话 state app.invoke(state) print(助手, state[response]) if 再见 in state.get(response, ): break调试这类有状态系统时我常用的技巧是在关键节点打印状态快照使用pdb设置断点检查状态流转保存历史状态到文件方便复现问题为每个节点编写单元测试比如可以添加调试节点def debug_node(state: AgentState): print(当前状态, state) return state workflow.add_node(debug, debug_node) workflow.add_edge(generate_response, debug)8. 性能优化与扩展思路当你的智能体开始处理真实流量时可能需要考虑状态存储优化使用Redis缓存热门会话状态实现状态压缩算法减少内存占用定期清理闲置会话对话质量提升添加敏感词过滤节点实现话题引导机制加入情感分析节点调整回复语气架构扩展将不同功能拆分为子图添加人工审核节点实现A/B测试框架一个典型的生产级优化是添加限流节点from datetime import datetime class AgentState(TypedDict): # 原有字段... last_request_time: datetime request_count: int def rate_limit_node(state: AgentState): now datetime.now() if (now - state.get(last_request_time, now)).seconds 1: state[request_count] 1 if state[request_count] 3: return {response: 请求过于频繁请稍后再试} else: state[request_count] 0 state[last_request_time] now return state构建有状态智能体就像教AI与人相处 - 需要耐心地定义每个交互细节同时保留足够的灵活性应对各种情况。我在实际项目中最大的收获是状态设计要像日记一样清晰明了节点功能要像乐高积木一样专注单一。这样当需求变更时你只需要调整图结构而不必重写大量代码。