☰
LangChain Agent 流式问答助手:检索与工具调用融合实战
2026/10/14 3:59:45 网站建设 项目流程

1. 从零理解这个流式问答助手到底在解决什么问题

1.1 为什么“能检索”和“会调工具”必须同时存在

很多人做问答助手,第一反应就是把文档切块、向量化、塞进向量库,用户提问就去检索最相似的片段,拼进提示词让模型回答。这套流程能跑通,但实际用起来问题很明显:用户问“帮我算一下上季度三个产品的毛利率分别是多少”,检索回来的可能是财务制度文档里关于毛利率定义的段落,模型只能照着念定义,算不出来。因为检索解决的是“找信息”,而计算、查询实时数据、调用外部接口这些事,检索本身干不了。

所以一个真正能用的助手,必须同时具备两种能力:一是从私有知识库里找到相关背景,二是根据问题类型决定要不要调用工具去执行具体操作。LangChain 里的 Agent 机制就是干这个的——它让模型自己判断“这个问题我该检索还是该调工具,还是两个都要”。而流式输出解决的是另一个体验问题:用户不想等十几秒才看到第一个字,边生成边显示才像在对话。

这两件事叠在一起,复杂度不是简单相加。检索是异步的、工具调用是有副作用的、流式输出要求 token 逐个吐出来,三者要在一条链路上协调好,才是这个项目真正的难点。

1.2 适合谁来参考这套方案

如果你已经写过基础的 LangChain 链,能跑通RetrievalQA,但发现它不够灵活;或者你做过 Agent 但输出是“憋一大段才出来”,体验很差;再或者你正在设计一个内部知识助手,需要它既能查文档又能查数据库、调接口,那这套思路就是给你准备的。

我不假设你是 LangChain 专家,但我会默认你知道什么是提示词、什么是向量检索、什么是 API 调用。如果你完全没接触过,建议先把 LangChain 的LLMChain和VectorStore两个概念搞清楚再回来,不然中间一些设计取舍你会看得云里雾里。

1.3 整体架构一句话说清

整个助手的核心是一条Agent 执行链,它内部挂了两类工具:一类是检索工具(Retriever Tool),一类是功能工具(比如计算器、时间查询、外部 API 封装)。用户输入进来后,Agent 先让模型做一次“思考”,决定调用哪个工具、传什么参数;工具返回结果后,模型再决定是继续调工具还是直接生成最终回答。整个过程通过流式回调把中间步骤和最终 token 实时推给前端。

听起来不复杂,但每一步都有坑。下面我按模块拆开讲,把选型理由、参数细节和踩过的坑都摊开说。

2. 核心模块拆解与选型背后的真实考量

2.1 检索工具怎么封装才不让 Agent 犯傻

LangChain 提供了create_retriever_tool这个便捷函数,把向量库的 retriever 包成一个工具。但直接包会有一个隐蔽问题:工具描述写得太笼统,模型不知道什么时候该用它。

我试过两种描述写法,效果差距很大。第一种是默认的“搜索知识库”,模型经常在应该调计算器的时候也去检索一下,浪费一轮调用。第二种是明确写清楚“当问题涉及公司内部政策、产品文档、历史记录等私有信息时使用此工具,不要用于数学计算或实时数据查询”,模型的选择准确率明显提升。

具体封装代码大概长这样:

from langchain.tools.retriever import create_retriever_tool retriever_tool = create_retriever_tool( retriever=vectorstore.as_retriever( search_type="mmr", search_kwargs={"k": 4, "fetch_k": 10, "lambda_mult": 0.6} ), name="search_internal_docs", description=( "用于检索公司内部文档、产品手册、历史项目记录等私有知识。" "当用户询问制度、流程、产品参数、过往案例时使用。" "不要用于数学运算、当前时间查询或外部实时数据获取。" ) )

这里有几个参数值得说。search_type="mmr"是最大边际相关性检索,它会在“和问题相关”与“结果之间不重复”之间做平衡。纯相似度检索容易返回一堆内容雷同的片段,MMR 能缓解这个问题。fetch_k=10表示先取 10 个候选,再从里面挑 4 个多样性最好的。lambda_mult=0.6是相关性和多样性的权重,0 表示完全看多样性,1 表示完全看相关性,0.6 是我实测下来比较稳的中间值。

注意:工具描述里一定要写“不要用于什么”,这比只写“用于什么”更能约束模型行为。这是我在多个项目里反复验证过的经验。

2.2 功能工具的设计原则:窄接口、强校验

除了检索,助手还需要一些功能工具。我一般会加三类:计算器、时间查询、以及一个通用的外部 API 调用封装。这里的原则是接口要窄,参数要强校验。

举个例子,计算器工具不要做成“传一个表达式字符串进来我 eval 一下”,那样既危险又容易出错。我通常用numexpr或自己写一个简单的四则运算解析,只接受数字和运算符:

import numexpr from langchain.tools import tool @tool def calculate(expression: str) -> str: """用于执行数学计算。输入必须是纯数学表达式,例如 '(120-80)/80*100'。 不要传入文字、单位或变量名。""" allowed = set("0123456789+-*/(). ") if not set(expression).issubset(allowed): return "错误:表达式包含非法字符,只允许数字和 + - * / ( ) ." try: result = numexpr.evaluate(expression).item() return f"计算结果:{result}" except Exception as e: return f"计算失败:{str(e)}"

时间查询工具更简单,但要注意时区。我一般直接返回带时区的 ISO 格式,让模型自己去格式化给用户看。外部 API 工具则要加超时和重试,并且把返回结果截断到合理长度,不然一个接口返回几万字,直接把上下文撑爆。

2.3 流式输出为什么不能只靠stream=True

很多人以为在 LLM 初始化时加个streaming=True就完事了。实际上在 Agent 场景下,这个参数只控制最终生成那一步的 token 流式,中间的工具调用、检索过程你是看不到的。用户会经历一段“什么都没发生”的空白期,然后突然开始吐字。

要解决这个问题,需要用astream_events这个 API。它会以事件流的形式把链路上每个环节都推出来,包括on_tool_start、on_tool_end、on_chat_model_stream等。前端可以根据事件类型分别处理:工具开始时显示“正在检索文档...”,工具结束时显示“已找到 4 条相关记录”,模型生成时逐字显示回答。

这个改造是体验提升的关键。我做过对比,同样一个问题,不加事件流用户平均要等 8 到 12 秒才看到第一个字;加了之后,1 秒内就能看到“正在检索”的提示,心理等待感大幅降低。

2.4 Agent 类型选择:为什么不用最火的那个

LangChain 里有好几种 Agent 类型,比如OPENAI_FUNCTIONS、REACT、STRUCTURED_CHAT等。我的选择是OPENAI_FUNCTIONS或对应的工具调用型 Agent,理由有三点。

第一,工具调用型 Agent 对参数格式的遵循度最高。ReAct 那种靠文本解析“Action: xxx”的方式,模型偶尔会写错格式导致解析失败,而工具调用是结构化输出,稳定性好很多。第二,它天然支持并行工具调用,虽然我这个场景里并行需求不强,但留着扩展空间。第三,流式事件里工具调用的边界更清晰,方便前端做状态展示。

当然,如果你用的模型不支持工具调用,那就只能退回 ReAct。但只要你用的是主流几家支持 function calling 的模型,我强烈建议用工具调用型。

3. 完整实操流程与关键环节实现

3.1 环境准备与依赖版本锁定

这个项目对版本比较敏感,LangChain 迭代快,不同版本 API 差异大。我用的组合是:

pip install langchain==0.2.x pip install langchain-openai==0.1.x pip install langchain-community==0.2.x pip install chromadb==0.4.x pip install numexpr==2.8.x

提示:不要盲目追最新版。我有一次升级到某个新版本后,astream_events的事件结构变了,前端解析全挂。建议在requirements.txt里锁死小版本号。

向量库我选 Chroma,原因是它本地跑、零配置、和 LangChain 集成度高。如果你数据量大或者要多用户并发,可以换 Milvus 或 Qdrant,但开发阶段 Chroma 足够。

3.2 文档入库与检索器构建

文档处理这块,我的流程是:加载 -> 切分 -> 向量化 -> 入库。切分参数很关键,我一般用RecursiveCharacterTextSplitter,chunk_size=500,chunk_overlap=80。

为什么是 500 和 80?chunk 太大,检索精度下降,因为一个块里混了多个主题;chunk 太小,上下文不完整,模型拿到碎片拼不出完整意思。500 个字符大约对应中文 250 到 300 字,能容纳一个完整段落。overlap 设 80 是为了防止关键信息刚好被切在边界上,两边各留一点冗余。

from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_community.vectorstores import Chroma from langchain_openai import OpenAIEmbeddings splitter = RecursiveCharacterTextSplitter( chunk_size=500, chunk_overlap=80, separators=["\n\n", "\n", "。", "!", "?", ";", " ", ""] ) chunks = splitter.split_documents(raw_docs) embeddings = OpenAIEmbeddings(model="text-embedding-3-small") vectorstore = Chroma.from_documents(chunks, embeddings, persist_directory="./db")

注意separators里我把中文标点加进去了。默认的分隔符只认英文标点和空格,中文文档切出来经常在句子中间断开。加上中文标点后,切分点更符合语义边界。

3.3 Agent 组装与提示词设计

Agent 的提示词我改了好几版。最初用默认的,模型经常在不需要工具的时候也硬调一个。后来我加了一段明确的决策规则:

system_prompt = """你是一个内部知识助手。你有两类工具可用: 1. search_internal_docs:检索私有文档,用于制度、流程、产品参数、历史案例类问题。 2. calculate / get_current_time / call_external_api:用于计算、时间查询、实时数据获取。 决策规则: - 如果问题涉及内部知识,先检索再回答。 - 如果问题需要计算,直接调 calculate,不要自己心算。 - 如果问题同时涉及内部知识和计算,先检索获取数据,再调 calculate 计算。 - 如果问题你无法通过工具解决,直接说明能力边界,不要编造。 """

这段提示词的核心是最后一条“不要编造”。没有这句话的时候,模型遇到检索不到的内容会硬编一个答案,这在内部助手场景里是致命的。

组装 Agent 用create_openai_tools_agent配合AgentExecutor:

from langchain.agents import create_openai_tools_agent, AgentExecutor agent = create_openai_tools_agent(llm, tools, prompt) executor = AgentExecutor( agent=agent, tools=tools, max_iterations=6, return_intermediate_steps=True, handle_parsing_errors=True )

max_iterations=6是防止模型陷入死循环。我遇到过模型反复调同一个工具七八次的情况,设个上限能兜底。handle_parsing_errors=True让解析失败时返回错误信息给模型,让它自己纠正,而不是直接抛异常中断。

3.4 流式事件处理与前端对接

这是整个项目最核心的部分。用astream_events消费事件流:

async def stream_answer(question: str): async for event in executor.astream_events( {"input": question}, version="v2" ): kind = event["event"] if kind == "on_tool_start": yield {"type": "status", "content": f"正在调用 {event['name']}..."} elif kind == "on_tool_end": yield {"type": "tool_result", "content": event["data"]["output"][:200]} elif kind == "on_chat_model_stream": chunk = event["data"]["chunk"] if chunk.content: yield {"type": "token", "content": chunk.content} elif kind == "on_chain_end" and event["name"] == "AgentExecutor": yield {"type": "done", "content": ""}

前端拿到这些事件后,status类型显示加载提示,tool_result可以折叠展示检索到的片段,token类型追加到回答区域,done表示结束。

注意:on_chat_model_stream事件在 Agent 的每一轮模型调用里都会触发,包括模型决定调工具的那一轮。所以你会看到一些“思考过程”的 token 也被推出来。我的处理方式是判断当前是否处于最终生成阶段,或者干脆把中间 token 也展示出来,让用户看到模型的推理过程,反而增加了透明度。

3.5 参数计算实例:检索条数怎么定

k=4这个值不是拍脑袋定的。我做过一组测试:同一个问题集,分别用 k=2、4、6、8 跑,看回答准确率和 token 消耗。

k 值回答准确率平均输入 token单次响应时间
272%约 12003.2s
489%约 21004.1s
691%约 30005.6s
891%约 39007.3s

k 从 2 到 4 准确率提升明显,从 4 到 6 提升很小但 token 涨了 40%,6 以上基本没收益。所以 4 是性价比拐点。当然这个结论依赖你的文档密度和问题类型,建议你自己跑一遍小规模测试再定。

4. 常见问题与排查技巧实录

4.1 模型不调工具,直接瞎编答案

这是最高频的问题。排查顺序如下:

第一,检查工具描述是否足够明确。如果描述写的是“搜索”,模型可能觉得“我直接答也行”。改成“必须用于查询内部私有信息”这种强约束措辞。

第二,检查提示词里有没有“不要编造”的指令。没有的话模型默认会倾向于直接生成。

第三,检查模型本身的能力。有些小模型对工具调用的支持很差,换一个支持 function calling 的模型试试。

第四,看max_iterations是不是设太小,模型还没来得及调工具就被截断了。

4.2 检索结果不相关,答非所问

先看切分粒度。如果 chunk 太大,一个块里混了好几个主题,检索出来的内容自然不聚焦。试着把 chunk_size 降到 300 到 400 再试。

再看 embedding 模型。不同模型对中文的语义理解差异很大,换一个中文效果好的 embedding 模型对比一下。

还有一个容易忽略的点:查询本身可能太短或太模糊。用户问“那个流程是什么”,检索器根本不知道“那个”指什么。可以在检索前加一步查询改写,让模型把问题补全成完整句子再检索。

4.3 流式输出卡顿或中断

如果前端收到几个 token 后卡住,大概率是工具调用和流式生成的切换没处理好。检查astream_events的version参数,v1 和 v2 的事件结构不同,用错了会漏事件。

另一个可能是网络问题导致 SSE 连接断开。建议在前端加心跳和自动重连,后端加超时控制。

4.4 工具调用参数格式错误

模型有时候会把参数写成自然语言,比如calculate("一百二十除以三")而不是calculate("120/3")。解决办法是在工具描述里给一个明确的输入示例,并且加上参数校验,非法输入直接返回错误提示让模型重试。

4.5 常见问题速查表

现象可能原因排查动作
模型不调工具工具描述模糊 / 提示词缺约束改描述、加“不要编造”指令
检索不相关chunk 过大 / embedding 不匹配降 chunk_size、换 embedding 模型
流式中断事件版本不对 / 网络断开检查 version 参数、加心跳重连
参数格式错工具描述缺示例加输入示例、加参数校验
响应太慢k 值过大 / 工具超时降 k、给工具加超时
重复调同一工具模型陷入循环设 max_iterations、优化提示词

4.6 几个我踩过的坑

第一个坑:一开始我把检索工具和计算工具的描述写得很像,都用了“查询”这个词,结果模型经常混淆。后来把检索工具的描述改成“检索文档”,计算工具改成“执行数学运算”,用词差异化后准确率上来了。

第二个坑:astream_events在 Agent 场景下会推送中间步骤的 token,我一开始没过滤,前端把模型的“思考过程”也当最终答案显示了。后来加了状态判断,只在最后一轮生成时展示 token。

第三个坑:Chroma 的持久化目录如果被多个进程同时写,会锁库报错。开发时用单进程,生产环境换支持并发的向量库。

第四个坑:工具返回结果太长直接把上下文撑爆。我后来给所有工具加了返回长度截断,超过 500 字符就截断并加省略号。

5. 性能优化与扩展方向

5.1 检索缓存与工具结果缓存

同一个问题反复问,每次都走一遍检索和模型调用很浪费。我加了一层基于问题文本的缓存,用functools.lru_cache或者 Redis 都行。命中缓存直接返回,响应时间从几秒降到毫秒级。

工具结果也可以缓存,尤其是那些不常变的外部数据查询。但要注意设置合理的过期时间,不然会返回陈旧数据。

5.2 多轮对话的上下文管理

单轮问答跑通后,下一步就是支持多轮。核心是把历史对话拼进提示词,但要注意长度控制。我的做法是只保留最近 5 轮对话,更早的做摘要压缩。摘要本身也用模型生成,把十几轮对话压成一段话,既保留关键信息又不占太多 token。

5.3 工具的动态注册与权限控制

当工具数量增多后,不是每个用户都能用所有工具。我设计了一个简单的权限层:每个工具带一个required_role标签,Agent 组装时根据当前用户角色过滤工具列表。这样同一个助手框架可以服务不同权限的用户,不用维护多套代码。

5.4 监控与日志

生产环境一定要加日志。我记录每次请求的:问题文本、调用了哪些工具、检索到的文档 ID、最终回答、总耗时、token 消耗。这些数据既能用来排查问题,也能用来分析用户都在问什么,反过来优化文档和工具设计。

日志我用结构化格式写,方便后续用脚本分析。比如每次请求输出一行 JSON,包含trace_id、question、tools_called、latency_ms、token_usage等字段。

6. 我个人在实际操作中的几点体会

这套东西我从第一版跑通到相对稳定,大概迭代了七八轮。最大的体会是:Agent 的稳定性不取决于模型多强,而取决于你对边界的约束有多清晰。工具描述写清楚“什么时候用、什么时候不用”,提示词写清楚“不能编造”,参数校验写清楚“什么格式合法”,这三件事做到位,哪怕用中等能力的模型,效果也不会差。

另一个体会是流式输出这件事,技术实现只是一半,另一半是前端的状态设计。用户需要知道“系统在干活”,哪怕只是显示“正在检索文档”这几个字,体验就完全不一样。我见过很多项目把流式当成纯技术优化,忽略了它本质上是用户体验设计。

最后说一个扩展思路:这套架构稍加改造就能支持多知识库切换。把检索工具的参数里加一个collection字段,模型根据问题判断该查哪个库,就能实现“产品文档走一个库、财务制度走另一个库”的效果。这个我还在试验阶段,等跑稳了再单独写一篇。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询