![综合案例 - AI 智能租房助手 [ 5 ]](images/news1.jpg)
主图 -- 智能分流到这里我们前期准备的所有子图就全部开发完成了。也就是说本案例中设计的核心功能点都已经逐一实现。首先是房源推荐功能我们依靠推荐子图完成开发其次是房源预定功能由预定子图实现还有常规问答功能也通过对应的问答子图落地。目前只剩下查询个人信息[get_user_preferences]这一个功能还没有开发这个功能我们会直接集成到主图当中。虽说各个细分功能对应的子图都已经编写完毕但想要实现流畅连贯的对话交互还需要依靠主图把所有子功能串联起来。因此主图主要承担两项核心工作第一将所有子图整合衔接第二单独实现查询个人信息的功能。结合之前梳理的整体流程接下来我们就正式编写主图相关代码。和开发子图的逻辑一致我们依旧先从状态定义入手只有明确主图需要用到的状态字段后续才能顺利编写各个功能节点。我们先梳理整体执行流程按照以往的开发思路一步步分析。主图的第一个节点作用是获取用户偏好信息。在之前开发推荐子图时大家也了解过如果用户提问时没有附带预算等偏好数据系统就会调取历史留存的用户偏好作为参考。所以在流程入口处单独设置节点拉取用户偏好能够为后续所有节点提供数据支撑。当然也可以不在入口统一获取等到其他节点需要使用时再单独查询两种方式都可行。这里我们选择单独抽离出一个节点统一获取这个节点最终会返回用户的偏好数据。状态定义基于这个逻辑我们先来定义主图的全局状态。首先必不可少的就是消息列表messages这是对话类应用的基础状态。其次就是用户偏好数据该数据存入全局state后主图和各个子图之间可以实现数据共享消息列表同样也支持全局共享。流程走到第二个节点该节点的核心作用是识别用户意图同时完成智能路由。系统会根据识别出的用户意图判断对话流程该走向哪一个后续节点如果识别出用户意图是房源推荐流程就进入推荐子图如果是房源预定则进入预定子图若是普通咨询类问题就流转到扩展子图如果是查询个人信息就进入查询历史信息的对应节点。用户意图是路由判断的核心依据所以我们需要把识别出的意图信息也存入全局状态。后续配置条件分支时程序就能读取state中的用户意图字段自动判断下一步的执行路径。再看其余节点和子图的输出内容预定子图、查询个人信息节点、扩展子图最终输出的都是AI消息也就是展示给用户的最终回复内容。推荐子图执行完成后同样会返回AI消息向用户展示匹配的房源。不过推荐流程之后还有一个特殊环节系统推荐完房源会主动向用户询问是否需要预定。我们会根据用户的回复决定后续走向用户回复 “需要”流程就跳转至预定子图回复 “不需要”对话直接结束。为了实现这个交互逻辑我们需要新增一个节点并且在这个节点中使用中断机制由用户的输入来掌控后续流程。用户输入的 “需要” 或 “不需要” 是字符串类型这个结果也需要临时存储作为后续条件分支的判断依据。这里区分两种状态第一种是全局共享状态比如消息列表、用户偏好、用户意图主图和所有子图都可以读取使用第二种是临时私有状态也就是存储用户预定选择的字段。这个字段仅作用于推荐流程之后的条件分支其他子图和节点都不会使用因此无需全局共享。简单说明一下状态共享的规则主图中定义的全局状态所有子图理论上都能获取但子图可以自主选择是否接收。如果子图代码中没有声明对应状态参数就不会读取主图的共享数据只有主动声明参数才会完成数据同步。而我们刚刚提到的 “是否预定” 相关数据仅在局部流程中临时使用因此将它定义为私有状态即可该状态不会出现在整个流程图的最终输出结果里只在流程执行的中间环节生效。下面代码实现主状态State全局共享状态。状态继承自MessagesState自动管理对话历史。user_intent用户意图表示用户输入的问题含义推荐 or 预定 or 查询 or 其它user_preferences用户偏好信息实现数据共享节点间传输私有状态NeedReserveOutput当推荐子图执行完成后用户获取用户是否想要预定房源的意向。从而决定是否执行预定子图。state/main.pyfrom 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_typeshuman) 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用户的历史偏好信息如下 f1. 最低预算{prefs.get(budget_min)} f2. 最高预算{prefs.get(budget_max)} f3. 已预定过的信息{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_typeshuman) 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用户的历史偏好信息如下 f1. 最低预算{prefs.get(budget_min)} f2. 最高预算{prefs.get(budget_max)} f3. 已预定过的信息{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_schemaContextSchema) 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(xrayTrue).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 端口前端页面正式调用线上接口最终完成整个租房智能助手项目的交付。
综合案例 - AI 智能租房助手 [ 5 ]
主图 -- 智能分流到这里我们前期准备的所有子图就全部开发完成了。也就是说本案例中设计的核心功能点都已经逐一实现。首先是房源推荐功能我们依靠推荐子图完成开发其次是房源预定功能由预定子图实现还有常规问答功能也通过对应的问答子图落地。目前只剩下查询个人信息[get_user_preferences]这一个功能还没有开发这个功能我们会直接集成到主图当中。虽说各个细分功能对应的子图都已经编写完毕但想要实现流畅连贯的对话交互还需要依靠主图把所有子功能串联起来。因此主图主要承担两项核心工作第一将所有子图整合衔接第二单独实现查询个人信息的功能。结合之前梳理的整体流程接下来我们就正式编写主图相关代码。和开发子图的逻辑一致我们依旧先从状态定义入手只有明确主图需要用到的状态字段后续才能顺利编写各个功能节点。我们先梳理整体执行流程按照以往的开发思路一步步分析。主图的第一个节点作用是获取用户偏好信息。在之前开发推荐子图时大家也了解过如果用户提问时没有附带预算等偏好数据系统就会调取历史留存的用户偏好作为参考。所以在流程入口处单独设置节点拉取用户偏好能够为后续所有节点提供数据支撑。当然也可以不在入口统一获取等到其他节点需要使用时再单独查询两种方式都可行。这里我们选择单独抽离出一个节点统一获取这个节点最终会返回用户的偏好数据。状态定义基于这个逻辑我们先来定义主图的全局状态。首先必不可少的就是消息列表messages这是对话类应用的基础状态。其次就是用户偏好数据该数据存入全局state后主图和各个子图之间可以实现数据共享消息列表同样也支持全局共享。流程走到第二个节点该节点的核心作用是识别用户意图同时完成智能路由。系统会根据识别出的用户意图判断对话流程该走向哪一个后续节点如果识别出用户意图是房源推荐流程就进入推荐子图如果是房源预定则进入预定子图若是普通咨询类问题就流转到扩展子图如果是查询个人信息就进入查询历史信息的对应节点。用户意图是路由判断的核心依据所以我们需要把识别出的意图信息也存入全局状态。后续配置条件分支时程序就能读取state中的用户意图字段自动判断下一步的执行路径。再看其余节点和子图的输出内容预定子图、查询个人信息节点、扩展子图最终输出的都是AI消息也就是展示给用户的最终回复内容。推荐子图执行完成后同样会返回AI消息向用户展示匹配的房源。不过推荐流程之后还有一个特殊环节系统推荐完房源会主动向用户询问是否需要预定。我们会根据用户的回复决定后续走向用户回复 “需要”流程就跳转至预定子图回复 “不需要”对话直接结束。为了实现这个交互逻辑我们需要新增一个节点并且在这个节点中使用中断机制由用户的输入来掌控后续流程。用户输入的 “需要” 或 “不需要” 是字符串类型这个结果也需要临时存储作为后续条件分支的判断依据。这里区分两种状态第一种是全局共享状态比如消息列表、用户偏好、用户意图主图和所有子图都可以读取使用第二种是临时私有状态也就是存储用户预定选择的字段。这个字段仅作用于推荐流程之后的条件分支其他子图和节点都不会使用因此无需全局共享。简单说明一下状态共享的规则主图中定义的全局状态所有子图理论上都能获取但子图可以自主选择是否接收。如果子图代码中没有声明对应状态参数就不会读取主图的共享数据只有主动声明参数才会完成数据同步。而我们刚刚提到的 “是否预定” 相关数据仅在局部流程中临时使用因此将它定义为私有状态即可该状态不会出现在整个流程图的最终输出结果里只在流程执行的中间环节生效。下面代码实现主状态State全局共享状态。状态继承自MessagesState自动管理对话历史。user_intent用户意图表示用户输入的问题含义推荐 or 预定 or 查询 or 其它user_preferences用户偏好信息实现数据共享节点间传输私有状态NeedReserveOutput当推荐子图执行完成后用户获取用户是否想要预定房源的意向。从而决定是否执行预定子图。state/main.pyfrom 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_typeshuman) 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用户的历史偏好信息如下 f1. 最低预算{prefs.get(budget_min)} f2. 最高预算{prefs.get(budget_max)} f3. 已预定过的信息{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_typeshuman) 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用户的历史偏好信息如下 f1. 最低预算{prefs.get(budget_min)} f2. 最高预算{prefs.get(budget_max)} f3. 已预定过的信息{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_schemaContextSchema) 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(xrayTrue).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 端口前端页面正式调用线上接口最终完成整个租房智能助手项目的交付。
