Atoms多模型同时调用业务场景落地完整代码

需同时调用Qwen、GLM、Claude三个模型来处理合同文本：Qwen用于条款摘要，GLM负责识别风险点，Claude提供合规建议。通过asyncio.gather实现并发执行，采用dataclass或小写key dict统一响应格式，并通过FastAPI暴露/analyze接口。

需要在同一个业务请求中并行调用多个大模型（如Qwen、GLM、Claude）完成不同子任务，并统一编排响应结果，比如用户提交一份合同文本，同时让Qwen做条款摘要、GLM做风险点识别、Claude做合规性建议。

准备多模型客户端与配置

安装必要依赖：pip install dashscope zhipuai anthropic python-dotenv。

在项目根目录创建 .env 文件，填入各平台 API Key：注意必须使用各自平台最新版 SDK 支持的环境变量名，旧版 key 名会导致初始化失败：

DASHSCOPE_API_KEY=sk-xxx（通义千问）
ZHIPUAI_API_KEY=123456（智谱GLM）
ANTHROPIC_API_KEY=sk-ant-api03-xxx（Claude）

新建 models.py，定义三个独立客户端实例，不共用 session 或超时设置——

：

编写并发调用核心函数

使用 asyncio.gather 并发触发三路请求，不采用线程池——模型 API 原生支持异步 HTTP，线程反而增加调度开销。

首先，定义单模型推理函数async_qwen()，其输入为text，输出为摘要字符串；其次，定义async_glm()，输入text后，返回JSON格式的风险列表（包含位置、类型、描述字段）；最后，定义async_claude()，输入text，返回带有段落编号的合规建议Markdown字符串。

将三个函数传入 asyncio.gather()，等待全部完成。这一步不能用 asyncio.wait() 或 create_task 单独 await，否则失去并发性。

构造统一响应结构

方法一：用 dataclass 定义 ResponseModel，包含 qwen_summary: str、glm_risks: list、claude_advice: str 三个字段。

方法二：直接用 dict 组装，键名固定为 "summary" / "risks" / "advice"，便于前端消费——

。

把 gather 返回的三元组按顺序解包，塞进 dict，最后 json.dumps() 输出。

接入 FastAPI 路由

创建 main.py，初始化 FastAPI 实例，添加 POST /analyze 后端接口。

接收 JSON body：{"contract_text": "甲方应于…"}，校验字段存在且非空，长度不超过 128KB（防止 OOM）。

调用上一步封装好的并发函数，await 得到结果字典，直接 return 它——FastAPI 会自动处理 JSON 序列化和 Content-Type 头。

启动服务：uvicorn main:app --reload --host 0.0.0.0 --port 8000。