Ollama 多模型协作与 Agent 工作流
单个模型不必包打天下：让小模型做分流、专业模型做任务、联网模型查资料，用结构化输出当调度协议，就能组合出又快又省的智能应用。
本章节实现一个多模型路由系统，并把它升级成带联网能力的 Agent 工作流。
架构设计：小模型路由 + 专家执行
核心思路是分工：判断"用户想干什么"这件事很简单，用最小的模型即可；真正干活的任务再交给对口的专业模型。
这样设计有三笔账：小模型路由只要几十毫秒，大模型不再处理低价值请求；各专家模型的参数可以按需独立调整；新任务类型只需加一个分支，不影响已有链路。
实现一：结构化输出做意图路由
路由的关键是让分类结果"机器可读"，结构化输出在这里充当调度协议。
实例
python
# 文件路径：router.py
from ollama import chat
from pydantic import BaseModel, Field
# 路由结果的结构化定义
class Route(BaseModel):
category: str = Field(description="问题类别：chat / coding / search")
reason: str = Field(description="一句话分类依据")
def classify(question: str) -> Route:
response = chat(
model='qwen3.5:4b', # 路由用小模型即可
messages=[{'role': 'user', 'content': question}],
format=Route.model_json_schema(),
options={'temperature': 0}, # 分类要稳定
)
return Route.model_validate_json(response.message.content)
# 快速自测
for q in ['Python 列表怎么去重？', '今天有什么 AI 新闻？', '讲个笑话']:
r = classify(q)
print(r.category, '|', r.reason)bash
$ python router.py
coding | 询问 Python 列表去重的编程方法
search | 询问当天的最新新闻，需要联网
chat | 闲聊类请求，直接回答即可实现二：三个专家处理器
每条路径一个处理器函数，统一签名方便调度。
实例
python
# 追加到 router.py
from ollama import chat, web_search, web_fetch
def handle_chat(question):
"""日常问答：小模型直接回答"""
r = chat(model='qwen3.5:4b',
messages=[{'role': 'user', 'content': question}])
return r.message.content
def handle_coding(question):
"""编程任务：交给专用代码模型"""
r = chat(model='runoob-coder',
messages=[{'role': 'user', 'content': question}],
options={'temperature': 0.2, 'num_ctx': 65536})
return r.message.content
def handle_search(question):
"""联网任务：小模型 + 搜索工具的 Agent loop"""
tools = {'web_search': web_search, 'web_fetch': web_fetch}
messages = [{'role': 'user', 'content': question}]
while True:
r = chat(model='qwen3.5:4b', messages=messages,
tools=[web_search, web_fetch])
messages.append(r.message)
if not r.message.tool_calls:
return r.message.content
for call in r.message.tool_calls:
func = tools.get(call.function.name)
# 工具结果截断，保护上下文窗口
result = str(func(**call.function.arguments))[:2000]
messages.append({'role': 'tool',
'tool_name': call.function.name,
'content': result})
HANDLERS = {'chat': handle_chat,
'coding': handle_coding,
'search': handle_search}
def ask(question):
route = classify(question)
print(f'[路由 -> {route.category}] {route.reason}')
return HANDLERS[route.category](question)路由器自身不产生答案，所以它的幻觉风险被限制在"分错类"这一件事上；分错的代价也只是多走一次纠正，不会污染最终回答。
实现三：本地与 Cloud 混合编排
每条路径都可以加一个"云端升级档"：本地模型搞不定的任务，同一个代码结构换个模型名即可。
实例
python
# 追加到 router.py
def ask_with_fallback(question, force_cloud=False):
"""带云端兜底的调度：本地先行，复杂任务升级云端"""
route = classify(question).category
if force_cloud or route == 'coding':
# 编码等重任务直接走云端旗舰规格
model = 'qwen3.5:cloud'
else:
model = 'qwen3.5:4b'
print(f'[路由 -> {route}] 使用模型 {model}')
r = chat(model=model,
messages=[{'role': 'user', 'content': question}])
return r.message.content注意一个能力差异：Cloud 模型暂不支持结构化输出，所以路由器必须由本地模型承担，这也正好符合"小模型做路由"的设计。
成本与延迟的综合评估
多模型策略的账要从三个维度一起算：
| 维度 | 本地小模型 | 本地大模型 | 云端模型 |
|---|---|---|---|
| 首字延迟 | 最低（毫秒级加载） | 中等 | 含网络往返 |
| 边际成本 | 约等于电费 | 电费 + 显存占用 | 按用量计费 |
| 回答质量 | 简单任务够用 | 接近旗舰 | 旗舰级 |
| 并发能力 | 受本机限制 | 受显存限制 | 弹性最好 |
由此得到一条实用的基线策略：路由器和轻任务用本地小模型打底；质量敏感任务交给本地大模型；重负载或超规格任务切换云端；三类模型的模型名全部走配置文件，随退役公告或硬件变化随时调整。
AI 思考中...