在合同智能审查场景中,通常需要同时调用 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 或超时配置——【每个模型都必须使用专属 client,若混用 client,容易出现 headers 冲突或鉴权失效问题】:
编写并发调用核心函数
使用 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",这样更方便前端解析与接口消费——【键名必须全部小写且不能包含空格,否则在 FastAPI 自动序列化时可能出现兼容问题】。
将 gather 返回的三元组按既定顺序解包后,填充到 dict 中,最后再通过 json.dumps() 输出标准 JSON 结果。
接入 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。
