多模型系统架构设计：4种常见模式解析与Python实践

在云栈社区的技术讨论中，多模型系统一直是个热门话题。单模型系统足够简单，多模型系统则具备更强的能力。但如果将二者放在一起比较，真正困难的地方不是模型怎么选，而是如何设计一套让多个模型高效协同工作的架构。

所以，多模型系统的重点并非模型数量，而是在合适的时机把合适的模型分配给合适的任务。

架构模式

多数场景都可以归纳到以下几种架构模式之中。我们先看一张对比总表。

记住一条原则：优先选择能够满足需求的最简单方案。架构一旦变复杂，后续的维护成本只会不断增加。

顺序架构

顺序架构把任务拆成多个阶段，每个阶段交给一个擅长对应工作的模型处理。

模式 1：Pipeline（流水线）

Pipeline 模式中，一个模型的输出直接作为下一个模型的输入。

class ModelPipeline:
    def __init__(self):
        self.models = [
            {"model": "qwen3-1.7b", "task": "classify"},
            {"model": "qwen3-8b", "task": "extract"},
            {"model": "qwen3-32b", "task": "reason"},
        ]

    def process(self, input: str) -> str:
        current = input
        for model_config in self.models:
            current = self.call_model(
                model_config["model"],
                self.create_prompt(model_config["task"], current)
            )
        return current

串行执行的问题在于延迟会层层累积。三个模型依次调用，整体延迟差不多就是单次调用的三倍。只有每一步确实需要不同模型的能力时，才值得采用这种模式。

模式 2：Router（路由）

Router 模式先判断任务类型，再把请求发送给对应的专用模型。

class ModelRouter:
    def __init__(self):
        self.classifier = "qwen2.5-1.5b"
        self.specialists = {
            "code": "qwen2.5-coder-7b",
            "math": "qwen2.5-32b",
            "creative": "claude-sonnet-4",
            "general": "qwen2.5-7b",
        }

    def route(self, prompt: str) -> str:
        task_type = self.classify(prompt)
        model = self.specialists.get(task_type, self.specialists["general"])
        return self.call_model(model, prompt)

分类器决定了整个路由流程的质量。一旦分类出错，请求就会落到错误的模型上，最终输出自然也会受到影响。如果分类标准足够清晰，一个规模较小的分类模型通常已经足够。

并行架构

并行架构适用于彼此独立的任务，可以同时交给多个模型处理。

模式 1：Fan-Out（扇出）

Fan-Out 将同一个 Prompt 并发发送给多个模型。

import asyncio

class ModelFanOut:
    def __init__(self):
        self.models = [
            "qwen2.5-7b",
            "qwen2.5-32b",
            "claude-sonnet-4",
        ]

    async def process(self, prompt: str) -> list[str]:
        tasks = [self.call_model(model, prompt) for model in self.models]
        return await asyncio.gather(*tasks)

这种模式适合比较不同模型的输出、进行 A/B 测试，或者在多个候选答案中挑选最佳结果。成本确实会增加，但对于关键决策而言，这点额外开销通常是值得的。

模式 2：Voting（投票）

Voting 让多个模型共同决策，以投票结果作为最终输出。

class ModelVoting:
    def __init__(self):
        self.models = [
            "qwen3-8b",
            "qwen3-32b",
            "claude-sonnet-4",
        ]

    def vote(self, prompt: str) -> str:
        responses = [self.call_model(model, prompt) for model in self.models]
        from collections import Counter
        votes = Counter(responses)
        return votes.most_common(1)[0][0]

多数投票在分类任务上很好用。换成生成任务，处理方式就没这么简单了，因为我们需要比较的是语义一致性，而不是字符串是否完全相同。

分层架构

分层架构让不同层级的模型各自承担不同职责。

模式 1：Planner-Executor（规划-执行）

Planner-Executor 由能力更强的模型负责规划，再把具体工作交给较小的模型执行。

class PlannerExecutor:
    def __init__(self):
        self.planner = "qwen3-32b"
        self.executors = {
            "code": "qwen2.5-coder-7b",
            "search": "qwen3-8b",
            "math": "qwen3-8b",
        }

    def process(self, task: str) -> str:
        plan = self.call_model(self.planner, f"Plan: {task}")
        results = []
        for step in self.parse_plan(plan):
            executor = self.executors.get(step["type"], "qwen2.5-7b")
            result = self.call_model(executor, step["prompt"])
            results.append(result)
        return self.call_model(self.planner, f"Synthesize: {results}")

规划器承担主要的推理工作，执行器负责完成各个具体步骤。当规划阶段成本较高而执行阶段相对轻量时，这种架构通常比较合适。

模式 2：Supervisor-Worker（监督-工人）

Supervisor-Worker 中，监督者负责分配任务，并审核各个 Worker 返回的结果。

class SupervisorWorker:
    def __init__(self):
        self.supervisor = "qwen3-32b"
        self.workers = ["qwen3-8b", "qwen2.5-coder-7b"]

    def process(self, task: str) -> str:
        assignments = self.call_model(self.supervisor, f"Assign: {task}")
        results = []
        for assignment in self.parse_assignments(assignments):
            result = self.call_model(
                assignment["worker"], assignment["task"]
            )
            results.append(result)
        return self.call_model(self.supervisor, f"Review: {results}")

监督者同时承担规划、任务分配和结果审核三项工作，因此很容易成为整个系统的瓶颈。它的响应速度如果跟不上，整个系统都会受到影响。

集成架构

集成架构适合用于关键决策，通过多个模型共同参与判断，提高结果的可靠性。

模式 1：Weighted Ensemble（加权集成）

Weighted Ensemble 会对各个模型的输出评分，再根据权重计算最终结果，选择得分最高的答案。

class WeightedEnsemble:
    def __init__(self):
        self.models = {
            "qwen3-32b": 0.5,
            "claude-sonnet-4": 0.3,
            "qwen3-8b": 0.2,
        }

    def decide(self, prompt: str) -> str:
        responses = {
            model: self.call_model(model, prompt)
            for model in self.models
        }
        scores = {}
        for model, response in responses.items():
            score = self.evaluate(response) * self.models[model]
            scores[response] = scores.get(response, 0) + score
        return max(scores, key=scores.get)

权重代表我们对不同模型输出的信任程度。实际使用时，应当依据真实业务中的表现不断调整，而不是直接参考基准测试结果。

模式 2：Consensus Ensemble（一致性集成）

Consensus Ensemble 要求多个模型先达成一致；如果无法满足一致性要求，再交由更强的模型继续处理。

class ConsensusEnsemble:
    def __init__(self, threshold: float = 0.7):
        self.threshold = threshold
        self.models = [
            "qwen3-32b",
            "claude-sonnet-4",
            "qwen3-8b",
        ]

    def decide(self, prompt: str) -> str:
        responses = [
            self.call_model(model, prompt)
            for model in self.models
        ]
        from collections import Counter
        votes = Counter(responses)
        max_votes = max(votes.values())

        if max_votes / len(self.models) >= self.threshold:
            return votes.most_common(1)[0][0]

        return self.call_model("qwen2.5-32b", prompt)

阈值决定了一致性要求有多严格。0.7 表示至少需要三分之二的模型给出一致结果。降低阈值可以更快得到结论；提高阈值则意味着只有在模型意见更加一致时才会直接返回结果，从而换来更高的置信度。

多模型系统何时值得用

如果业务中存在不同类型的工作负载、关键决策需要更高的输出质量，或者需要在成本和延迟之间做优化，那么多模型系统就有发挥价值的空间。

相反，如果所有任务的复杂度都差不多、还处于原型验证阶段，又或者系统的简单性比优化更重要，那就没必要引入多模型架构。

一个实用的经验法则：先从一个模型开始。只有真正遇到成本、延迟或质量方面的瓶颈时，再考虑引入更多模型。不要在问题尚未出现之前，就把系统设计得过于复杂。

总结

每一种架构都有对应的取舍，没有一种模式适用于所有场景。真正需要关注的是当前系统受什么约束，再选择最符合这些约束的模式。

作者：Rost Glukhov