
前几天写一个项目,遇到一个很典型的情况:我需要DeepSeek帮我写一段Python爬虫代码,写完又要切到Claude让它帮我把技术方案整理成文档发给团队。两个网页来回切,复制粘贴,烦得要死。
后来我发现这种“切来切去”的操作其实有规律可循。写代码用DeepSeek,写长文档用Claude,搜索信息用通义千问。每种任务该用哪个模型,我自己脑子里有一套判断逻辑。既然是固定的规则,那就可以用代码自动化。
于是我写了一个路由层,把几个大模型的API接到一起。你只管发消息,程序自动判断你的意图,把请求转给最合适的模型。用起来就像只有一个超级助手,背后其实有好几个模型在协作。
核心是一个ModelRouter类,它做三件事:接收用户输入、判断任务类型、把请求转发给对应的模型。
任务分类这块,我没有用大模型来做(太浪费token了),而是用关键词和规则匹配。比如消息里出现“写代码”、“实现”、“函数”这类词,就归为代码任务;出现“写文章”、“总结”、“文档”就归为长文任务。规则覆盖不到的情况,用一个默认模型兜底。
所有模型的调用接口统一封装,对外看起来就是一个chat()方法,内部的路由逻辑对用户完全透明。
pip install openai httpxhttpx用来做HTTP请求,有些模型的API用openai库不太方便,直接发请求更稳。
# pip install openai httpx
# 所有API key需要自己去各平台申请,下面用占位符
import re
import time
from dataclasses import dataclass
from openai import OpenAI
# ============ 第一步:定义各模型配置 ============
@dataclass
class ModelConfig:
name: str
base_url: str
model: str
api_key: str
max_tokens: int = 4096
# 这里配置你有的模型
MODELS = {
"deepseek": ModelConfig(
name="DeepSeek",
base_url="https://api.deepseek.com/v1",
model="deepseek-chat",
api_key="YOUR_DEEPSEEK_API_KEY",
),
"claude": ModelConfig(
name="Claude",
base_url="https://api.anthropic.com/v1",
model="claude-sonnet-4-20250514",
api_key="YOUR_CLAUDE_API_KEY",
max_tokens=8192, # Claude长文能力强,给多点token
),
"qwen": ModelConfig(
name="通义千问",
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
model="qwen-plus",
api_key="YOUR_QWEN_API_KEY",
),
}
# ============ 第二步:统一模型调用封装 ============
class ModelClient:
"""统一的模型调用客户端,屏蔽不同API的差异"""
def __init__(self, config: ModelConfig):
self.config = config
# Claude的API格式和其他的不太一样,需要单独处理
self.is_claude = "anthropic"inconfig.base_url
if not self.is_claude:
self.client = OpenAI(
api_key=config.api_key,
base_url=config.base_url,
)
def chat(self, messages: list, temperature: float = 0.7) ->str:
"""
发送对话请求,返回文本回复
messages格式统一用 [{"role": "user", "content": "..."}] 这种
"""
if self.is_claude:
return self._chat_claude(messages, temperature)
else:
return self._chat_openai(messages, temperature)
def _chat_openai(self, messages: list, temperature: float) ->str:
"""兼容OpenAI格式的调用(DeepSeek、通义等都走这个)"""
response = self.client.chat.completions.create(
model=self.config.model,
messages=messages,
temperature=temperature,
max_tokens=self.config.max_tokens,
)
return response.choices[0].message.content
def _chat_claude(self, messages: list, temperature: float) ->str:
"""
Claude的API格式和OpenAI不一样
需要用httpx直接发请求
"""
import httpx
# 把OpenAI格式的messages转成Claude格式
system_msg = ""
claude_messages = []
for msg in messages:
if msg["role"] == "system":
system_msg = msg["content"]
else:
claude_messages.append({
"role": msg["role"],
"content": msg["content"],
})
headers = {
"x-api-key": self.config.api_key,
"anthropic-version": "2023-06-01",
"content-type": "application/json",
}
body = {
"model": self.config.model,
"max_tokens": self.config.max_tokens,
"temperature": temperature,
"messages": claude_messages,
}
if system_msg:
body["system"] = system_msg
resp = httpx.post(
f"{self.config.base_url}/messages",
headers=headers,
json=body,
timeout=120,
)
resp.raise_for_status()
data = resp.json()
return data["content"][0]["text"]
# ============ 第三步:任务分类器 ============
class TaskClassifier:
"""
根据用户输入判断任务类型
用关键词匹配,不消耗额外token
"""
# 每种任务类型对应的关键词
RULES = {
"code": {
"keywords": [
"写代码", "代码", "函数", "类", "实现", "编程",
"debug", "报错", "bug", "算法", "脚本",
"python", "javascript", "typescript", "java",
"API", "接口", "爬虫", "数据库", "SQL",
"def ", "class ", "import ", "pip",
"重构", "优化代码", "单元测试",
],
"weight": 1.0,
},
"long_text": {
"keywords": [
"写文章", "文章", "文档", "报告", "总结",
"方案", "邮件", "文案", "翻译", "润色",
"周报", "月报", "述职", "简历",
"详细说明", "展开讲讲", "写一篇",
],
"weight": 1.0,
},
"search": {
"keywords": [
"什么是", "有哪些", "怎么理解", "解释一下",
"对比", "区别", "优缺点", "推荐",
"最新的", "2024", "2025", "趋势",
"搜索", "查找", "了解",
],
"weight": 0.8,
},
"reasoning": {
"keywords": [
"分析", "推理", "逻辑", "为什么",
"数学", "计算", "概率", "策略",
"决策", "方案对比", "评估",
],
"weight": 0.9,
},
}
# 任务类型到模型的映射
TASK_MODEL_MAP = {
"code": "deepseek", # 代码任务用DeepSeek
"long_text": "claude", # 长文本用Claude
"search": "qwen", # 搜索/知识类用通义
"reasoning": "deepseek", # 推理也交给DeepSeek,逻辑能力强
"general": "deepseek", # 兜底默认用DeepSeek
}
@classmethod
def classify(cls, user_input: str) ->tuple:
"""
返回 (任务类型, 目标模型, 置信度)
置信度是关键词命中数量,用来在多个类型都匹配时取最高的
"""
scores = {}
input_lower = user_input.lower()
for task_type, rule in cls.RULES.items():
hits = sum(1forkwinrule["keywords"] ifkw.lower() ininput_lower)
# 加权得分
scores[task_type] = hits*rule["weight"]
# 取得分最高的类型
best_type = max(scores, key=scores.get)
best_score = scores[best_type]
# 如果没有任何关键词命中,归为通用类型
if best_score == 0:
return"general", cls.TASK_MODEL_MAP["general"], 0
target_model = cls.TASK_MODEL_MAP.get(best_type, "deepseek")
return best_type, target_model, best_score
# ============ 第四步:路由核心 ============
class ModelRouter:
"""
模型路由器:接收消息 -> 分类 -> 转发 -> 返回结果
用起来就像在和一个助手对话
"""
def __init__(self):
# 初始化所有模型的客户端
self.clients = {}
for key, config in MODELS.items():
self.clients[key] = ModelClient(config)
self.classifier = TaskClassifier()
# 对话历史,每个模型单独维护一份
self.histories = {key: [] forkeyinMODELS}
def chat(self, user_input: str) ->dict:
"""
发送消息,自动路由到合适的模型
返回值包含:回复内容、使用的模型、任务类型
"""
# 分类任务
task_type, target_model, confidence = self.classifier.classify(user_input)
model_name = MODELS[target_model].name
print(f" [路由] 任务类型: {task_type} → 模型: {model_name} (置信度: {confidence})")
# 构建消息(带对话历史)
messages = [{"role": "system", "content": "你是一个专业的AI助手,请认真回答用户的问题。"}]
# 只带最近5轮对话,避免token超标
recent_history = self.histories[target_model][-10:]
messages.extend(recent_history)
messages.append({"role": "user", "content": user_input})
# 调用模型
start = time.time()
try:
reply = self.clients[target_model].chat(messages)
elapsed = time.time() -start
# 更新对话历史
self.histories[target_model].append({"role": "user", "content": user_input})
self.histories[target_model].append({"role": "assistant", "content": reply})
return {
"reply": reply,
"model": model_name,
"task_type": task_type,
"time": round(elapsed, 2),
"error": None,
}
except Exception as e:
elapsed = time.time() -start
return {
"reply": "",
"model": model_name,
"task_type": task_type,
"time": round(elapsed, 2),
"error": str(e),
}
def force_model(self, user_input: str, model_key: str) ->dict:
"""
手动指定模型,不走路由
有时候你明确知道想用哪个模型
"""
if model_key not in self.clients:
return {"reply": "", "error": f"未知模型: {model_key}"}
model_name = MODELS[model_key].name
messages = [
{"role": "system", "content": "你是一个专业的AI助手。"},
{"role": "user", "content": user_input},
]
start = time.time()
reply = self.clients[model_key].chat(messages)
elapsed = time.time() -start
return {
"reply": reply,
"model": model_name,
"task_type": "manual",
"time": round(elapsed, 2),
"error": None,
}
# ============ 第五步:交互入口 ============
def main():
router = ModelRouter()
print("="*50)
print("多模型路由助手")
print("自动根据你的问题类型选择最合适的模型")
print("特殊命令:")
print(" /deepseek <消息> - 强制用DeepSeek")
print(" /claude <消息> - 强制用Claude")
print(" /qwen <消息> - 强制用通义千问")
print(" quit - 退出")
print("="*50)
while True:
user_input = input("\n你:").strip()
if not user_input:
continue
if user_input.lower() in ("quit", "exit", "q"):
break
# 检查是否手动指定了模型
force_match = re.match(r"^/(\w+)\s+(.+)", user_input, re.DOTALL)
if force_match:
model_key = force_match.group(1)
actual_msg = force_match.group(2)
result = router.force_model(actual_msg, model_key)
else:
result = router.chat(user_input)
if result["error"]:
print(f"\n出错: {result['error']}")
else:
print(f"\n[{result['model']}] ({result['task_type']}, {result['time']}s)")
print(f"{result['reply']}")
if __name__ == "__main__":
main()程序跑起来之后,你正常输入就行,路由层会自己判断该用哪个模型:
你:帮我用Python写一个JWT认证的中间件
[路由] 任务类型: code → 模型: DeepSeek (置信度: 3.0)
[DeepSeek] (code, 4.2s)
好的,这里给你一个基于FastAPI的JWT认证中间件...
(DeepSeek写代码确实靠谱,结构清晰,能直接跑)
你:帮我把上面的方案整理成技术文档,发给团队评审用
[路由] 任务类型: long_text → 模型: Claude (置信度: 2.0)
[Claude] (long_text, 6.8s)
# JWT认证中间件技术方案
## 1. 背景与目标
...
(Claude生成的文档格式漂亮,措辞专业,直接能拿去用)
你:通义千问最新的模型有哪些版本?
[路由] 任务类型: search → 模型: 通义千问 (置信度: 2.4)
[通义千问] (search, 2.1s)
通义千问目前有以下几个主要版本...
(关于自家产品的信息,通义回答得最准)用 /deepseek、/claude、/qwen 可以手动指定模型,跳过路由。有时候路由判断不准或者你就是想指定某个模型,用这个就行。
TaskClassifier里面的RULES字典是可以随便改的。你可以根据自己的使用习惯调整关键词和权重。比如你经常让通义千问写代码,就把code的映射从deepseek改成qwen。
关键词也可以加。你平时问什么类型的问题多,就往对应的keywords列表里加词。这个列表越完善,路由的准确率越高。
还有一个思路是用小模型做分类。比如调一次DeepSeek的API,让它判断“这条消息属于什么类型的任务”,返回一个JSON。这样做更准,但每次请求会多花一点时间和token。适合对路由准确率要求高的场景。
一个路由器把多个大模型拼成一个“全能助手”,代码任务走DeepSeek,长文任务走Claude,搜索走通义。你只管问问题,选模型的事交给代码。
“无他,惟手熟尔”!有需要的用起来!
本文分享自 Nicholas与Pypi 微信公众号,前往查看
如有侵权,请联系 cloudcommunity@tencent.com 删除。
本文参与 腾讯云自媒体同步曝光计划 ,欢迎热爱写作的你一起参与!