1. 为什么要做“检索工具流式”三合一的助手1.1 单模型问答的三个硬伤先聊点背景。我前面在 LangChain 系列里写过 RAG 问答、Prompt 优化和记忆管理每一篇解决一个具体问题。但这段时间在帮团队做内部知识库助手时发现单独拿出来任何一块落地效果都不够。原因很简单靠一次大模型推理很难同时满足三件事知识够新、数据够准、响应够快。第一知识库问题。模型训练有截止时间你问它最近的产品版本更新、最新政策条款它要么胡编要么道歉。这个必须靠检索增强RAG把企业私有文档灌进去。第二实时数据问题。有些问题必须查外部系统比如库存、价格、天气、数据库状态。模型没有手所以得靠工具调用Tool Calling让它“会用”函数。第三等待体验问题。一次生成几百字的回答如果等五秒才一次性吐出用户早就划走了。流式输出是刚需它让首字延迟降到几百毫秒打字机式反馈会极大改善体验。这三个硬伤凑在一起就催生了标题里这个“能检索又会调工具的流式问答助手”。它本质上是一个把 RAG、Function Calling、Streaming 串起来的智能体雏形也是实际业务里出现频率最高的一种需求组合。1.2 技术选型为什么用 LangChain 这套组合最省心选型上我坚持用 LangChain 而不是纯手写 HTTP 调用不是因为框架有多神秘而是因为它把三个能力抽象得非常干净检索VectorStoreRetriever 一行代码就能拿到相似文档工具tool 装饰器加 bind_tools普通函数秒变可调用的工具流式链对象自带 astream天然支持 token 级别传输这种组合的价值在迭代期最明显。纯手写的话你要自己管理 embedding 调用、向量库查询、tool_calls 解析、SSE 协议、消息历史轮回……工作量和框架差不多但代码完全是自己的技术债。LangChain 把 90% 的模板逻辑吃掉剩下 10% 的业务逻辑由你自己控制。整体架构我拆成三层检索层负责把文档变成可查询的向量工具层负责注册函数、绑定模型、执行工具循环流式层负责把偏文本和状态信息实时推到用户端。这三层解耦之后后面换 embedding 模型、换向量库、加两个新工具都不需要推倒重来。2. 三个核心能力逐个拆解从原理到细节2.1 检索增强知识库是怎么被“喂”给模型的RAG 看起来就是“搜索几段文档塞进 Prompt 里”但想让模型真正用上这些背景资料有几个细节会决定质量。先看文档切分。很多人直接按字符硬切结果语义就被切断在了两个 chunk 的边界上。我常用的参数是chunk_size500chunk_overlap50。这个 size 是经验值既能让每段文字保持足够上下文又不会超出模型 Token 上限太多。overlap 保证前后文有交叠关键信息不会被硬生生劈成两半。对比几个常见配置切分方式场景优点缺点固定字符 200~500通用简单稳定适合快速验证可能切断语义按分隔符递归切分半结构化文本保留段落结构召回更准长度波动大按 Markdown 标题切分文档结构清晰每块自带标题上下文依赖文档结构规范按语义切分长文本语义最完整计算成本高然后是 embedding。离线演示和知识库不大时本地跑BAAI/bge-small-zh或者用在线服务都行。我自己测试下来中文问答场景里 OpenAI 的text-embedding-3-small和 bge 系列都有不错表现关键不是比个高低而是 embedding 和向量检索的相似度算法要匹配。FAISS 默认用 L2 距离但很多 embedding 模型是用余弦相似度训练出来的所以我会在检索器里显式设置距离策略为余弦否则召回结果会明显偏移。检索层的另一个常见坑是top_k太小或太大。top_k 设 2漏召回风险高设 10噪声和 Token 成本都上去了。我习惯先取 4 到 5 段再在 Prompt 里告诉模型“资料排序不代表可靠性请按内容相关性判断”。2.2 工具调用让模型“会用”函数的关键设计工具调用是“助手能动手干活”的关键。LangChain 里把普通函数变成工具非常简单但模型能不能正确调用完全取决于你“说明书”写得好不好。函数签名必须带完整类型注解docstring 必须写清楚函数功能和各参数含义。因为这些信息会被拼进模型的工具描述里模型是“读”这段描述来决定要不要调用这个函数、传入什么参数。你可以把 docstring 理解为给模型看的功能说明书写不清楚模型就只能瞎猜。我一般这样设计工具tool def get_weather(city: str) - str: 查询指定城市的实时天气。 参数说明 - city: 城市中文名例如“杭州”。 返回该城市的天气状况与温度。 weather_map { 上海: 多云24℃东风3级, 北京: 晴18℃微风, 杭州: 小雨21℃东南风2级 } return weather_map.get(city, f{city}暂无数据请稍后重试)模型在收到一个问题时会先看工具列表里有没有匹配函数有的话就返回一个tool_calls数组而不是直接生成文本。这个数组里包含工具名、参数和调用 ID。执行完工具后把结果以ToolMessage回传给模型模型再决定是继续调用工具还是产出最终回答。工具设计的原则我总结成三条返回文本要短而要结论化不要回传一坨 JSON 让模型再解析异常要在函数内部捕获并转成友好文本别让异常炸断整个链路工具名要一眼能看出用途模型对“语义化名称”的分辨率比想象中高。2.3 流式输出从 SSE 到 LangChain 的 token 流流式输出的底层原理并不神秘。模型服务端把生成的 token 切片通过 Server-Sent EventsSSE逐帧传给客户端。LangChain 在这个基础上封装了stream和astream调用链只要声明为 LCEL 结构就能一层层把 token 流透传出来。但有一个细节很多人会踩坑不是链上所有环节都支持流式。向量库检索、文档重排序这类操作是“一次性返回”的它们发生在流式开始之前真正支持 token 流的只有 LLM 调用环节。所以我的做法是把检索动作放在前等上下文准备好之后再进入流式生成阶段。另一个容易困惑的点是如果模型决定调用工具这个阶段往往没有文本 token 产生。站在用户终端看就表现为“问题问完了界面一直没反应然后突然蹦出一段工具调用状态”。这不是 bug而是工具调用本质上是在做决策和动作决策过程没有自然语言输出。为了不让人以为服务挂了我通常会在前端加“思考中”或“正在调用工具”的状态提示。3. 完整代码实战跑通一个可流式的问答助手3.1 安装依赖与初始化建议用虚拟环境装依赖版本锁的是当前最新稳定版。pip install langchain langchain-openai langchain-community langchain-core faiss-cpu模型我选了 OpenAI 兼容接口这样本地如果想换 DeepSeek、通义或其他 OpenAI 格式的服务只需要改 base_url 和 api_key代码完全复用。环境变量建议放到项目根目录的.env里。export OPENAI_API_KEYsk-xxx export OPENAI_BASE_URLhttps://api.openai.com/v13.2 搭建检索层我先准备一份测试文本模拟企业内部的产品手册片段存成product_docs.txt。然后用RecursiveCharacterTextSplitter切分并建立向量库。from langchain_community.document_loaders import TextLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import FAISS raw_docs TextLoader(product_docs.txt, encodingutf-8).load() splitter RecursiveCharacterTextSplitter(chunk_size500, chunk_overlap50) doc_chunks splitter.split_documents(raw_docs) embeddings OpenAIEmbeddings(modeltext-embedding-3-small) vectorstore FAISS.from_documents(doc_chunks, embeddings) retriever vectorstore.as_retriever(search_kwargs{k: 4})这里有个细节值得说TextLoader读进来是带元数据的文档对象切分后保存到 FAISS。本地验证时用 FAISS 足够了速度极快且无需额外服务。如果到线上多进程部署建议换成 pgvector 或 Milvus 这类支持持久化与并发查询的向量库。search_kwargs{k: 4}表示每次最多拉 4 段相关内容。同时我会写一个格式化函数把多段文档拼成 Prompt 里的上下文块。def format_docs(docs): return \n\n---\n\n.join([d.page_content for d in docs])3.3 定义工具集工具我注册两个一个是查询天气一个是查询股票价格。真实业务里你可以换成查订单状态、查库存、写工单等等。工具函数本身不需要接入 LangChain 的任何基类普通函数加tool就行。tool def get_weather(city: str) - str: 查询指定城市的实时天气。 Args: city: 城市中文名比如“杭州”。 weather_map {杭州: 小雨21℃, 上海: 多云24℃, 北京: 晴18℃} return weather_map.get(city, f{city}暂无数据) tool def get_stock_price(code: str) - str: 查询 A 股股票的实时价格。 Args: code: 六位股票代码比如“600519”。 price_map {600519: 1499.00 元, 000001: 11.20 元} return f股票 {code} 最新报价{price_map.get(code, 暂无数据)} tools [get_weather, get_stock_price] tools_map {tool.name: tool for tool in tools}你一定想问docstring 用中文还是英文实测下来模型对两种语言的理解都不差但参数描述最好用中文因为中文问题的场景下模型用中文描述来生成参数值匹配率会高一点。另外记得函数的返回一定要是字符串模型后续是要把它拼接进上下文继续推理的。3.4 构建支持工具循环的流式链这是整篇的核心部分。我先把模型实例化并绑定工具。from langchain_openai import ChatOpenAI from langchain_core.messages import SystemMessage, HumanMessage, AIMessage, ToolMessage model ChatOpenAI(modelgpt-4o-mini, temperature0, streamingTrue) model_with_tools model.bind_tools(tools)temperature0是为了让工具选择和参数生成尽量稳定问答类场景不需要模型发挥创造力。接着定义主循环。我选择手动管理消息列表而不是引入复杂的 Agent 执行器因为这样能清楚看到每一轮发生了什么对理解工具循环最有帮助。async def ask_agent_stream(question: str): # 第一步检索相关文档 docs await retriever.ainvoke(question) context format_docs(docs) # 组装初始消息 system_prompt ( 你是一个企业问答助手。请优先使用参考资料中的信息回答 如果问题需要实时数据自行调用工具获取。 ) messages [ SystemMessage(contentsystem_prompt), HumanMessage(contentf问题{question}\n\n参考资料\n{context}) ] for _ in range(MAX_ITERATIONS): # 第二步流式接收模型输出 chunks [] async for chunk in model_with_tools.astream(messages): if chunk.content: yield chunk.content chunks.append(chunk) if not chunks: return # 合并流式片段中的完整工具调用信息 tool_calls chunks[-1].tool_calls content .join(c.content for c in chunks if c.content) if not tool_calls: # 没有工具调用说明答案生成完毕 return # 第三步把模型回复归档到消息列表执行工具 messages.append(AIMessage(contentcontent, tool_callstool_calls)) for tc in tool_calls: name tc[name] args tc[args] yield f\n[[工具调用] {name}({args})]\n try: result tools_map[name].invoke(args) except Exception as exc: result f工具执行异常{exc} messages.append(ToolMessage(contentresult, tool_call_idtc[id]))你在读这段代码时重点看“循环”是怎么闭合的模型输出AIMessage(tool_calls...)→ 追加到消息列表 → 执行工具 → 追加ToolMessage→ 再次调用astream。下一轮模型已经在上下文里看到了工具的真实返回结果所以它能继续做决策。可能你会疑惑为什么取chunks[-1].tool_calls而不是合并所有 chunk 的tool_call_chunks。严格方式是合并加总每一段的 tool_call_chunks但对大多数场景取最后一个 chunk 的完整调用元数据就够用了。如果你做的是多工具、多轮复杂决策建议迁移到 LangGraph 的 StateGraph。我在文末会再聊。3.5 从控制台到 Web API把流式转发给前端上面的生成器在控制台跑没问题但要接到真实前端需要处理 HTTP 流式转发。我通常用 FastAPI 写一个StreamingResponse接口把生成器的每个 chunk 原样发送给客户端。from fastapi import FastAPI from fastapi.responses import StreamingResponse app FastAPI() app.post(/chat) async def chat(question: str): async def gen(): async for piece in ask_agent_stream(question): yield piece return StreamingResponse(gen(), media_typetext/plain; charsetutf-8)前端用fetch配合ReadableStream接口逐帧读取或者直接用 EventSource。注意我上面返回的是纯文本流前端渲染时如果希望支持 Markdown还需要分段累积后交给渲染器。我自己上线时会在流里额外携带结构化字段比如retrieval_done、tool_start、token这样的 JSON 块前端根据字段显示不同的 UI 状态。这一步能明显提升“有人工参与”的专业感用户不会再对着转圈发呆。4. 踩坑记录与排查手册4.1 高频问题速查表这里直接列一张速查表都是我在这个项目里真实踩过的坑。现象可能原因解决办法流式一直不输出最后一下全吐出来模型接口没开 streaming或链路里混用了非流式节点导致缓存确认模型参数streamingTrue检查 LCEL 链路里没有中间.invoke()工具被反复调用不结束没有设置最大迭代次数工具结果描述不清导致模型判断没拿到答案加MAX_ITERATIONS保护工具返回文本要结论化模型乱传参导致工具报错参数描述不明确未做类型校验在 docstring 里加示例值函数内部加try/except回答没用上检索内容Prompt 没有强调优先使用资料chunk 内容与问题语义距离远重写 system prompt调大 top_k检查 embedding 匹配策略检索结果总是缺关键段落chunk_size 过大切断了上下文缩小 chunk_size 并增加 overlap尝试按标题切分首字延迟很高检索阶段没有流式输出用户提示把“正在检索”状态先行返回之后进入 token 流4.2 三个只有实战才能体会的经验第一个经验是“检索和工具要分离但工具优先级要高于检索”。当问题和实时数据强相关时比如“现在杭州天气怎么样”检索资料不仅没用还可能污染模型的判断。我在代码里做了一个简单路由先判断问题是否明显匹配工具意图如果匹配就跳过检索直接进工具循环。这个规则写在 system prompt 里实测可以把误用率降到很低。第二个经验是“工具结果也要参与流式体验设计”。工具执行通常会耗时几百毫秒到几秒如果终端毫无反馈用户会觉得“卡死了”。我现在的做法是进入工具循环时先输出一个极简状态行执行完毕再输出结果摘要然后回到正常生成。用户看到的是一个有节奏感的对话过程而不是两种状态之间的黑盒。第三个经验是“日志必须记录工具调用轨迹”。线上排查问题时没有日志简直寸步难行。我每次工具调用都会把tool_call_id、参数、耗时、返回结果前 200 个字符打点记录。这样用户说“答案不对”时我能马上定位到底是检索没召回、工具结果错误还是模型推理跑偏。再补一点关于迁移 LangGraph 的建议当你的工具数量超过三个或者状态要跨多轮维护手写循环会开始丑陋。LangGraph 的核心优势是把状态机显式化每一步节点做什么、条件边怎么走都画得清清楚楚。我这个示例刻意用手写循环是为了展示工具循环的本质生产环境我建议直接用 LangGraph 的create_react_agent或自定义 StateGraph后续维护成本低几个量级。最后再分享一个我在实际使用中的小技巧流式输出时别只顾着吐 token记得把“检索到的文档来源”也一并送回前端。用户看到答案底部列出的参考来源信任度会提升非常多。这个细节在很多产品里被忽略但它恰恰是内部问答工具能否被团队接受的关键一环。