用 100 行代码实现一个混合智能体——让 DeepSeek、豆包等多个模型协作回答问题

用户问题
    ↓
路由器 Agent（分析问题类型）
    ↓
┌────────┬────────┬────────┐
│ 分析师 │ 编码师 │ 总结师 │  ← 不同模型负责不同任务
│DeepSeek│ 豆包   │ DeepSeek│
└────────┴────────┴────────┘
    ↓
合成 Agent（整合结果）
    ↓
最终回答

# moa_agent.py
from openai import OpenAI
import os, json
from dotenv import load_dotenv
load_dotenv()

clients = {
    "deepseek": OpenAI(api_key=os.getenv("DEEPSEEK_KEY"), base_url="https://api.deepseek.com/v1"),
    "doubao": OpenAI(api_key=os.getenv("DOUBAO_KEY"), base_url="https://ark.cn-beijing.volces.com/api/v3"),
}

MODEL_MAP = {"deepseek": "deepseek-chat", "doubao": "doubao-2.1-pro"}

def call_llm(provider, system_prompt, user_prompt, temperature=0.3):
    client = clients[provider]
    resp = client.chat.completions.create(
        model=MODEL_MAP[provider],
        messages=[{"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt}],
        temperature=temperature,
    )
    return resp.choices[0].message.content

class MoAgent:
    """混合智能体：分析问题 → 分派任务 → 多模型协作 → 整合结果。"""

    def answer(self, question):
        # 1. 路由器：分析问题类型并分派
        plan = self._route(question)

        # 2. 并行调用各专家模型
        results = {}
        for expert in plan["experts"]:
            role = expert["role"]
            prompt = expert["prompt"]
            provider = expert.get("provider", "deepseek")
            results[role] = call_llm(provider, prompt, question)

        # 3. 合成最终答案
        return self._synthesize(question, results)

    def _route(self, question):
        route_prompt = f"""分析以下问题，确定需要哪些专家角色。

返回 JSON：
{{"experts":[{{"role":"角色名","prompt":"系统提示","provider":"deepseek或doubao"}}]}}

分析以下问题：{question}"""

        result = call_llm("deepseek", "", route_prompt)
        try:
            return json.loads(result)
        except:
            return {"experts": [{"role": "通用助手", "prompt": "回答以下问题", "provider": "deepseek"}]}

    def _synthesize(self, question, results):
        findings = "\n\n".join(f"### {role}\n{content}" for role, content in results.items())
        return call_llm("deepseek",
            "你是一个综合分析师。整合多位专家的观点，给出一个完整、准确的最终答案。",
            f"问题：{question}\n\n专家意见：\n{findings}"
        )

agent = MoAgent()

# 使用
answer = agent.answer("Python 中如何处理大文件读取而不导致内存溢出？")
print(answer)

问题：Python 中如何处理大文件读取而不导致内存溢出？

🤖 路由器分析：
  - 分析师（DeepSeek）：分析问题，提供理论方案
  - 编码师（豆包）：提供代码实现

📊 分析师（DeepSeek）：
  大文件读取的核心策略是流式处理，主要有三种方式：
  1. 逐行读取（readline）
  2. 分块读取（read(chunk_size)）
  3. 内存映射（mmap）

💻 编码师（豆包）：
  ```python
  # 首选方案：分块读取
  with open("large.csv", "r") as f:
      while chunk := f.read(8192):
          process(chunk)
  ```

🎯 综合答案：
  对于大文件读取，推荐使用流式分块读取，每次读取 8KB...