LangGraph实战：构建带智能记忆的AI对话系统

1. 项目概述：为什么我们需要会话记忆？

如果你正在开发一个AI应用，无论是客服机器人、智能助手还是创意写作工具，你肯定遇到过这样的场景：用户问“我昨天提到的那个项目进展如何了？”，而你的AI助手一脸茫然地回答：“抱歉，我不记得您之前提到过什么项目。” 这种对话的割裂感，会让用户体验大打折扣，也让AI显得“很笨”。这就是“会话记忆”要解决的核心问题——让AI能够记住并理解对话的上下文。

在传统的聊天机器人开发中，实现记忆功能往往意味着你需要手动管理一个对话历史列表，每次调用大模型时，把整个历史记录一股脑地塞进提示词（Prompt）里。这种方法简单粗暴，但问题也显而易见：随着对话轮次增加，上下文会越来越长，不仅消耗大量Token（意味着更高的成本和更慢的响应），还可能因为超出模型的最大上下文长度而被截断，导致记忆丢失。

而LangGraph的出现，为我们提供了一种更优雅、更强大的解决方案。它不是一个独立的大模型，而是构建在LangChain之上的一个框架，专门用于创建有状态的、多步骤的工作流。你可以把它想象成一个流程图设计工具，但每个节点都是一个可以执行特定任务（如调用LLM、查询数据库）的“智能体”，节点之间的连线定义了数据（也就是“状态”）的流动规则。正是这种对“状态”的精细化管理能力，让实现复杂、可控的会话记忆变得可能。

在接下来的内容里，我不会只教你“如何用几行代码开启记忆功能”，那太浅了。我会带你深入LangGraph的内部，拆解它的三要素（State、Node、Edge），并手把手教你设计一个既能记住关键信息，又能自动清理冗余记忆的智能会话系统。无论你是刚学完前端想转型AI开发，还是正在面试AI应用开发岗位，理解并掌握这套方法，都将是你技术栈中一个亮眼的加分项。

2. LangGraph核心三要素深度解析

要玩转LangGraph，实现高级记忆功能，你必须吃透它的三个核心概念：State（状态）、Node（节点）和Edge（边）。很多人看了官方文档还是云里雾里，觉得抽象，接下来我用最直白的方式给你讲明白。

2.1 State：记忆的载体与蓝图

State是LangGraph的灵魂，它定义了你整个工作流中需要记住和传递的所有信息。你可以把它理解为一个Python的TypedDict或者Pydantic的BaseModel。它不仅仅是一个存储对话历史的“垃圾桶”，更是一份精心设计的“数据蓝图”。

一个设计良好的State，应该只包含必要的信息。比如，对于一个基础的聊天应用，你的State可能长这样：

from typing import TypedDict, List, Annotated from langgraph.graph.message import add_messages import operator class State(TypedDict): # 核心：消息历史。Annotated是LangGraph的魔法，add_messages是一个归约函数。 messages: Annotated[List[dict], add_messages] # 业务相关：从对话中提取出的结构化信息，如用户偏好、任务目标等。 user_profile: dict current_task: str # 系统相关：控制流程的标记，比如是否需要查询知识库。 requires_search: bool

这里的关键是messages字段。我们使用了Annotated[List[dict], add_messages]。add_messages是一个归约函数，它的作用是：当新消息到来时，它不是简单地替换掉旧的messages列表，而是将新消息追加到列表末尾。这是实现累积式记忆的基础。没有这个归约操作，状态就无法在节点间正确传递和更新。

实操心得：在设计State时，一定要遵循“最小必要”原则。不要一股脑地把所有中间计算结果都塞进去。只放那些需要在不同节点间共享，并且对后续步骤有决定作用的数据。过多的状态字段会让图变得难以理解和调试。比如，requires_search这样的控制标志就非常有用，它可以让一个节点决定下一步是调用LLM还是先执行搜索。

2.2 Node：执行具体任务的单元

Node就是图中的一个个“工作站点”。每个Node都是一个普通的Python函数（或可调用对象），它接收当前的整个State作为输入，然后返回一个包含要更新字段的字典。

一个典型的LLM调用节点可能是这样的：

from langchain_openai import ChatOpenAI llm = ChatOpenAI(model=“gpt-4o”) def call_llm(state: State): # 1. 从状态中获取消息历史 messages = state[“messages”] # 2. 调用LLM，传入历史消息作为上下文 response = llm.invoke(messages) # 3. 将LLM的回复作为一条新消息，准备更新到状态中 new_message = {“role”: “assistant”, “content”: response.content} # 4. 返回一个字典，LangGraph会用这个字典来更新State return {“messages”: [new_message]}

这个函数干了三件事：读取状态、执行逻辑（调用AI）、返回更新。LangGraph会把你返回的字典{“messages”: [new_message]}，与当前的State按照State定义中的规则（比如add_messages）进行合并，从而产生新的State。

你可以创建各种功能的节点：查询数据库的节点、调用外部API的节点、进行条件判断的节点等等。一个复杂的AI应用，就是由这些各司其职的节点协作完成的。

2.3 Edge：决定流程走向的规则

Edge定义了节点执行完毕后，接下来该去哪个节点。这是LangGraph实现复杂逻辑和循环的关键。边分为两种：普通边（Conditional Edge）和条件边（Conditional Edge）。

普通边是固定的连接，比如“节点A执行完，无条件进入节点B”。条件边则允许你根据当前State的内容，动态决定下一步。这通常通过一个路由函数来实现。

def route_after_search(state: State): # 根据State中的某个标志位决定下一步 if state.get(“requires_search”): # 如果需要搜索，则前往“search_knowledge_base”节点 return “search_knowledge_base” else: # 否则，直接去调用LLM生成回复 return “call_llm”

在构建图时，你会这样使用它：

from langgraph.graph import StateGraph, END workflow = StateGraph(State) # 添加节点 workflow.add_node(“call_llm”, call_llm) workflow.add_node(“search_knowledge_base”, search_function) # 设置条件边 workflow.add_conditional_edges( “start_node”, # 从哪个节点出发 route_after_search, # 路由函数 {“search_knowledge_base”: “search_knowledge_base”, “call_llm”: “call_llm”} # 可能的目的地映射 ) # 设置普通边 workflow.add_edge(“search_knowledge_base”, “call_llm”)

通过State、Node、Edge的灵活组合，你就能构建出像“先判断用户意图 -> 如需则查询知识库 -> 结合查询结果生成回复”这样的智能工作流。记忆（State）在这个流程中被每个节点读取和修改，从而实现了有状态的、上下文感知的对话。

3. 构建带记忆的会话工作流：从零到一

理解了核心概念，我们现在来动手搭建一个真正的、带记忆的聊天应用。我们将构建一个相对完整的流程，它不仅能记住对话历史，还能根据对话内容决定是否要调用一个模拟的“知识库”来获取更精准的信息。

3.1 环境准备与State设计

首先，安装必要的库。我建议使用虚拟环境来管理依赖。

pip install langgraph langchain-openai

接下来，我们设计一个更健壮的State。除了基本的消息历史，我们增加一些字段来支持更复杂的逻辑。

from typing import TypedDict, List, Optional, Annotated from langgraph.graph.message import add_messages import operator class ChatState(TypedDict): “”“会话状态定义”“” # 核心对话历史，使用add_messages自动追加 messages: Annotated[List[dict], add_messages] # 从用户最新消息中解析出的意图 detected_intent: Optional[str] # 是否需要查询知识库 needs_knowledge_lookup: bool # 从知识库查询到的结果（如果有） knowledge_result: Optional[str] # 一个简单的对话轮次计数器，可用于触发记忆总结 turn_count: Annotated[int, operator.add] # 使用加法归约，每次+1

注意turn_count字段，我们使用了operator.add作为归约函数。这意味着每次有节点返回{“turn_count”: 1}时，这个1会被加到现有的turn_count值上，完美实现了计数功能。

3.2 创建功能节点

我们将创建四个节点，构成一个完整的工作流：意图识别节点、知识库查询节点、LLM回复生成节点和记忆管理节点。

节点1：意图识别节点这个节点负责分析用户的最新消息，判断其意图，并决定是否需要查询知识库。

def detect_intent(state: ChatState): “”“分析用户意图，判断是否需要外部知识”“” # 获取最新的用户消息 messages = state[“messages”] last_message = messages[-1][“content”] if messages else “” # 这里为了演示，使用简单的关键词匹配。在实际项目中，你可以用一个小型分类模型或更复杂的NLU逻辑。 needs_lookup = False intent = “general_chat” keywords_need_knowledge = [“什么是”， “如何”， “教程”， “解释一下”， “公司的政策”] for kw in keywords_need_knowledge: if kw in last_message: needs_lookup = True intent = “knowledge_query” break # 返回要更新的状态部分 return { “detected_intent”: intent, “needs_knowledge_lookup”: needs_lookup }

节点2：知识库查询节点这是一个模拟节点，在实际应用中，这里会连接你的数据库、向量搜索引擎或API。

def query_knowledge_base(state: ChatState): “”“模拟查询知识库”“” query = state[“messages”][-1][“content”] # 模拟根据查询返回结果 mock_knowledge = { “什么是LangGraph”: “LangGraph是用于构建有状态多智能体应用的框架。”, “如何退款”: “请登录官网，在‘我的订单’页面申请退款。”, “公司的政策”: “公司规定每周五可以远程办公。” } result = mock_knowledge.get(query, “未找到相关信息。”) return {“knowledge_result”: result}

节点3：生成回复节点这是核心节点，它综合对话历史、用户意图和知识库结果，生成最终的回复。

from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate llm = ChatOpenAI(model=“gpt-3.5-turbo”, temperature=0.7) def generate_response(state: ChatState): “”“生成AI助手的回复”“” messages = state[“messages”] knowledge = state.get(“knowledge_result”) intent = state.get(“detected_intent”) # 构建系统提示词，将知识作为上下文注入 system_prompt = “““你是一个有帮助的助手。请根据对话历史和以下知识（如果有的话）来回答用户问题。 知识上下文：{knowledge} 当前用户意图被识别为：{intent} ”””.format(knowledge=knowledge if knowledge else “无”, intent=intent) # 准备完整的消息列表：系统提示 + 历史对话 full_messages = [{“role”: “system”, “content”: system_prompt}] + messages # 调用LLM response = llm.invoke(full_messages) # 将助手的回复作为一条新消息返回，用于更新状态 return {“messages”: [{“role”: “assistant”, “content”: response.content}]}

节点4：记忆管理节点（高级功能）这是一个可选但非常重要的节点。当对话轮次太多时，我们可以触发这个节点来对历史记忆进行“总结”或“压缩”，防止上下文过长。

def manage_memory(state: ChatState): “”“每5轮对话后，尝试压缩早期记忆”“” turn_count = state.get(“turn_count”, 0) if turn_count % 5 == 0 and turn_count > 0: # 每5轮触发一次 old_messages = state[“messages”] if len(old_messages) > 10: # 如果消息太多 # 提取需要长期记住的关键信息（这里简化处理，实际可用LLM总结） # 例如，让LLM总结前10轮对话的要点 summary_prompt = f“请用一段话总结以下对话的核心内容：{str(old_messages[:10])}” # 调用LLM生成总结... # summary = llm.invoke(summary_prompt) # 模拟一个总结 summary = “[系统已将早期对话总结为：用户咨询了关于AI开发和学习路径的问题。]” # 用总结替换掉早期的详细消息，保留近期对话 new_messages = [{“role”: “system”, “content”: summary}] + old_messages[10:] return {“messages”: new_messages} # 如果不需要压缩，则返回空字典，不改变状态 return {}

3.3 组装工作流图

现在，我们把所有节点用边连接起来，形成一个有向图。

from langgraph.graph import StateGraph, START, END # 1. 创建图，并指定状态类型 workflow = StateGraph(ChatState) # 2. 添加所有节点 workflow.add_node(“detect_intent”, detect_intent) workflow.add_node(“query_knowledge”, query_knowledge_base) workflow.add_node(“generate_response”, generate_response) workflow.add_node(“manage_memory”, manage_memory) # 3. 设置入口 workflow.set_entry_point(“detect_intent”) # 4. 设置条件边：意图识别后，根据是否需要查询知识库来路由 def route_after_intent(state: ChatState): if state.get(“needs_knowledge_lookup”): return “query_knowledge” else: return “generate_response” workflow.add_conditional_edges( “detect_intent”, route_after_intent, {“query_knowledge”: “query_knowledge”, “generate_response”: “generate_response”} ) # 5. 设置普通边：知识查询后，必然去生成回复 workflow.add_edge(“query_knowledge”, “generate_response”) # 6. 设置普通边：生成回复后，进入记忆管理节点 workflow.add_edge(“generate_response”, “manage_memory”) # 7. 记忆管理节点执行完后，一轮对话结束，回到入口等待下一条用户消息。 # 这里我们让记忆管理节点连接回`detect_intent`，同时增加一个计数器更新。 # 但注意，我们需要在`manage_memory`节点里也更新`turn_count`，或者单独加一个节点。 # 更清晰的做法：在`generate_response`节点后，先更新计数器，再管理记忆。 # 让我们调整一下：增加一个专门更新计数的节点。 def increment_turn(state: ChatState): “”“增加对话轮次计数”“” return {“turn_count”: 1} workflow.add_node(“increment_turn”, increment_turn) # 调整边：生成回复 -> 增加计数 -> 管理记忆 -> 结束本轮（回到入口前实际上需要等待新输入，这由外部驱动） workflow.add_edge(“generate_response”, “increment_turn”) workflow.add_edge(“increment_turn”, “manage_memory”) workflow.add_edge(“manage_memory”, END) # 先设为END，实际流控由外部调用决定。 # 8. 编译图 app = workflow.compile()

3.4 运行与测试

图编译好后，我们就可以运行它了。LangGraph应用（app）的invoke方法需要一个初始状态。

# 初始化状态，包含第一条用户消息 initial_state: ChatState = { “messages”: [{“role”: “user”, “content”: “什么是LangGraph？”}], “detected_intent”: None, “needs_knowledge_lookup”: False, “knowledge_result”: None, “turn_count”: 0 } # 执行图 try: final_state = app.invoke(initial_state) print(“最终回复：”, final_state[“messages”][-1][“content”]) print(“当前状态：”, {k: v for k, v in final_state.items() if k != ‘messages’}) print(“消息历史长度：”, len(final_state[“messages”])) except Exception as e: print(“执行出错：”, e)

执行上述代码，你会看到工作流依次经过：检测到意图为“knowledge_query” -> 触发知识库查询 -> 结合查询结果生成回复 -> 增加计数 -> 管理记忆（可能不触发）-> 结束。最终状态里包含了完整的对话历史和更新后的各种标志。

注意事项：这里的图以END结束，意味着一次invoke只处理单轮交互。在实际的聊天服务器中（比如用FastAPI），你会维护一个持久化的app实例，并且每次用户发送新消息时，都将当前的总状态（包含所有历史）作为输入，调用app.invoke。新的用户消息会通过add_messages的归约功能自动追加到messages列表中，从而实现跨轮次的记忆。这正是LangGraph管理会话状态的精妙之处。

4. 高级记忆模式与优化策略

基础的记忆功能实现了，但要让AI真正显得“聪明”，我们还需要更高级的策略。直接塞入全部历史是最差的方法，下面介绍几种实用的记忆优化模式。

4.1 记忆窗口与滑动窗口

这是最简单有效的策略：只保留最近N轮对话。在LangGraph中，你可以在State更新时截断messages列表。

def apply_sliding_window(state: ChatState, window_size: int = 10): “”“应用滑动窗口，只保留最近N条消息”“” messages = state[“messages”] if len(messages) > window_size: # 保留最后的window_size条消息，可以根据需要保留一条系统总结在最前面 state[“messages”] = messages[-window_size:] return state

你可以将这个函数作为一个独立的节点插入到图中（例如在manage_memory节点中调用），或者在每次invoke之前作为预处理步骤。

4.2 记忆总结与压缩

当对话很长时，丢弃早期信息可能丢失重要上下文（比如用户最初设定的目标）。更好的方法是将遥远的记忆压缩成一段简短的摘要。这就是前面manage_memory节点演示的思路。更高级的实现可以：

1. 提取实体和关键事实：使用LLM或NER模型从早期对话中提取人名、地点、任务目标等关键信息，将其转化为结构化的数据存入State的独立字段（如user_profile,task_goal），而不是放在冗长的messages里。
2. 增量式总结：不要每次都从头总结所有历史。可以维护一个“记忆摘要”字段，每经过几轮对话，就让LLM基于“旧的摘要”和“新的几轮对话”生成一个“更新的摘要”。

class AdvancedState(TypedDict): messages: Annotated[List[dict], add_messages] # 近期详细对话 memory_summary: str # 长期记忆摘要 # ... 其他字段 def summarize_memory_incrementally(state: AdvancedState): “”“增量更新记忆摘要”“” new_messages = state[“messages”][-3:] # 假设只总结最新的3轮对话 old_summary = state.get(“memory_summary”, “”) prompt = f“““ 旧的记忆摘要：{old_summary} 最近发生的对话：{new_messages} 请将最近对话中的重要信息整合到记忆摘要中，生成一个更新的、简洁的摘要。只输出摘要文本。 ””” # new_summary = llm.invoke(prompt) new_summary = f“{old_summary} 最近用户询问了关于API密钥的问题。” # 模拟 return {“memory_summary”: new_summary, “messages”: []} # 清空已总结的messages

然后在生成回复时，将memory_summary和近期的messages一起作为上下文提供给LLM。

4.3 基于向量数据库的长期记忆

对于需要记忆海量、非结构化信息（如用户上传的文档、过往的邮件内容）的场景，滑动窗口和总结都力有未逮。这时，就需要引入向量数据库。

其核心思想是：

1. 将对话历史中的每一段文本（或用户提供的文档）转换成向量（Embedding）。
2. 将这些向量和原始文本一起存入向量数据库（如Chroma, Weaviate, Pinecone）。
3. 当需要回忆时，将当前用户的问题也转换成向量，在向量数据库中搜索与之最相关的几段历史文本。
4. 将这些搜索到的“相关记忆”作为上下文，注入到当前的Prompt中。

这实现了“按需记忆”：LLM不需要看到全部历史，只需要看到与当前问题最相关的片段。这在LangGraph中可以作为一个独立的“记忆检索”节点来实现。

from langchain_chroma import Chroma from langchain_openai import OpenAIEmbeddings # 初始化向量库（示例，持久化需配置persist_directory） vectorstore = Chroma(embedding_function=OpenAIEmbeddings(), collection_name=“chat_memory”) def store_memory(state: AdvancedState): “”“将上一轮对话的重要信息存入向量库”“” last_interaction = state[“messages”][-2:] # 假设存储最近一轮的问答对 if last_interaction: text_to_store = f“User: {last_interaction[0][‘content’]}\nAssistant: {last_interaction[1][‘content’]}” # 生成向量并存储，可以关联一个会话ID vectorstore.add_texts([text_to_store], metadatas=[{“session_id”: “user_123”}]) return {} # 不修改核心状态 def retrieve_memory(state: AdvancedState): “”“根据当前问题检索相关记忆”“” current_question = state[“messages”][-1][“content”] # 从向量库中搜索最相关的3条记忆 docs = vectorstore.similarity_search(current_question, k=3, filter={“session_id”: “user_123”}) retrieved_memories = “\n”.join([doc.page_content for doc in docs]) return {“retrieved_context”: retrieved_memories}

然后，在generate_response节点中，将state[“retrieved_context”]也加入到系统提示词里。这样，AI就拥有了一个庞大且精准的“长期记忆库”。

实操心得：向量检索记忆非常强大，但要注意“记忆污染”问题。如果检索到了不相关或过时的信息，反而会干扰AI的判断。因此，设计好的元数据过滤（如按时间、会话ID、主题过滤）和检索后的重排序（Rerank）机制至关重要。对于简单会话，滑动窗口+总结通常就足够了；对于知识密集型助手，向量数据库是必选项。

5. 常见问题、调试技巧与性能优化

在实际开发中，你肯定会遇到各种奇怪的问题。下面是我踩过坑后总结的一些实战经验。

5.1 状态更新不生效？

这是新手最常见的问题。请务必检查：

• 归约函数是否正确使用：对于列表追加，必须用Annotated[List, add_messages]。对于数值累加，用Annotated[int, operator.add]。如果你希望直接替换某个值，则不需要Annotated。
• 节点返回值格式：节点函数必须返回一个字典，字典的键必须是State中定义的字段名。返回{“messages”: [new_message]}，LangGraph会通过归约函数将其合并到State的messages列表里。如果你返回{“messages”: “hello”}，一个字符串，就会因为类型不匹配而出错。
• 编译后修改了State定义：如果你修改了State类的定义（比如增删字段），必须重新编译图（重新运行workflow.compile()）。否则，图会使用旧的状态结构，导致错误。

5.2 如何调试复杂的图？

当你的图有多个分支和循环时，跟踪执行流会很困难。

• 使用stream方法：app.stream(initial_state)会返回一个生成器，每执行一个节点就产出一次当前状态。你可以遍历它，打印出每个节点执行后的状态快照，清晰看到数据是如何流动和变化的。

for step, output in app.stream(initial_state): print(f“步骤后状态: {step}”) print(output)

• 可视化你的图：LangGraph提供了内置的可视化功能。app.get_graph().draw_mermaid_png()可以生成图像（需要安装pygraphviz）。虽然指令中禁止在输出里用Mermaid，但你在本地调试时这是个神器。
• 打印节点输入输出：在每个节点的函数开头和结尾添加print语句，输出接收到的state和返回的更新值。这是最直接的调试方法。

5.3 处理流式输出

用户希望看到AI一个字一个字地回复，而不是等待全部生成完。LangGraph原生支持流式输出，关键在于generate_response节点中使用支持流式的LLM（如ChatOpenAI的stream方法），并将图配置为支持流式。

from langchain_core.runnables import RunnableConfig def generate_response_streaming(state: ChatState): messages = state[“messages”] # ... 准备prompt ... # 使用stream方法 stream = llm.stream(full_messages) collected_chunks = [] for chunk in stream: content = chunk.content if content is not None: collected_chunks.append(content) # 关键：使用yield来流式返回部分结果 yield {“messages”: [{“role”: “assistant”, “content”: content}]} # 流结束后，可以做一些最终处理 full_reply = “”.join(collected_chunks) # 注意：在流式节点中，通常最终状态由框架自动处理，你可能不需要返回最终字典。 # 在图中，需要将节点标记为支持流式 workflow.add_node(“generate_response”, generate_response_streaming)

然后使用app.stream(initial_state, stream_mode=“values”)来调用，你就可以在客户端逐步收到AI的回复了。这对提升用户体验至关重要。

5.4 性能与成本优化

• Token管理：记忆是最大的Token消耗源。务必实施记忆压缩策略（总结、滑动窗口）。在调用LLM前，可以计算一下messages的大致Token数（使用tiktoken等库），如果超过阈值，则触发记忆总结节点。
• 异步执行：如果图中有多个可以并行执行的节点（比如同时查询两个不同的数据库），LangGraph支持异步节点。使用async def定义节点函数，并在添加节点时注明，可以显著提升吞吐量。
• 缓存：对于纯函数式、输入相同的节点（如某些数据转换节点），可以考虑使用functools.lru_cache进行缓存，避免重复计算。但注意，对于调用LLM或查询可变数据库的节点，不能缓存。

5.5 与FastAPI等Web框架集成

在生产环境中，你的LangGraph应用通常作为一个后台服务。集成模式如下：

from fastapi import FastAPI, HTTPException from pydantic import BaseModel import asyncio app_fastapi = FastAPI() # 假设你的LangGraph应用实例为 `agent_app` class ChatRequest(BaseModel): message: str session_id: str # 内存中存储不同会话的状态（生产环境应使用Redis等） session_states = {} @app_fastapi.post(“/chat”) async def chat_endpoint(request: ChatRequest): session_id = request.session_id # 获取或初始化该会话的状态 current_state = session_states.get(session_id) if current_state is None: current_state = {“messages”: [], “turn_count”: 0, “session_id”: session_id} # 确保状态符合你的State定义 # 这里需要将字典转换为你的State类型，可能需要一些处理 # 将用户新消息添加到状态中 # 注意：这里模拟了add_messages的归约。更严谨的做法是调用一个专门添加消息的节点。 new_user_message = {“role”: “user”, “content”: request.message} current_state[“messages”] = current_state.get(“messages”, []) + [new_user_message] try: # 调用LangGraph应用处理当前状态 final_state = await agent_app.ainvoke(current_state) # 更新会话状态 session_states[session_id] = final_state # 提取最新的助手回复 last_message = final_state[“messages”][-1] if last_message[“role”] != “assistant”: raise HTTPException(status_code=500, detail=“No assistant response generated”) return {“reply”: last_message[“content”], “session_id”: session_id} except Exception as e: raise HTTPException(status_code=500, detail=str(e))

这个模式清晰地将LangGraph作为对话引擎，Web框架处理HTTP请求、会话管理和状态持久化。