lLangGraph实战：状态即上下文，智能对话的四大管理法则

董厂长

850人浏览 · 2026-01-30 23:55:05

董厂长 · 2026-01-30 23:55:05 发布

在构建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"基于您的偏好，我的推荐：\n{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"请用一段话总结以下对话的核心内容：\n{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成本	总结旧历史、保留关键信息
选择	提取相关上下文	任务专业化处理	预算规划只关注财务相关对话
隔离	创建独立上下文	子任务、避免污染	机票搜索不接触餐饮偏好