首页
学习
活动
专区
圈层
工具
发布
社区首页 >专栏 >我把Claude和DeepSeek接进了同一个工作流,不同任务用不同模型

我把Claude和DeepSeek接进了同一个工作流,不同任务用不同模型

作者头像
用户11081884
发布2026-07-20 20:38:53
发布2026-07-20 20:38:53
410
举报

前几天写一个项目,遇到一个很典型的情况:我需要DeepSeek帮我写一段Python爬虫代码,写完又要切到Claude让它帮我把技术方案整理成文档发给团队。两个网页来回切,复制粘贴,烦得要死。

后来我发现这种“切来切去”的操作其实有规律可循。写代码用DeepSeek,写长文档用Claude,搜索信息用通义千问。每种任务该用哪个模型,我自己脑子里有一套判断逻辑。既然是固定的规则,那就可以用代码自动化。

于是我写了一个路由层,把几个大模型的API接到一起。你只管发消息,程序自动判断你的意图,把请求转给最合适的模型。用起来就像只有一个超级助手,背后其实有好几个模型在协作。

方案

核心是一个ModelRouter类,它做三件事:接收用户输入、判断任务类型、把请求转发给对应的模型。

任务分类这块,我没有用大模型来做(太浪费token了),而是用关键词和规则匹配。比如消息里出现“写代码”、“实现”、“函数”这类词,就归为代码任务;出现“写文章”、“总结”、“文档”就归为长文任务。规则覆盖不到的情况,用一个默认模型兜底。

所有模型的调用接口统一封装,对外看起来就是一个chat()方法,内部的路由逻辑对用户完全透明。

完整代码

代码语言:javascript
复制
pip install openai httpx

httpx用来做HTTP请求,有些模型的API用openai库不太方便,直接发请求更稳。

代码语言:javascript
复制
# 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()

运行效果

程序跑起来之后,你正常输入就行,路由层会自己判断该用哪个模型:

代码语言:javascript
复制
你:帮我用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,搜索走通义。你只管问问题,选模型的事交给代码。

“无他,惟手熟尔”!有需要的用起来!

本文参与 腾讯云自媒体同步曝光计划,分享自微信公众号。
原始发表:2026-07-06,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 Nicholas与Pypi 微信公众号,前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体同步曝光计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
目录
  • 方案
  • 完整代码
  • 运行效果
  • 路由规则怎么调
  • 实际效果
领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档