综合案例 - AI 智能租房助手 [ 5 ]
主图 -- 智能分流
到这里，我们前期准备的所有子图就全部开发完成了。也就是说，本案例中设计的核心功能点都已经逐一实现。
首先是房源推荐功能，我们依靠推荐子图完成开发；其次是房源预定功能，由预定子图实现；还有常规问答功能，也通过对应的问答子图落地。目前只剩下查询个人信息[get_user_preferences]这一个功能还没有开发，这个功能我们会直接集成到主图当中。
虽说各个细分功能对应的子图都已经编写完毕，但想要实现流畅连贯的对话交互，还需要依靠主图把所有子功能串联起来。因此主图主要承担两项核心工作：
第一，将所有子图整合衔接；
第二，单独实现查询个人信息的功能。
结合之前梳理的整体流程，接下来我们就正式编写主图相关代码。和开发子图的逻辑一致，我们依旧先从状态定义入手，只有明确主图需要用到的状态字段，后续才能顺利编写各个功能节点。
我们先梳理整体执行流程，按照以往的开发思路一步步分析。主图的第一个节点，作用是获取用户偏好信息。在之前开发推荐子图时大家也了解过，如果用户提问时没有附带预算等偏好数据，系统就会调取历史留存的用户偏好作为参考。所以在流程入口处单独设置节点拉取用户偏好，能够为后续所有节点提供数据支撑。
当然，也可以不在入口统一获取，等到其他节点需要使用时再单独查询，两种方式都可行。这里我们选择单独抽离出一个节点统一获取，这个节点最终会返回用户的偏好数据。
状态定义
基于这个逻辑，我们先来定义主图的全局状态。首先必不可少的就是消息列表messages，这是对话类应用的基础状态。其次就是用户偏好数据，该数据存入全局state后，主图和各个子图之间可以实现数据共享，消息列表同样也支持全局共享。
流程走到第二个节点，该节点的核心作用是识别用户意图，同时完成智能路由。系统会根据识别出的用户意图，判断对话流程该走向哪一个后续节点：如果识别出用户意图是房源推荐，流程就进入推荐子图；如果是房源预定，则进入预定子图；若是普通咨询类问题，就流转到扩展子图；如果是查询个人信息，就进入查询历史信息的对应节点。
用户意图是路由判断的核心依据，所以我们需要把识别出的意图信息也存入全局状态。后续配置条件分支时，程序就能读取state中的用户意图字段，自动判断下一步的执行路径。
再看其余节点和子图的输出内容，预定子图、查询个人信息节点、扩展子图，最终输出的都是AI消息，也就是展示给用户的最终回复内容。推荐子图执行完成后，同样会返回AI消息，向用户展示匹配的房源。
不过推荐流程之后还有一个特殊环节：系统推荐完房源，会主动向用户询问是否需要预定。我们会根据用户的回复决定后续走向：用户回复 “需要”，流程就跳转至预定子图；回复 “不需要”，对话直接结束。
为了实现这个交互逻辑，我们需要新增一个节点，并且在这个节点中使用中断机制，由用户的输入来掌控后续流程。用户输入的 “需要” 或 “不需要” 是字符串类型，这个结果也需要临时存储，作为后续条件分支的判断依据。
这里区分两种状态：
第一种是全局共享状态，比如消息列表、用户偏好、用户意图，主图和所有子图都可以读取使用；
第二种是临时私有状态，也就是存储用户预定选择的字段。这个字段仅作用于推荐流程之后的条件分支，其他子图和节点都不会使用，因此无需全局共享。
简单说明一下状态共享的规则：主图中定义的全局状态，所有子图理论上都能获取，但子图可以自主选择是否接收。如果子图代码中没有声明对应状态参数，就不会读取主图的共享数据；只有主动声明参数，才会完成数据同步。而我们刚刚提到的 “是否预定” 相关数据，仅在局部流程中临时使用，因此将它定义为私有状态即可，该状态不会出现在整个流程图的最终输出结果里，只在流程执行的中间环节生效。
下面代码实现：
主状态State：全局共享状态。状态继承自MessagesState，自动管理对话历史。
user_intent：用户意图，表示用户输入的问题含义（推荐 or 预定 or 查询 or 其它）user_preferences：用户偏好信息，实现数据共享
节点间传输私有状态NeedReserveOutput：当推荐子图执行完成后，用户获取用户是否想要预定房源的意向。从而决定是否执行预定子图。
state/main.py
from typing import TypedDict from langgraph.graph import MessagesState class State(MessagesState): user_intent: str # 用户意图 user_preferences: dict # 用户偏好 class NeedReserveOutput(TypedDict): reserve: str # 这个字段不会出现在最终状态中节点实现
梳理完所有状态之后，我们就对照整体流程图，逐行编写对应代码。
首先实现第一个节点：查询持久化存储中的用户信息[get_store_info]。编写节点函数时，入参包含主图全局状态、上下文信息runtime以及存储对象store，大家要注意store的导入路径，避免引用错误。
具体逻辑为：先从上下文runtime中取出用户ID，用用户ID构建命名空间namespace，再通过store根据命名空间检索对应的用户数据，检索结果会以列表形式返回。之后做逻辑判断：如果检索到有效数据，就将用户偏好数据返回并存入全局状态；如果检索结果为空，则返回空字典。这个节点逻辑比较简单，核心就是拉取持久化的用户偏好信息。
# 节点：查询持久化信息 def get_store_info(state: State, runtime: Runtime[ContextSchema], * , store: BaseStore): # 搜索用户信息 user_id = runtime.context.get("user_id") namespace = (user_id, "preferences") prefs_result = store.search(namespace) if prefs_result and prefs_result[0]: return { "user_preferences": prefs_result[0].value } else: return { "user_preferences": {} }接下来实现第二个节点：识别用户意图[identify_question]。这里我们借助大模型的结构化输出能力来实现。全局状态中存有完整的对话消息列表，我们提取其中的用户提问内容，交给大模型进行解析。
首先定义结构化数据模型，规定用户意图仅分为四类：房源推荐、房源预定、查询个人信息、其他内容，大模型最终只能在这四类结果中返回其一。同时配置对应的系统提示词，明确要求模型严格根据用户语义判断意图，不允许猜测、编造信息。
组装系统提示词和用户对话消息，调用大模型并绑定好结构化输出规则，模型解析完成后，我们提取出最终的意图结果，存入全局状态。实际开发中，我们也可以只截取最新一条用户消息进行解析，无需加载全部历史对话，两种方式都能实现意图识别的效果。
class UserMessage(BaseModel): type: Literal["recommend_house", "reserve_house", "get_info", "others"] = Field( description="根据用户问题描述判断问题类型：推荐房源、预定房源、获取信息、其它内容" ) # 节点：识别用户意图 def identify_question(state: State): # state["messages"] # 用户问题 -》 LLM -> 结构化输出（type） : 推荐、预定、我的、其它 user_intent = model.with_structured_output(UserMessage).invoke( [SystemMessage(content="你是一个根据描述提取信息的提取专家。请从用户的描述中提取想要咨询的相关信息。" "严谨根据语义推断信息，但是不能猜测或者编造信息。"), state["messages"][-1]] ) return { "user_intent": user_intent.type # 条件边使用 }到这里，意图识别节点就开发完成了。目前推荐子图、预定子图、扩展子图、意图识别节点、拉取用户偏好节点都已编写完毕，还剩余两个节点需要实现：一是搭配中断机制、询问用户是否预定房源的节点；二是查询并返回用户历史偏好信息的节点。
我们先来开发带中断机制的节点[need_reserve]。这个节点的作用是：系统完成房源推荐后，主动弹窗询问用户是否需要预定房源。该节点的入参依旧是主图全局状态，返回结果则是我们之前定义的私有状态，专门用来存放用户的回复内容。
在节点内部编写交互提示语，明确告知用户输入规则：输入 “需要” 则进入预定流程，输入 “不需要” 则结束对话，其余输入内容均无效。接着调用中断方法，将提示语展示给用户，同时接收用户的输入内容，最后把内容存入私有状态。这个节点的核心就是利用中断功能，实现人机交互并捕获用户选择。
# 节点：中断询问是否需要帮助预定房源 def need_reserve(state: State) -> NeedReserveOutput: prompt = f"已经为您推荐合适的房源，是否需要帮您预订房源？\n" prompt += "如果不需要,请输入'**不需要**'。\n" prompt += "如果需要,请输入'**需要**'。\n(注意输入其它值无效)\n" answer = interrupt(prompt) return {"reserve": answer} # 条件边获取到后，是否执行预定子图最后我们来开发查询并返回用户偏好信息的节点[get_user_preferences]。
第一步，从全局状态中读取此前拉取到的用户偏好数据，同时做兼容处理，若数据为空则默认赋值为空字典。
第二步，筛选对话消息列表，单独提取出所有用户发出的消息。因为对话支持持久化，一轮会话中会累积多条历史消息，筛选出用户消息后，我们就能精准拿到用户本次的提问内容。
第三步，对原始偏好数据做格式转换。用户偏好数据里，已预定房源信息是以列表形式存储的，直接交给大模型会影响展示效果。因此我们手动遍历列表，把每一条预定记录整理成通顺的文本格式，拼接成完整字符串；如果用户没有任何预定记录，则统一赋值为 “无”。
第四步，组装对话消息并调用大模型。首先设置系统提示词，要求模型结合用户偏好信息回复问题，数据为空时不得编造内容，同时回复风格要自然生动，不要生硬罗列数据。随后把整理好的预算、历史预定记录等偏好信息、用户本次提问依次组装成消息列表，调用大模型生成回复。模型返回的
AI消息，就是最终展示给用户的结果。
# 节点：返回用户偏好信息 def get_user_preferences(state: State): # 获取最新历史偏好信息（参考答案） prefs = state.get("user_preferences", {}) # 筛选用户消息（获取到用户问题） user_messages = filter_messages(state["messages"], include_types="human") reserved_info = prefs.get("reserved_info", []) if reserved_info: # 有预定过的信息 reserved_str = "\n" for i, item in enumerate(reserved_info, 1): reserved_str += f"{i}. 预定工单ID: {item.get('order_id')}，" \ f"房源标题：{item.get('title')}，" \ f"预定电话：{item.get('phone_number')}\n" else: # 没有预定 reserved_str = "无" result = model.invoke( [SystemMessage(content="""你是一个乐于助人的助手，可以根据用户偏好信息进行回复。 如果有的偏好数据为空，不要猜测或编造数据。 不要直接回复偏好数据是什么，要结合问题进行生动回复。 如果问题与用户偏好数据无关，直接回复即可。""") , HumanMessage(content="用户的历史偏好信息如下" f"1. 最低预算：{prefs.get('budget_min')}" f"2. 最高预算：{prefs.get('budget_max')}" f"3. 已预定过的信息：{reserved_str}" ) , user_messages[-1] # 问题 ] ) return { "messages": [result] }至此，主图的所有功能节点就全部开发完成了。
整个流程里，意图识别节点配合全局状态实现了智能路由，根据用户意图将流程分发到不同子图或查询节点；推荐流程后的中断节点，搭配私有状态和条件分支，实现了预定环节的二次交互判断。整个主图一共配置了两处条件分支，分别对应不同的路由逻辑。
所有节点开发完毕后，下一步我们就开始搭建完整的流程图并进行测试。本次测试会覆盖案例中全部功能，既可以在前端页面中直观测试交互效果，也能通过接口调用的方式，验证后端逻辑的运行状态。
node/main.py【整合代码】
from langchain_core.messages import SystemMessage, filter_messages, HumanMessage from langgraph.runtime import Runtime from langgraph.store.base import BaseStore from langgraph.types import interrupt from pydantic import BaseModel, Field from typing_extensions import Literal from src.agent.common.context import ContextSchema from src.agent.common.llm import model from src.agent.state.main import State, NeedReserveOutput # 节点：查询持久化信息 def get_store_info(state: State, runtime: Runtime[ContextSchema], * , store: BaseStore): # 搜索用户信息 user_id = runtime.context.get("user_id") namespace = (user_id, "preferences") prefs_result = store.search(namespace) if prefs_result and prefs_result[0]: return { "user_preferences": prefs_result[0].value } else: return { "user_preferences": {} } class UserMessage(BaseModel): type: Literal["recommend_house", "reserve_house", "get_info", "others"] = Field( description="根据用户问题描述判断问题类型：推荐房源、预定房源、获取信息、其它内容" ) # 节点：识别用户意图 def identify_question(state: State): # state["messages"] # 用户问题 -》 LLM -> 结构化输出（type） : 推荐、预定、我的、其它 user_intent = model.with_structured_output(UserMessage).invoke( [SystemMessage(content="你是一个根据描述提取信息的提取专家。请从用户的描述中提取想要咨询的相关信息。" "严谨根据语义推断信息，但是不能猜测或者编造信息。"), state["messages"][-1]] ) return { "user_intent": user_intent.type # 条件边使用 } # 节点：中断询问是否需要帮助预定房源 def need_reserve(state: State) -> NeedReserveOutput: prompt = f"已经为您推荐合适的房源，是否需要帮您预订房源？\n" prompt += "如果不需要,请输入'**不需要**'。\n" prompt += "如果需要,请输入'**需要**'。\n(注意输入其它值无效)\n" answer = interrupt(prompt) return {"reserve": answer} # 条件边获取到后，是否执行预定子图 # 节点：返回用户偏好信息 def get_user_preferences(state: State): # 获取最新历史偏好信息（参考答案） prefs = state.get("user_preferences", {}) # 筛选用户消息（获取到用户问题） user_messages = filter_messages(state["messages"], include_types="human") reserved_info = prefs.get("reserved_info", []) if reserved_info: # 有预定过的信息 reserved_str = "\n" for i, item in enumerate(reserved_info, 1): reserved_str += f"{i}. 预定工单ID: {item.get('order_id')}，" \ f"房源标题：{item.get('title')}，" \ f"预定电话：{item.get('phone_number')}\n" else: # 没有预定 reserved_str = "无" result = model.invoke( [SystemMessage(content="""你是一个乐于助人的助手，可以根据用户偏好信息进行回复。 如果有的偏好数据为空，不要猜测或编造数据。 不要直接回复偏好数据是什么，要结合问题进行生动回复。 如果问题与用户偏好数据无关，直接回复即可。""") , HumanMessage(content="用户的历史偏好信息如下" f"1. 最低预算：{prefs.get('budget_min')}" f"2. 最高预算：{prefs.get('budget_max')}" f"3. 已预定过的信息：{reserved_str}" ) , user_messages[-1] # 问题 ] ) return { "messages": [result] }工作流定义
接下来我们需要完成主图的搭建工作。我们把主图相关代码统一放在agrnt/graph.py文件中，这个文件里原本有框架自带的模板代码，我们先将原有模板内容全部删除，从零开始编写属于本项目的主图逻辑。
首先在文件中导入主图所需的状态类，同时引入ContextSchema上下文结构。完成基础导入后，我们开始逐个添加流程节点与节点之间的连线，整体按照前期梳理好的流程图来实现。
from typing import Literal from langgraph.constants import START, END from langgraph.graph import StateGraph from src.agent.extend import extend_graph from src.agent.recommend import recommended_graph from src.agent.reserve import reserve_graph from src.agent.common.context import ContextSchema from src.agent.node.main import get_store_info, identify_question, get_user_preferences, need_reserve from src.agent.state.main import State, NeedReserveOutput第一步先完成节点的添加。整个主图一共包含七个节点，我们依次进行配置：第一个是获取用户偏好数据节点；第二个是识别用户意图节点；接着依次引入各个子图，包括房源推荐子图、房源预定子图、常规问答扩展子图，然后是查询用户偏好信息节点，最后是询问是否预定房源的中断节点。添加子图时需要注意命名规范，保证节点名称和流程图保持一致，推荐子图命名为recommend_graph，预定子图命名为reserve_graph，避免名称冲突。
所有节点添加完毕后，开始配置节点之间的连线。首先设置固定连线：流程起点START直接连接到获取用户偏好数据节点，这是整个工作流的入口。执行完用户意图识别节点后，就需要依靠条件分支实现智能路由，因此我们在这里配置条件边。
builder = StateGraph(State, context_schema=ContextSchema) builder.add_node(get_store_info) builder.add_node(identify_question) builder.add_node("recommended_graph", recommended_graph) builder.add_node("reserve_graph", reserve_graph) builder.add_node("extend_graph", extend_graph) builder.add_node(get_user_preferences) builder.add_node(need_reserve) builder.add_edge(START, "get_store_info") builder.add_edge("get_store_info", "identify_question")我们先编写路由函数router_message，函数入参为主图全局状态state，作用是读取状态中的用户意图字段，根据不同的意图返回对应下游节点的名称。用户意图是在上一个意图识别节点中，通过大模型结构化输出得到的，结果只会是预设的四类内容：房源推荐、房源预定、查询个人信息、其他问题。
函数内部做逻辑判断：如果用户意图为recommend_house，就路由到推荐子图；如果是reserve_house，则路由到预定子图；若为get_info，进入查询用户偏好信息节点；其余情况全部流转到扩展子图。这里要说明，我们是以子图作为普通节点添加进主图的，这种方式才能保证主图和子图之间正常共享状态数据。
# 智能路由 def router_message(state: State) -> Literal["recommended_graph", "reserve_graph", "extend_graph", "get_user_preferences"]: user_intent = state["user_intent"] if user_intent == "recommend_house": return "recommended_graph" elif user_intent == "reserve_house": return "reserve_graph" elif user_intent == "get_info": return "get_user_preferences" else: return "extend_graph"路由逻辑编写完成后，为条件边配置路由映射关系，绑定好所有可能跳转的目标节点。接下来继续梳理剩余的连线规则，整个流程一共分为四组路由逻辑。
builder.add_conditional_edges( "identify_question", router_message, ["recommended_graph", "reserve_graph", "extend_graph", "get_user_preferences"] )第一组路由：推荐子图执行完成后，不会直接结束流程，而是固定连接到询问是否预定房源的中断节点。随后从该中断节点再引出一组条件边，读取私有状态中用户的回复内容做判断：如果用户回复 “需要”，就跳转到预定子图；如果回复 “不需要”，流程直接走到终点END。这部分是整个流程里最特殊的逻辑，依靠固定边搭配条件边，实现人机二次交互。
第二组路由：预定子图执行完毕后，直接连接流程终点。 第三组路由：查询用户偏好信息节点执行完成后，同样直接走到终点。 第四组路由：扩展子图执行结束，也直接终止整个流程。
# 路由1：推荐子图：根据用户中断信息决定后续是否继续预定 builder.add_edge("recommended_graph", "need_reserve") def should_reserve(state: NeedReserveOutput): reserve = state["reserve"] if reserve == "需要": return "reserve_graph" else: return END builder.add_conditional_edges( "need_reserve", should_reserve, ["reserve_graph", END] ) # 路由2：预定子图 builder.add_edge("reserve_graph", END) # 路由3：查询我的 builder.add_edge("get_user_preferences", END) # 路由4：其它 builder.add_edge("extend_graph", END) graph = builder.compile() # print(graph.get_graph(xray=True).draw_mermaid())所有连线和分支逻辑配置完成后，调用builder.compile()方法编译流程图，生成最终可运行的主图实例。我们还可以执行代码打印流程图结构，验证节点、连线、分支是否和设计图一致。
执行代码后程序如果出现了循环依赖的报错。问题可能出现在导入方式上：我们从初始化文件中引入子图，而初始化文件又会反向引用当前的
graph.py，从而形成循环引用。解决办法很简单，修改导入路径，不再从初始化文件引入，直接从各个子图对应的源文件中单独导入，修改后重新运行，流程图就能正常打印出来。
代码编写完成，接下来进入整体测试阶段，我们通过终端结合可视化平台来运行主图，测试全流程功能。首先清理代码中多余的打印语句，避免日志冗余，随后启动项目。启动后我们修改项目标识名称，方便区分项目，接着正式开始功能测试。
我们新建一条会话，指定用户 ID 为 456，在输入框中提交测试语句：“在西安，我的预算是 1000~2000 元一个月，请帮我推荐 4 套雁塔区的房子”。
流程开始正常运转：首先执行获取用户偏好数据节点，接着识别用户意图，判定为房源推荐，随即进入推荐子图。子图按照规则筛选房源，最终成功推荐出 4 套符合要求的房源。推荐完成后，流程回到主图，进入中断节点，页面弹出交互提示，询问用户是否需要预定房源。我们输入 “需要” 并提交，流程自动跳转到预定子图。
按照预定流程的要求，依次填写房源名称、联系电话、身份证号等信息并提交，预定子图执行全部逻辑后，返回预定成功的提示，随后流程直接走向终点。查看持久化存储可以发现，用户 ID 为 456 的账号下，已经成功保存了本次填写的预算范围以及预定订单信息，第一条正向业务流程测试顺利完成。
完成房源推荐 + 预定的流程测试后，我们继续测试查询历史订单功能。新建一条会话，输入内容：“查询一下我的历史订单记录”。程序正常运行，页面完整展示出此前的房源标题、工单号、预订电话等历史记录，查询功能调试完成。
接下来测试常规问答功能，新建会话后先后输入 “你好”、“讲一个笑话”。系统识别出这类内容不属于推荐、预定、查信息三类意图，统一路由到扩展子图，子图正常做出对应回复，常规问答功能也运行正常。
至此，案例中四大核心功能：房源推荐、房源预定、查询个人信息、常规问答，全部完成基础测试。我们再针对边界场景进行补充测试，模拟用户只给出上限预算的场景，输入：“帮我租房子，预算 5000 以内”。
其实流程正常进入推荐子图后发现新问题：代码只提取到了最高预算 5000，最低预算字段为空，系统反复向用户询问预算信息。这是一处代码 Bug，我们分析问题根源：原有逻辑对预算空值、数值 0 做了同等判断，当用户只填写 “5000 以内” 时，系统默认最低预算为 0，却因为判断逻辑将 0 判定为无效值，导致无法正常更新状态。
针对这个问题我们调整判断规则：区分 “字段为空” 和 “数值为 0” 两种情况，只要字段不为空，无论数值是否为 0，都允许更新到状态和持久化存储中。修改代码后重新测试，再次输入 “预算 3000 以内”，系统自动将预算范围识别为 0~3000，不再重复询问预算，并且成功推荐房源。后续继续完成预定操作，查看持久化数据，最低预算也成功更新为 0，边界场景的 Bug 修复完毕。
经过多轮功能测试、问题排查与代码调试，整套基于 LangGraph 的后端工作流已经全部开发并验证完毕。目前我们都是通过可视化平台来测试流程，而在实际项目部署中，后端需要对外提供接口，供前端页面调用。
我们前期已经演示过接口调用的基础方式，后续可以基于接口文档，使用接口测试工具模拟前端请求，完整复现上述所有业务流程，验证接口的可用性。
还有需要注意，编译 graph 时，无需设置 checkpointer 和 store。将来进行 LangSmith 部署时，持久化可以通过配置进行设置。
API 测试
到这里，我们整套 LangGraph 项目里的主图、各类子图的功能代码就全部编写完成，并且经过多轮调试与功能测试了。
我们最终要交付的是一套包含前端页面的完整租房智能助手综合案例。前端页面想要实现对话交互，就必须调用后端服务，所以后端需要对外提供接口，供前端请求调用。接下来我们结合项目页面的实际需求，分析后端需要设计哪些接口。
整个项目的静态页面部分只需要编写固定布局和按钮，开发难度较低，核心功能集中在聊天对话模块。在聊天区域，用户输入内容并点击发送按钮，本质上就是触发后端的工作流，完整执行一遍我们设计好的 LangGraph 主图，并将执行结果返回给前端展示。基于这个逻辑，我们首先需要设计执行工作流的接口。
另外，LangGraph 的执行依托于会话（线程），用户每打开一次对话窗口，系统都要先创建一个独立的会话线程，后续所有的对话交互、工作流执行，都必须在这个线程的基础上完成。因此第二个必备接口就是创建会话线程接口。
同时，前端展示的对话效果要求文字逐字输出，也就是流式输出。我们在调用执行工作流的接口时，需要配置对应的运行模式。由于最终返回给用户的回复都封装为AI Message类型，所以执行模式中必须指定message模式，以此实现 AI 回复内容的流式推送。
除了正常对话之外，项目中还用到了中断交互逻辑。当工作流触发中断时，后端会把中断提示推送给前端，等待用户输入回应内容；用户填写信息并再次点击发送，就相当于恢复被暂停的工作流。所以执行工作流的接口其实承担了两类作用：一是发起全新的对话、正常运行工作流；二是接收用户输入，恢复中断的工作流。总结下来，整个项目只需要依托创建线程、执行 / 恢复工作流这两个核心接口，就能实现全部对话功能。
这两个接口我们之前也做过基础测试：创建线程使用对应的接口方法，执行工作流依靠thread runs相关接口；如果需要流式效果，则启用stream流式运行方式。而恢复中断的工作流，需要在请求中携带专属的command指令。接下来我们借助 API Fox 工具，一步步模拟前端调用流程，完整验证接口功能。
首先打开 API Fox，新建请求来测试创建线程接口。该接口的请求方式为POST，请求头中配置Content-Type为application/json，请求体不需要额外参数，直接使用本地服务的 IP + 端口发起请求。此时项目服务处于正常运行状态，发送请求后，接口成功返回了会话线程 ID，代表线程创建成功。
拿到线程 ID 后，继续测试执行工作流接口。同样使用POST请求，修改请求路径，并把上一步获取的线程 ID 替换到请求地址中。请求头依旧保持 JSON 格式，接下来重点配置请求体参数：
- 第一，填写
assistant ID，该参数和我们代码中定义的 Agent 名称保持一致，本项目中填写对应租房助手的标识； - 第二，配置
input参数，它本质是向工作流状态传参，我们只需要传入对话消息。消息遵循固定格式，外层为字典，message字段是列表，列表中定义消息类型为用户消息，并填写消息内容； - 第三，配置
context上下文参数，在其中传入自定义的user ID，用来区分不同用户； - 第四，配置流式相关模式：添加
update模式，可以查看每一个节点执行后的状态更新；添加message模式，实现 AI 回复内容流式输出；开启stream_subgraph参数，让子图的执行过程也支持流式推送。
{ "assistant_id": "house_agent", "input": { "mesages": [{"role": "human", "content": "帮我推荐几套房子"}] }, "context": { "user_id": "789" }, "stream_mode": [ "updates", "messages" ], "stream_subgraphs": true }接口成功返回 SSE 格式的流式响应，我们逐一解析响应中的各类事件：message metadata事件：代表即将开始流式输出 AI 消息，相当于前置标记；message part事件：是流式输出的核心，会将 AI 回复内容拆分为一个个字符（Token）分段推送，对应前端逐字展示的效果；update事件：代表某个节点执行完毕，同步推送该节点更新后的状态数据，比如识别用户意图、更新用户偏好等；interrupt事件：表示工作流触发中断，后端推送中断提示，等待前端传回用户的回应内容。
本次测试我们输入 “帮我推荐几套房子”，系统识别后发现缺少城市、预算等关键信息，于是触发中断并返回提示。结合返回的update事件也能看到，当前用户 ID 对应的偏好信息为空，和预期逻辑完全相符。
接下来模拟恢复中断工作流的操作。新建请求，沿用之前的接口地址、请求头、assistant ID、上下文、运行模式等配置，核心改动在请求体中：新增command参数，这是恢复中断的关键指令，参数值填写我们补充的信息，例如城市为西安、预算 2000-3000 元 / 月。配置完成后发送请求，工作流从中断位置继续执行。
接口持续推送流式数据，先是执行各个节点并同步状态更新，随后再次触发message相关事件，流式输出房源推荐结果，前端页面就会呈现出文字逐字弹出的效果。房源推荐完成后，工作流再次触发中断，询问用户是否需要预定房源。我们继续在command中传入 “需要” 来恢复流程，工作流进入预定子图。
预定子图会多次触发中断，依次要求填写房源名称、联系电话、身份证号。我们每次都通过在command中传入对应内容、调用接口恢复流程的方式，一步步走完预定流程。全部信息提交完成后，工作流执行完毕，最终的update事件会返回完整的状态数据，在消息列表中可以看到 “预定成功” 的最终回复，整个业务流程闭环完成。
我们再额外测试查询历史订单的场景。新建会话线程，在请求消息中输入 “查询我的历史订单记录”，工作流识别用户意图后路由到查询节点，直接流式输出用户过往的预定信息，执行完成后流式连接自动断开，功能运行正常。
到这里，后端对外提供的两个核心接口就全部测试完毕。对于前端开发而言，只需要对接这两个接口：解析接口返回的 SSE 流式事件，区分message、update、interrupt等不同事件类型，提取其中的文本内容，再按照交互逻辑展示在页面上即可。
目前，后端代码编写、功能调试、接口联调模拟等所有本地准备工作都已经完成。接下来我们进入项目部署环节，本次采用 Docker 容器化的方式，将整套服务部署到云服务器上。LangGraph 官方也提供了成熟的部署方案，部署完成后，接口地址就会替换为云服务器的公网 IP + 端口，前端页面正式调用线上接口，最终完成整个租房智能助手项目的交付。