用LangChain打造可检索、会调工具、流式输出的智能问答助手

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_size=500，chunk_overlap=50。这个 size 是经验值，既能让每段文字保持足够上下文，又不会超出模型 Token 上限太多。overlap 保证前后文有交叠，关键信息不会被硬生生劈成两半。

对比几个常见配置：

然后是 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 Events（SSE）逐帧传给客户端。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_KEY="sk-xxx" export OPENAI_BASE_URL="https://api.openai.com/v1"

3.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", encoding="utf-8").load() splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50) doc_chunks = splitter.split_documents(raw_docs) embeddings = OpenAIEmbeddings(model="text-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(model="gpt-4o-mini", temperature=0, streaming=True) model_with_tools = model.bind_tools(tools)

temperature=0是为了让工具选择和参数生成尽量稳定，问答类场景不需要模型发挥创造力。

接着定义主循环。我选择手动管理消息列表，而不是引入复杂的 Agent 执行器，因为这样能清楚看到每一轮发生了什么，对理解工具循环最有帮助。

async def ask_agent_stream(question: str): # 第一步：检索相关文档 docs = await retriever.ainvoke(question) context = format_docs(docs) # 组装初始消息 system_prompt = ( "你是一个企业问答助手。请优先使用参考资料中的信息回答；" "如果问题需要实时数据，自行调用工具获取。" ) messages = [ SystemMessage(content=system_prompt), HumanMessage(content=f"问题：{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(content=content, tool_calls=tool_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(content=result, tool_call_id=tc["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_type="text/plain; charset=utf-8")

前端用fetch配合ReadableStream接口逐帧读取，或者直接用 EventSource。注意我上面返回的是纯文本流，前端渲染时如果希望支持 Markdown，还需要分段累积后交给渲染器。

我自己上线时会在流里额外携带结构化字段，比如retrieval_done、tool_start、token这样的 JSON 块，前端根据字段显示不同的 UI 状态。这一步能明显提升“有人工参与”的专业感，用户不会再对着转圈发呆。

4. 踩坑记录与排查手册

4.1 高频问题速查表

这里直接列一张速查表，都是我在这个项目里真实踩过的坑。

4.2 三个只有实战才能体会的经验

第一个经验是“检索和工具要分离，但工具优先级要高于检索”。当问题和实时数据强相关时，比如“现在杭州天气怎么样”，检索资料不仅没用，还可能污染模型的判断。我在代码里做了一个简单路由：先判断问题是否明显匹配工具意图，如果匹配，就跳过检索直接进工具循环。这个规则写在 system prompt 里，实测可以把误用率降到很低。

第二个经验是“工具结果也要参与流式体验设计”。工具执行通常会耗时几百毫秒到几秒，如果终端毫无反馈，用户会觉得“卡死了”。我现在的做法是：进入工具循环时先输出一个极简状态行，执行完毕再输出结果摘要，然后回到正常生成。用户看到的是一个有节奏感的对话过程，而不是两种状态之间的黑盒。

第三个经验是“日志必须记录工具调用轨迹”。线上排查问题时，没有日志简直寸步难行。我每次工具调用都会把tool_call_id、参数、耗时、返回结果前 200 个字符打点记录。这样用户说“答案不对”时，我能马上定位到底是检索没召回、工具结果错误，还是模型推理跑偏。

再补一点关于迁移 LangGraph 的建议：当你的工具数量超过三个，或者状态要跨多轮维护，手写循环会开始丑陋。LangGraph 的核心优势是把状态机显式化，每一步节点做什么、条件边怎么走都画得清清楚楚。我这个示例刻意用手写循环，是为了展示工具循环的本质；生产环境我建议直接用 LangGraph 的create_react_agent或自定义 StateGraph，后续维护成本低几个量级。

最后再分享一个我在实际使用中的小技巧：流式输出时别只顾着吐 token，记得把“检索到的文档来源”也一并送回前端。用户看到答案底部列出的参考来源，信任度会提升非常多。这个细节在很多产品里被忽略，但它恰恰是内部问答工具能否被团队接受的关键一环。