lLangGraph实战：状态即上下文，智能对话的四大管理法则
在构建AI应用时，我们常常听到"上下文"这个术语，但LangGraph通过"状态"这一概念，将上下文管理提升到了新的高度。本文将通过一个实战例子，深入探讨状态即上下文的核心理念，并详解四种上下文管理方法。
引言：为什么需要上下文管理？
想象一下这个场景：你正在和旅行助手聊天，已经讨论了目的地、预算和兴趣。当你说"帮我订一个那里附近的酒店"时，AI需要理解"那里"指的是之前提到的目的地，"附近"需要结合之前的兴趣点。这就是上下文的力量。
传统AI应用往往在上下文管理上力不从心，而LangGraph通过状态（State） 这一核心概念，为我们提供了一套完整的解决方案。
实战例子：智能旅行规划助手
让我们从一个完整的例子开始，逐步深入理解状态即上下文的概念：
python
import os
from typing import TypedDict, List, Annotated, Optional, Literal
from langgraph.graph import StateGraph, END, START
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from datetime import datetime
# 定义状态结构 - 这就是我们的上下文容器
class TravelState(TypedDict):
"""旅行规划的状态/上下文"""
messages: Annotated[List, add_messages] # 对话历史上下文
user_id: str # 用户身份上下文
destination: Optional[str] # 目的地上下文
budget: Optional[float] # 预算上下文
interests: List[str] # 兴趣偏好上下文
travel_dates: Optional[str] # 旅行日期上下文
current_step: str # 当前流程步骤上下文
extracted_info: dict # 已提取信息上下文
conversation_summary: str # 对话摘要上下文状态即上下文：LangGraph的核心哲学
在LangGraph中，状态不仅仅是数据容器，更是工作流的智能记忆系统。让我们看看这个旅行助手如何使用状态管理上下文：
python
# 初始化LLM
llm = ChatOpenAI(model="gpt-3.5-turbo")
def extract_travel_info(state: TravelState):
"""提取旅行信息节点 - 展示上下文写入"""
# 获取完整的对话历史作为上下文
conversation = state["messages"]
# 使用LLM从对话中提取结构化信息
extraction_prompt = f"""
从以下对话中提取旅行信息：
{conversation}
请提取以下信息（如果提到）：
1. 目的地
2. 预算（数字）
3. 兴趣（列表）
4. 旅行日期
以JSON格式返回。
"""
response = llm.invoke([{"role": "user", "content": extraction_prompt}])
# 解析响应并更新状态（写入上下文）
import json
try:
extracted = json.loads(response.content)
except:
extracted = {}
# 更新状态 - 这就是上下文写入
updates = {
"destination": extracted.get("destination"),
"budget": extracted.get("budget"),
"interests": extracted.get("interests", []),
"travel_dates": extracted.get("travel_dates"),
"extracted_info": extracted,
"current_step": "planning"
}
# 同时将提取结果添加到对话历史
updates["messages"] = [{
"role": "assistant",
"content": f"已提取您的旅行偏好：{extracted}"
}]
return updates
def generate_recommendations(state: TravelState):
"""生成推荐节点 - 展示上下文使用"""
# 基于完整上下文生成个性化推荐
prompt = f"""
基于以下旅行上下文为用户生成推荐：
用户信息：
- 目的地：{state.get('destination', '未指定')}
- 预算：{state.get('budget', '未指定')}
- 兴趣：{', '.join(state.get('interests', []))}
- 旅行日期：{state.get('travel_dates', '未指定')}
对话历史摘要：{state.get('conversation_summary', '无')}
请提供：
1. 3个景点推荐
2. 餐饮建议
3. 活动推荐
"""
response = llm.invoke([{"role": "user", "content": prompt}])
return {
"messages": [{
"role": "assistant",
"content": f"基于您的偏好，我的推荐：
{response.content}"
}],
"current_step": "completed"
}四大上下文管理方法
方法一：写入 - 构建完整上下文
写入是上下文管理的基础，确保所有相关信息都被记录到状态中：
python
def collect_destination(state: TravelState):
"""收集目的地信息 - 上下文写入示例"""
# 从对话历史中获取最新的用户消息
last_user_message = None
for msg in reversed(state["messages"]):
if msg["role"] == "user":
last_user_message = msg["content"]
break
if last_user_message and "巴黎" in last_user_message:
# 写入目的地上下文
# 同时写入相关上下文（如假设的预算和兴趣）
return {
"destination": "巴黎",
"budget": 5000.0, # 假设的默认预算
"interests": ["艺术", "美食"], # 假设的兴趣
"current_step": "collecting_details",
"messages": [{
"role": "assistant",
"content": "巴黎是个好选择！您是喜欢艺术还是美食更多呢？"
}]
}
return {"messages": [{"role": "assistant", "content": "您想去哪里旅行呢？"}]}关键点：写入不仅仅是添加数据，更是建立数据间的关联，形成丰富的上下文网络。
方法二：压缩 - 管理上下文长度
当对话历史过长时，我们需要压缩上下文以节省token并保持相关性：
python
def compress_conversation(state: TravelState, max_messages: int = 10):
"""压缩对话历史 - 上下文压缩示例"""
messages = state["messages"]
if len(messages) <= max_messages:
return state # 无需压缩
# 策略1：保留系统消息和最近对话
system_messages = [m for m in messages if m["role"] == "system"]
recent_messages = messages[-max_messages//2:] # 保留最近的一半
# 策略2：总结旧的历史
old_messages = messages[:len(messages)-max_messages//2]
summary_prompt = f"请用一段话总结以下对话的核心内容：
{old_messages}"
summary_response = llm.invoke([{"role": "user", "content": summary_prompt}])
summary_message = {
"role": "system",
"content": f"对话历史摘要：{summary_response.content}"
}
# 构建压缩后的消息列表
compressed_messages = system_messages + [summary_message] + recent_messages
# 更新状态
return {
"messages": compressed_messages,
"conversation_summary": summary_response.content
}
# 在图中添加压缩节点
def compression_node(state: TravelState):
"""压缩节点 - 自动触发压缩"""
if len(state["messages"]) > 15: # 超过15条消息时压缩
return compress_conversation(state)
return {} # 无更新方法三：选择 - 智能上下文筛选
不是所有上下文对每个任务都重要，选择让我们能提取最相关的部分：
python
def select_relevant_context(state: TravelState, context_type: str):
"""根据任务类型选择相关上下文"""
if context_type == "budget_planning":
# 对于预算规划，选择与预算相关的上下文
relevant_messages = []
for msg in state["messages"]:
content = msg.get("content", "").lower()
if any(word in content for word in ["预算", "价格", "花费", "成本", "money", "budget"]):
relevant_messages.append(msg)
# 添加上下文摘要
context_summary = f"""
预算规划相关上下文：
目的地：{state.get('destination')}
当前预算：{state.get('budget')}
相关对话历史：{len(relevant_messages)}条相关消息
"""
return {
"selected_context": relevant_messages,
"context_summary": context_summary
}
elif context_type == "activity_suggestion":
# 对于活动建议，选择兴趣相关的上下文
interests = state.get("interests", [])
relevant_messages = []
for msg in state["messages"]:
content = msg.get("content", "")
if any(interest in content for interest in interests):
relevant_messages.append(msg)
return {
"selected_context": relevant_messages,
"context_summary": f"用户兴趣：{', '.join(interests)}"
}
return {"selected_context": state["messages"][-5:]} # 默认选择最近5条
# 应用选择策略的节点
def budget_advisor_node(state: TravelState):
"""预算建议节点 - 使用选择的上下文"""
# 选择与预算相关的上下文
selected = select_relevant_context(state, "budget_planning")
prompt = f"""
基于以下上下文提供预算建议：
{selected['context_summary']}
相关对话片段：
{selected['selected_context'][-3:] if selected['selected_context'] else '无'}
请提供详细的预算分配建议。
"""
response = llm.invoke([{"role": "user", "content": prompt}])
return {
"messages": [{"role": "assistant", "content": response.content}]
}方法四：隔离 - 上下文边界管理
有些任务需要独立的上下文空间，避免信息污染：
python
def create_subgraph_for_flight_search(state: TravelState):
"""创建机票搜索子图 - 上下文隔离示例"""
# 提取机票搜索所需的最小上下文
flight_context = {
"destination": state.get("destination"),
"travel_dates": state.get("travel_dates"),
"user_id": state.get("user_id")
}
# 在隔离的上下文中执行机票搜索
# 这里不会接触到主对话的其他细节
def flight_search_subgraph(flight_state):
"""机票搜索子图 - 独立上下文"""
# 模拟API调用
flights = [
{"airline": "Airline A", "price": 800, "time": "10:00"},
{"airline": "Airline B", "price": 750, "time": "14:00"}
]
# 根据预算过滤（仅使用子图自己的上下文）
budget = state.get("budget", float('inf'))
filtered_flights = [f for f in flights if f["price"] <= budget/3] # 假设机票占预算1/3
return {
"available_flights": filtered_flights,
"search_completed": True
}
# 执行子图
flight_results = flight_search_subgraph(flight_context)
# 将结果整合回主上下文
return {
"extracted_info": {
**state.get("extracted_info", {}),
"flight_options": flight_results["available_flights"]
},
"messages": [{
"role": "assistant",
"content": f"找到{len(flight_results['available_flights'])}个航班选项"
}]
}完整的工作流示例
让我们将这些方法整合到一个完整的旅行规划图中：
python
from langgraph.graph import StateGraph, END
def build_travel_planner():
"""构建完整的旅行规划图"""
builder = StateGraph(TravelState)
# 添加节点
builder.add_node("start_conversation", collect_destination)
builder.add_node("extract_info", extract_travel_info)
builder.add_node("compress_context", compression_node)
builder.add_node("budget_advice", budget_advisor_node)
builder.add_node("flight_search", create_subgraph_for_flight_search)
builder.add_node("make_recommendations", generate_recommendations)
# 设置流程
builder.set_entry_point("start_conversation")
# 定义边
builder.add_edge("start_conversation", "extract_info")
builder.add_conditional_edges(
"extract_info",
lambda state: "compress" if len(state.get("messages", [])) > 15 else "budget_advice"
)
builder.add_edge("compress_context", "budget_advice")
builder.add_edge("budget_advice", "flight_search")
builder.add_edge("flight_search", "make_recommendations")
builder.add_edge("make_recommendations", END)
return builder.compile()
# 使用示例
async def main():
travel_planner = build_travel_planner()
# 初始状态
initial_state = {
"messages": [
{"role": "system", "content": "你是一个专业的旅行规划助手"},
{"role": "user", "content": "我想去巴黎旅行，预算5000元"}
],
"user_id": "user123",
"destination": None,
"budget": None,
"interests": [],
"travel_dates": None,
"current_step": "start",
"extracted_info": {},
"conversation_summary": ""
}
# 执行图
result = await travel_planner.ainvoke(initial_state)
print("最终推荐:", result["messages"][-1]["content"])
print("提取的信息:", result["extracted_info"])
print("当前步骤:", result["current_step"])四种方法的对比总结
| 方法 | 目的 | 适用场景 | 示例 |
|---|---|---|---|
| 写入 | 记录和丰富上下文 | 信息收集阶段 | 记录用户偏好、提取结构化数据 |
| 压缩 | 减少上下文长度 | 长对话、节省token成本 | 总结旧历史、保留关键信息 |
| 选择 | 提取相关上下文 | 任务专业化处理 | 预算规划只关注财务相关对话 |
| 隔离 | 创建独立上下文 | 子任务、避免污染 | 机票搜索不接触餐饮偏好 |
最佳实践建议
-
分层管理上下文：将上下文分为会话级、用户级、应用级
-
适时压缩：定期检查上下文长度，避免无限增长
-
智能选择：根据任务类型动态选择相关上下文
-
明确隔离边界：清晰定义子任务的输入输出，避免隐式依赖
结语
LangGraph通过状态即上下文的理念，为AI应用提供了强大的上下文管理能力。掌握写入、压缩、选择和隔离这四种方法，你将能够构建出真正智能、高效且可维护的对话系统。
上下文不再是隐式的、难以管理的负担，而是成为了我们构建智能应用的强大工具。在LangGraph的世界里，状态不仅记录了发生了什么，更指导着接下来会发生什么——这就是现代AI应用上下文管理的艺术。
