Ollama 多模型协作与 Agent 工作流

单个模型不必包打天下：让小模型做分流、专业模型做任务、联网模型查资料，用结构化输出当调度协议，就能组合出又快又省的智能应用。

本章节实现一个多模型路由系统，并把它升级成带联网能力的 Agent 工作流。

架构设计：小模型路由 + 专家执行

核心思路是分工：判断"用户想干什么"这件事很简单，用最小的模型即可；真正干活的任务再交给对口的专业模型。

多模型路由架构：路由器分类，三个专家模型执行

这样设计有三笔账：小模型路由只要几十毫秒，大模型不再处理低价值请求；各专家模型的参数可以按需独立调整；新任务类型只需加一个分支，不影响已有链路。

实现一：结构化输出做意图路由

路由的关键是让分类结果"机器可读"，结构化输出在这里充当调度协议。

实例

# 文件路径：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)

$ python router.py
coding | 询问 Python 列表去重的编程方法
search | 询问当天的最新新闻，需要联网
chat   | 闲聊类请求，直接回答即可

实现二：三个专家处理器

每条路径一个处理器函数，统一签名方便调度。

实例

# 追加到 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 混合编排

每条路径都可以加一个"云端升级档"：本地模型搞不定的任务，同一个代码结构换个模型名即可。

实例

# 追加到 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 模型暂不支持结构化输出，所以路由器必须由本地模型承担，这也正好符合"小模型做路由"的设计。

成本与延迟的综合评估

多模型策略的账要从三个维度一起算：

维度	本地小模型	本地大模型	云端模型
首字延迟	最低（毫秒级加载）	中等	含网络往返
边际成本	约等于电费	电费 + 显存占用	按用量计费
回答质量	简单任务够用	接近旗舰	旗舰级
并发能力	受本机限制	受显存限制	弹性最好

由此得到一条实用的基线策略：路由器和轻任务用本地小模型打底；质量敏感任务交给本地大模型；重负载或超规格任务切换云端；三类模型的模型名全部走配置文件，随退役公告或硬件变化随时调整。

返回顶部

菜鸟教程