证据召回节点-springAI

一、前言证据召回相当于一个rag吧。如果你用graph来实现那就要依赖 state 来共享数据要做好前后节点的承接。召回什么内容看程序想实现什么功能来定。查询重写就是将用户模糊的问题重写得更完整与上下文文更相关去年呢昨天的学会sink 和 flux 写和读数据往页面推送提升体验感。类型很多StringChatResponseFluxChatResponseFluxGraphResponseStreamingOutputMapString, Object StringBuilder 等等需要弄清楚不要搞混了。皆个人理解该死的枯燥符号我实在记不住只能不断靠思考类比去理解并记住有误欢迎指正。感谢阅读。二、流程继承NodeAction接口实现 MapString, Objectapply方法。1.从传入的state变量中提取出问题 question 智能体ID agentId 上下文 multiTurn2.传入question和multiTurn构建查询重写提示词prompt和构建意图识别提示词一样⬇️构建意图识别提示词https://blog.csdn.net/Blink_BT/article/details/159735724?spm1001.2014.3001.55013.传入 prompt 调用 LLM 进行查询重写返回FluxChatResponseresponseFlux4.构建Sink管道evidenceDisplaySink用于后续将证据写入5.final resultMap 是 Flux流 内部用来暂存状态的地方6.构建一个生成器generator向前端推送查询重写内容并将查询重写agentIdevidenceDisplaySink用于触发证据召回getEvidences()将结果存入 resultMap7.再构建一个生成器evidenceFlux将evidenceDisplaySink转换为 Flux 类型并向前端推送证据召回的内容8.return用concatWith串行拼接先执行 generator 再执行evidenceFlux。三、讲解1.三个获取// 从state中提取question和agentId String question StateUtil.getStringValue(state, INPUT_KEY); String agentId StateUtil.getStringValue(state, AGENT_ID); Assert.hasText(agentId, Agent ID cannot be empty.); String multiTurn StateUtil.getStringValue(state, MULTI_TURN_CONTEXT, (无));2.构建提示词跳转链接看之前的3.LLM 执行调用 LLM 根据提示词重写一套话术 responseFlux 逻辑下次展开FluxChatResponse responseFlux llmService.callUser(prompt);4.SinkSinks.ManyString evidenceDisplaySink Sinks.many().multicast().onBackpressureBuffer();56.generator 推送查询重写并触发证据召回456一起讲先说 Flux。我们可以理解成“电视机”flux 流是“频道”被前端订阅因此就能“播放”在前端页面告诉用户“正在查询重写......”“正在证据召回......”“正在向量化文件......”等等让进程透明化也不会让用户干等一个加载圈圈优化体验这就是“流式输出”。那有人要问了“小白小白我看的一些网课都是直接agent.stream()来实现流式输出的呀为什么要包装成 Flux 呢”。看下面这段代码FluxUtil 是流的拼接器通过Flux.just(...)添加两段头尾流正在查询重写以更好召回evidence...和“\n查询重写完成”与中间的查询重写流 responseFlux 拼接这样就变成了一段流。而一般直接调用大模型.stream()就是直接输出大模型的回答agent.stream()只给你“大模型原样吐出来的东西”就像看直播——镜头只对着舞台。而我们要在“重写开始前”和“重写结束后”插播自己的字幕“正在查询重写……”相当于导播要在直播里切广告、加片头片尾——这就要自己拼一个 Flux 出来用Flux.just(...)造几段“人工片段”跟大模型的流拼在一起。这就是FluxUtil.createStreamingGenerator干的事。讲完 Flux再说 Sink。Flux读端订阅它数据流出来Sink写端往里tryEmitNext数据流进去sink.asFlux()就是把写端转成读端——所以它俩本质是一体两面不是敌对关系。Sink 和 Flux 是同一个管道的一进一出Flux 是“出水的龙头”订阅即出水Sink 是“进水的口子”你往里灌水。把 Sink 转成 Flux就是把这个进水管接到下一个出水管上。最后一个“诡异”的就是这个result - {}它是回调函数。你可以把它理解成“谢幕动作”正片演完了导演才喊“卡”然后干收尾的活。先说result是什么。LLM 是一条一条吐字的流式等它吐完了才把整段内容拼成一个完整的result递给你。这段逻辑是FluxUtil.createStreamingGenerator干的事它不仅拼接flux还用将 flux 流过的内容一点点收集再转换成 String 类型。所以result就是“重写后的完整查询”。再说大括号里的活。流结束了要干两件事getEvidences(result, agentId, evidenceDisplaySink)——①拿着这个重写好的查询去向量库召回证据。②顺手把证据塞进evidenceDisplaySink交给下一个 Flux 去展示。return resultMap—— 把召回到的证据汇总进resultMap作为这个节点的最终产出。所以result - {}不是“中途在干什么”而是“等第一个流播完触发的收尾交接”这也解释了为什么有“两个返回”——它们其实是两个 Flux 各自的“谢幕动作”不是同一次返回第一个 Flux 的谢幕拿重写结果去查证据塞进 sink。第二个 Flux 的谢幕证据已经播完了直接把resultMap交出去当最终结果。用于传入到 state 中两个 Flux 用concatWith串成一条前一个的“谢幕动作”给后一个准备了素材所以看起来“返回了两次”其实是接力。final MapString, Object resultMap new HashMap(); FluxGraphResponseStreamingOutput generator FluxUtil.createStreamingGenerator( this.getClass(), state, responseFlux, Flux.just(ChatResponseUtil.createResponse(正在查询重写以更好召回evidence...), ChatResponseUtil.createPureResponse(TextType.JSON.getStartSign())), Flux.just(ChatResponseUtil.createPureResponse(TextType.JSON.getEndSign()), ChatResponseUtil.createResponse(\n查询重写完成)), result - { resultMap.putAll(getEvidences(result, agentId, evidenceDisplaySink)); return resultMap; });7.evidenceFlux 推送召回的证据FluxGraphResponseStreamingOutput evidenceFlux FluxUtil.createStreamingGenerator( this.getClass(), state, evidenceDisplaySink.asFlux().map(ChatResponseUtil::createPureResponse), Flux.empty(), Flux.empty(), result - resultMap);8.return先执行 generator 再执行 evidenceFlux。return Map.of(EVIDENCE, generator.concatWith(evidenceFlux));