☰
Ollama+LangChain流式渲染与工具调用闭环实践
2026/10/26 14:05:55 网站建设 项目流程

1. 裸奔的输出为什么劝退:从一次演示翻车说起

前阵子帮一个朋友看他做的本地问答小工具,功能其实已经跑通了:Ollama 起在本地,LangChain 接上去,问一句答一句,结果是对的。他把笔记本投到会议室大屏上演示,第一句话问出去,屏幕上开始蹦字——但不是一行一行往下走,而是在原地不停地重画,字符叠在一起,代码块里的缩进全乱,最后答案出来了,是一段挤成一坨的 Markdown 源码,星号和反引号全都露在外面。会议室里有人问了一句"这是还没做完吧",他当场就卡壳了。

这个场景特别典型。LangChain + Ollama 这套组合,把"能跑"的门槛降得非常低,但"能看"和"能干活"是两个完全不同的问题。美化输出解决的是前者,Function-calling 解决的是后者。很多人卡在这一步不是因为不会写代码,而是因为不知道这两个东西各自要处理哪些边界情况,踩了坑之后又找不到明确的报错信息,只能瞎试。

我打算把这两件事放在一起讲,因为它们在实际项目里是纠缠在一块的:你做了工具调用,模型会先输出一段"我要查一下天气"的思考文本,然后触发工具,拿到结果再组织语言,这个过程中的每一段都需要分别渲染,如果渲染器只考虑了单轮流式输出,一遇到工具调用就会乱套。反过来说,如果你只做了漂亮的渲染,模型却拿不到实时数据,那界面再好看也只是个玩具。

这篇内容适合已经能跑通最基础的 Ollama 对话、想把本地这套东西做到"能拿出去给人看"的读者。全文会围绕两个核心关键词展开:流式输出的分层渲染,以及Function-calling 的声明、解析、回填闭环。代码基于langchain_ollama这个包,也就是现在官方推荐的那条接入路径,不是老的langchain_community里的那套封装。如果你还在用旧包,建议先换过来,API 差异不小。

2. Ollama 接入 LangChain 时容易被忽略的接口细节

2.1 本地服务和云端接口的差异清单

很多人第一次从云端模型切到本地模型,会觉得"不就是换个 model 名字",然后被一堆莫名其妙的超时和截断教育。本地这套东西有几个点是硬差异,必须提前知道。

第一是上下文窗口。云端模型动辄 128K,本地跑一个 7B 或者 8B 的模型,默认上下文可能只有 2K 到 4K,num_ctx这个参数不显式设置的话,前面的对话轮次会被悄悄挤掉。表现出来就是"聊到第三轮它忘了第一轮说过什么",你以为是模型笨,其实是窗口满了在滑窗丢弃。我一般会把num_ctx显式写到 8192,除非显存实在吃紧。

第二是首次加载延迟。模型文件从磁盘载入内存,第一次请求可能要等十几秒甚至更久。如果你的代码里 timeout 设的是 5 秒,第一次调用必然失败,但第二次又成功了,这种"偶发失败"最难查。稳妥的做法是把 timeout 调到 60 秒以上,或者提前发一个空请求做预热。

第三是并发能力。本地服务默认的并行请求数很低,你要是开了多个线程同时打,后面几个会排队。做工具调用的时候尤其要注意,如果工具体内部又去调了一次模型,很容易形成自己等自己的死锁。

维度本地运行的常见默认值建议配置不配置的后果
上下文窗口2048 或 40968192多轮对话悄悄丢历史
请求超时5 到 30 秒60 秒以上首次冷启动必失败
温度0.80.1 到 0.3工具调用参数乱编
输出长度未限制或很短按场景显式设定答案被腰斩
并行数1 到 4视显存而定并发请求排队

温度这一项特别说明一下。做纯聊天,0.8 会让回答更生动;但一旦涉及工具调用,高温意味着模型可能把参数名写错、把 JSON 结构写歪。我在工具调用场景下基本都压到 0.1 到 0.2,宁可答案死板一点,也要保证参数能被正确解析。

from langchain_ollama import ChatOllama llm = ChatOllama( model="qwen2.5:7b", base_url="http://127.0.0.1:11434", temperature=0.2, num_ctx=8192, num_predict=1024, )

这段配置看起来平淡,但每一项都是被坑出来的。base_url显式写死比依赖环境变量好,因为有时候你本地起了两个实例,端口不一样,环境变量指错了方向,排查起来要命。

2.2 流式开关、超时与上下文长度

ChatOllama的流式能力是内置的,invoke和stream走的是同一条链路,只是前者把分片拼完再返回。这里有个细节:如果你用invoke,实际上底层也在流式接收,只是被聚合了,所以你在回调里挂的处理器照样会被触发。这个特性可以拿来做事后分析,但在做终端渲染的时候,老老实实用stream更直观。

buffer = "" for chunk in llm.stream("用三句话解释什么是向量检索"): buffer += chunk.content print(chunk.content, end="", flush=True)

上面这五行是最朴素的做法,能出字,但也就止步于"能出字"。它的问题在于:所有内容平铺直叙地堆在终端里,没有段落感,代码块和正文混在一起,长回答滚屏之后根本没法回看。这就是下一节要解决的事情。

还有一点值得单独提:流式模式下,工具调用的分片是按字段增量给出的。也就是说,第一个分片里可能只有工具名的一部分,第二个分片才把参数补全。这意味着你不能看到一个tool_calls非空就立刻去执行,必须等流结束、把所有分片合并之后,拿到完整的调用对象再分发。这个坑我在第一次写流式工具调用时踩过,症状是工具收到的参数永远是残缺的 JSON。

3. 分层渲染:把 token 流变成好看的终端界面

3.1 增量渲染的坑:为什么不能每来一个 token 就重新渲染全文

最直觉的做法是:每收到一个分片,就把缓冲区里的全文重新交给 Markdown 渲染器,然后清屏重画。这个方案在小段落上没问题,但一旦回答变长,问题就集中爆发。

第一个问题是闪烁。清屏重画整屏,视觉上就是一次闪烁。token 来得快的时候,一秒钟闪十几次,看久了眼睛疼。解决办法是用Live这类局部刷新容器,它只重绘变化区域,不会整屏清空。同时把刷新频率压到 10 到 15 次每秒,低于这个数会显得卡顿,高于这个数纯粹是浪费算力,肉眼分辨不出来。

第二个问题是渲染异常。Markdown 是块级语法,一个代码块以三个反引号开始,必须以三个反引号结束。流到一半的时候,缓冲区里是一个未闭合的代码块,渲染器要么把它当成普通文本处理(导致前半段显示成源码,后半段突然变成高亮块,视觉上跳一下),要么直接抛异常。表格也是同理,| a | b |这一行没写完的时候,列数不匹配,渲染出来是一堆错位的竖线。

第三个问题是中文宽度。终端的字符宽度计算里,中日韩字符占两列,英文字符占一列。如果你自己手写对齐逻辑,遇到中英混排必然错位。rich这类库内部处理了东亚宽字符,但前提是你不要自己去算 padding,一旦自己拼字符串,它也没法救你。

注意:不要在循环里对每个 chunk 都调用一次console.print,那会在终端里堆出成百上千行,翻都翻不回来。要么用Live覆盖刷新,要么按段落攒批输出。

3.2 代码块与表格的"半截状态"处理

针对半截状态,我试过三种策略,最后稳定下来的是第三种。

第一种是按行渲染:只在遇到换行符时才提交一次渲染。这个方案实现简单,闪烁问题天然解决,但缺点是模型输出长段落时(有些模型喜欢一口气写两百字不带换行),你会盯着一个不动的光标干等,流式的意义就没了。

第二种是滞后一帧:把缓冲区末尾的一小段(比如最后 40 个字符)扣住不渲染,只渲染前面的部分。这样未闭合的语法结构有很大概率被挡在门外。缺点是末尾的字会延迟出现,观感上有点像"卡带"。

第三种是状态机判定:维护一个极简的解析状态,记录当前是否在代码块内、是否在表格内、是否有未闭合的行内标记。提交渲染时,如果处于"不安全"状态,就把缓冲区截断到最后一个安全位置(比如最后一个完整的代码块结束标记之后),剩下的部分继续攒着;如果处于"安全"状态,就整段提交。这个方案的代码量大概三十行,但效果最接近商业产品。

class SafeBuffer: def __init__(self): self.text = "" self.committed = 0 self.in_fence = False def feed(self, piece: str) -> str: self.text += piece # 统计反引号围栏的开合状态 fences = self.text.count("```") self.in_fence = fences % 2 == 1 if self.in_fence: # 处于代码块内部,不提交新内容 return self.text[: self.committed] # 找最后一个空行作为安全切点,避免把半句话切开 cut = self.text.rfind("\n\n") if cut > self.committed: self.committed = cut return self.text[: self.committed]

这段逻辑不完美,比如嵌套列表里的空行也会被当成切点,但实测下来误判率很低,而且即使误判也只是提前提交,视觉上不会有明显问题。真正需要严格保证的只有代码块这一项,因为它的语法破坏最显眼。

配合Live使用大概是这样:

from rich.console import Console from rich.live import Live from rich.markdown import Markdown console = Console() buf = SafeBuffer() with Live(console=console, refresh_per_second=12, vertical_overflow="visible") as live: for chunk in llm.stream(prompt): snapshot = buf.feed(chunk.content) if snapshot: live.update(Markdown(snapshot)) # 流结束后把剩余的尾巴补上 live.update(Markdown(buf.text))

vertical_overflow="visible"这个参数值得留意。默认值在某些终端里会让超出屏幕的内容被裁掉,回答长了就看不到最新的字,改成visible之后超出的部分会交给终端自己滚动,跟普通输出的行为一致。

3.3 一个可复用的渲染器骨架

把上面的东西封装一下,我平常用的结构是这样的:一个TerminalRenderer类,对外暴露start()、feed(chunk)、finish()三个方法,内部管理 Live 容器和缓冲区。这样做的好处是后面加工具调用的时候,可以给同一个渲染器加"段落分隔"能力,而不需要重写。

class TerminalRenderer: def __init__(self, console: Console): self.console = console self.buf = SafeBuffer() self.live = None def start(self): self.live = Live( console=self.console, refresh_per_second=12, vertical_overflow="visible", ) self.live.start() def feed(self, piece: str): snapshot = self.buf.feed(piece) if snapshot: self.live.update(Markdown(snapshot)) def divider(self, label: str): # 工具调用、工具返回这类事件插入一条分隔提示 self.live.update(self.console.render_str(f"[dim]── {label} ──[/dim]")) def finish(self): self.live.update(Markdown(self.buf.text)) self.live.stop()

divider这个方法是为后面的工具调用准备的。模型在调用工具之前通常会说一句"我来查一下",然后触发调用,拿到结果之后再继续说。如果不加分隔,用户看到的就是一段话被硬生生切断,中间空了一段时间,然后接上另一段,完全不知道发生了什么。加一条弱化的分隔线,把"正在调用 XX 工具"和"工具返回结果"标出来,整个交互过程就清晰了。

4. Function-calling 的三段式:声明、解析、回填

4.1 工具声明:描述文本比函数签名更重要

很多人写工具的第一反应是纠结参数类型、必填项、默认值。这些东西当然要写对,但决定模型会不会正确调用你的工具、会不会把参数填对的,是那段自然语言描述,不是类型标注。

模型看到的只有文本,它不知道你函数内部怎么实现的,也不知道业务含义。如果你的描述写的是"查询天气",模型只知道这个工具跟天气有关,但它不知道参数city应该写"杭州"还是写"杭州市"还是写"hangzhou",也不知道这个工具能不能查历史天气、能不能查未来预报。这些信息全靠描述文本传递。

我写工具描述的习惯是固定三段:做什么、什么时候用、参数怎么写。举个例子:

from langchain_core.tools import tool @tool def get_weather(city: str) -> str: """查询指定城市当前时刻的实时天气情况。 当用户询问某地的天气、气温、是否下雨、是否需要带伞时使用此工具。 不要用它查询历史天气或未来多日预报,那类问题没有对应数据。 Args: city: 城市中文名称,不带"市"字后缀。例如用户说"杭州市"时, 应传入"杭州";用户说"北京"时直接传"北京"。 """ # 实际实现,这里用占位数据 return f"{city} 当前晴,气温 23 摄氏度,东南风 2 级"

对比一下,如果把描述写成"查询天气",模型在遇到"明天杭州会下雨吗"这种问题时,很可能照样调用它,然后拿着今天的天气回答你,而且回答得理直气壮。这不是模型蠢,是你没告诉它边界在哪。

另一个容易忽略的点是工具名。函数名会直接作为工具名传给模型,所以别起func1、helper这种名字,用get_weather、search_docs、calc_expression这种自带语义的。名字本身就是最强的信号。

还有一个小技巧:如果参数是枚举型的,不要用 Python 的Literal就完事,在描述里把可选值列一遍。本地小模型对类型约束的敏感度远低于对大段文本的敏感度。

4.2 从 tool_calls 到真实执行的分发

绑定的写法很简单,一行:

tools = [get_weather] llm_with_tools = llm.bind_tools(tools)

调用之后返回的AIMessage上会有一个tool_calls列表。要注意的是,这个列表非空不代表模型一定要执行,有些模型会同时给你一段content文本和一组tool_calls,文本里写着"我需要调用天气工具",tool_calls里也给了参数。这时候正确的做法是执行工具,然后把文本作为中间过程展示给用户。

分发逻辑我一般这么写:

tool_map = {t.name: t for t in tools} def run_tool_calls(msg): results = [] for call in msg.tool_calls: name = call.get("name") args = call.get("args") or {} call_id = call.get("id") or f"call_{uuid.uuid4().hex[:8]}" fn = tool_map.get(name) if fn is None: results.append(ToolMessage( content=f"工具 {name} 不存在,可用工具:{list(tool_map)}", tool_call_id=call_id, )) continue try: output = fn.invoke(args) except Exception as e: output = f"工具执行失败:{type(e).__name__}: {e}" results.append(ToolMessage(content=str(output), tool_call_id=call_id)) return results

这段代码里有三个防御点,每一个都对应过我踩过的坑。

第一,call_id兜底。部分本地模型返回的tool_calls里id字段是空的,或者多个调用共用一个 id。LangChain 在组装消息历史时依赖这个 id 做匹配,缺了或者重复了,后续推理会报错,而且报错信息往往指向消息格式不合法,跟你实际的问题(模型没给 id)隔着十万八千里。自己补一个 uuid 是最省事的做法。

第二,工具名不存在时的兜底。模型有时候会凭空捏造一个工具名,尤其是它见过类似工具但当前没绑定时。这时候直接抛异常会让整个流程崩掉,更聪明的做法是把可用工具列表返回给它,让它自己纠正。实测这个兜底能救回相当一部分幻觉调用。

第三,异常转文本。工具内部报错(网络超时、参数格式不对、除零)如果直接往上抛,用户看到的是堆栈,体验很差。把异常转成一段文本回填给模型,模型往往能自己调整参数重试,或者至少给用户一个像样的解释。

4.3 结果回填与二次推理

拿到工具结果之后,要把AIMessage和若干条ToolMessage依次塞回消息列表,再调一次模型。

from langchain_core.messages import HumanMessage, ToolMessage messages = [HumanMessage("杭州现在适合出门吗")] resp = llm_with_tools.invoke(messages) if resp.tool_calls: messages.append(resp) messages.extend(run_tool_calls(resp)) final = llm_with_tools.invoke(messages) else: final = resp

这里注意消息顺序:AIMessage必须在ToolMessage之前,因为ToolMessage是对前面那次调用的响应。顺序错了,接口会直接拒绝请求。我在第一次写的时候把结果消息 append 到了前面,调了半天以为是模型不支持工具,实际上是我自己把格式搞坏了。

还有一个隐含约束:ToolMessage的数量和顺序,最好跟tool_calls一一对应。有些实现允许缺失,有些不允许,跨模型时行为不一致。统一按顺序全量回填最保险。

多轮工具调用怎么办?用一个循环包起来,设一个最大轮次上限(我一般设 5),超过就打断并要求模型直接给出答案。不加限制的话,某些小模型会陷入"调用工具—拿到结果—觉得不够—再调用"的循环,尤其是工具返回的内容跟它预期不符的时候。

for step in range(5): resp = llm_with_tools.invoke(messages) if not resp.tool_calls: break messages.append(resp) messages.extend(run_tool_calls(resp)) else: messages.append(HumanMessage("请直接基于已有信息回答,不要再调用工具")) resp = llm.invoke(messages)

for...else这个写法在这里特别好用,循环正常跑完(也就是一直有工具调用)才会进 else 分支,中途 break 就跳过。

5. 本地模型做工具调用的失败模式与补救

5.1 五种高频翻车现场

工具调用这套机制在云端大模型上已经很稳了,但换成本地跑的 7B、8B 模型,成功率会明显下降。我整理了一下自己遇到过的失败模式,大概能归成五类。

失败模式典型症状根本原因
该调不调明明需要实时数据,却直接用训练知识回答工具描述跟用户问法对不上
不该调乱调闲聊也触发工具,参数还是瞎编的描述里没写清禁用场景
参数残缺args里少了必填字段,或者值是空串描述里没给参数示例
参数结构错args是字符串而不是字典模型把 JSON 当文本输出
写了不调content里有 JSON,tool_calls是空的模型不支持原生工具调用

前三种靠改描述能解决大半,第四种需要加一层解析兜底,第五种最麻烦,得看模型本身支不支持。

第四种的具体处理是这样:拿到tool_calls之后,先检查args的类型,如果是字符串,尝试json.loads;解析失败就再试一次,把首尾的杂字符剥掉(模型有时候会在 JSON 前后加一句"参数如下:")。这个兜底逻辑不长,但能显著提升小模型的可用性。

import json def normalize_args(raw): if isinstance(raw, dict): return raw if isinstance(raw, str): s = raw.strip() # 剥掉可能包裹的代码围栏 if s.startswith("```"): s = s.strip("`") if s.lower().startswith("json"): s = s[4:] start, end = s.find("{"), s.rfind("}") if start != -1 and end > start: try: return json.loads(s[start:end + 1]) except json.JSONDecodeError: return {} return {}

第五种情况需要单独处理。有些本地模型不支持原生的工具调用协议,bind_tools之后它只是把工具定义拼进提示词里,然后期望模型自己输出结构化的 JSON。这时候tool_calls永远是空的,但你会在content里看到一段类似{"name": "get_weather", "arguments": {"city": "杭州"}}的文本。

遇到这种情况,我的做法是加一条"文本兜底解析"分支:tool_calls为空时,扫一遍content,如果发现里面有一段能解析成带name和arguments(或args)字段的 JSON,就当成一次工具调用来处理。这条兜底不能无条件开启,否则正常回答里出现的 JSON 示例也会被误判,所以要加一些限定条件,比如 JSON 里必须有name字段,且name的取值必须在已注册的工具名集合里。

5.2 参数、提示词、兜底校验三条补救线

第一层是参数层面。除了上面说的normalize_args,我还会在工具函数内部做二次校验。比如city传进来是空串,直接在工具里返回"城市名不能为空,请从用户问题中提取",而不是硬跑然后返回一个错误结果。让模型看到明确的错误原因,它下一轮改对的概率很高。

第二层是提示词层面。在系统提示里加一句约束,效果比想象中好:

你可以在需要实时数据时调用工具。调用前先用一句话说明你要做什么,调用后基于返回结果作答,不要编造工具没有返回的信息。如果工具返回的内容不足以回答,直接告诉用户缺什么。

这段话里最关键的是最后半句。小模型有个通病:工具返回了三行数据,它能给你编出十行结论。明确告诉它"不足就说不足",能压掉大部分幻觉。

第三层是兜底校验。在把最终答案返回给用户之前,检查一下:如果这轮发生了工具调用,最终答案里是否出现了工具返回结果中的关键数字或字符串?如果完全没有,大概率是模型无视了工具结果。这种情况下我会选择性地再跑一轮,或者至少在界面上标注"本回答未使用工具返回数据",让用户心里有数。

这三层加起来代码量不小,但每一层都对应着真实遇到过的问题。做本地模型应用有个心理准备:你写的代码里,真正调用模型的部分可能只占三成,剩下七成都在做容错。

6. 组装成一个能长期用的小工程

6.1 模块划分与启动流程

零散脚本跑通之后,下一步是把它变成一个能反复用的东西。我的划分方式大概是这样四层:

  • 配置层:模型名、地址、温度、上下文长度、最大工具轮次,全部集中到一个字典或者配置文件里,不要在代码里散落魔法数字。
  • 工具层:每个工具一个文件,函数体内只做纯粹的业务逻辑,描述文本写在 docstring 里,不掺任何打印语句。
  • 编排层:负责消息组装、工具分发、轮次控制、异常兜底。这一层是核心,也是最容易写乱的地方,建议把"跑一轮对话"抽成一个函数,输入是消息列表,输出是最终消息加事件流。
  • 展示层:就是前面那个TerminalRenderer,只关心怎么把事件画出来,不关心业务。

编排层和展示层之间用事件解耦,是这个设计里最关键的一步。编排层不直接 print,而是往外抛事件:on_token、on_tool_start、on_tool_end、on_finish。展示层订阅这些事件决定怎么画。这样做的好处是,同一套编排逻辑可以接终端渲染,也可以接 Web 的 SSE,还能接日志文件,不用改核心代码。

def chat_turn(user_input, messages, llm, tools, events): messages.append(HumanMessage(user_input)) for step in range(5): collected = "" tool_calls = [] for chunk in llm.bind_tools(tools).stream(messages): if chunk.content: collected += chunk.content events.on_token(chunk.content) if getattr(chunk, "tool_calls", None): tool_calls.extend(chunk.tool_calls) if not tool_calls: messages.append(AIMessage(content=collected)) events.on_finish(collected) return messages ai_msg = AIMessage(content=collected, tool_calls=tool_calls) messages.append(ai_msg) for call in tool_calls: events.on_tool_start(call["name"], call.get("args", {})) results = run_tool_calls(ai_msg) for r in results: events.on_tool_end(r.content) messages.extend(results) events.on_finish("已达到工具调用轮次上限,停止。") return messages

这段代码有一个细节值得说:流式接收时chunk.tool_calls是增量的,直接extend会拿到一堆半截的调用对象。稳妥的做法是等流结束后统一从聚合结果里取,或者用 LangChain 提供的AIMessageChunk加法运算把它累加起来。我上面为了展示流程写的是简化版,实际用的时候建议用累加而不是 extend。

6.2 我在实际使用中踩出来的几条经验

渲染器要有"降级模式"。不是所有终端都支持富文本,重定向到文件、通过管道传给别的程序、在某些简易终端里跑,Live的刷新会变成一堆控制字符。加一个--plain开关,检测到非 TTY 环境就自动切成纯文本逐段输出,这个功能上线之后救过我好几次自动化脚本。

工具执行要有独立超时。工具里如果涉及网络请求,一定单独设超时,不要依赖模型调用的超时。一个工具卡住三十秒,用户会以为整个程序死了。我一般给工具包一层concurrent.futures的ThreadPoolExecutor,超过 5 秒直接返回"工具超时",让模型基于已有信息作答。

上下文要主动裁剪,别指望自动。多轮对话加上工具返回的长文本,上下文涨得比想象中快。我的做法是保留最近 N 轮完整对话,更早的轮次只保留用户提问和最终回答,把中间的工具调用过程丢掉。工具调用的中间结果对后续对话的价值通常很低,但占的位置很大。

评分和对比一定要做。改完提示词或者换了模型之后,别凭感觉判断"好像变好了"。准备二十条固定的测试问题,覆盖直接回答、单工具调用、多工具调用、需要澄清、明显不需要工具这五类,每次改完跑一遍,看通过率。我最初跳过这一步,结果某次调整描述之后,单工具调用成功率涨了,但"该不调的时候乱调"的比例翻了三倍,上线之后被用户吐槽得很难受。

还有一个不那么技术但很实在的体会:本地模型的工具调用能力,跟模型参数量不是线性关系。有些小模型在工具调用上的表现明显好于同尺寸的其他模型,因为它们训练时专门做过这部分对齐。选模型的时候,别只看排行榜上的通用分数,拿你自己的工具集和测试问题实际跑一遍,结果往往跟预期不一样。我现在的习惯是同时挂两个模型,一个负责对话,一个专门负责工具调用决策,后者只输出"要不要调、调哪个、参数是什么",前面那个负责组织语言。这个拆分看起来笨,但把两件事解耦之后,各自的调优难度都降下来了,而且渲染层完全不用改。

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

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

立即咨询