大模型接入实战从 API 调用到多模型管理手把手教你搭个万能接口系列导读这是「AI 应用开发」第 2 篇。上一篇咱们聊了全景图今天直接上手写代码——怎么把各家大模型GPT-4、Claude、文心一言、通义千问…统一封装起来让上层业务代码无感切换。一、问题引入一个真实的烦恼假设你们公司刚开始用 AI技术选型还没定。今天产品经理说“先接 OpenAI 试试”下周老板说“国内合规要求高切文心一言”再过一阵技术总监说“成本太高咱们自己部署个 Llama…”你作为负责接入的工程师内心是崩溃的“每个模型 API 格式都不一样难道我要写三套代码”OpenAI 的调用长这样importopenai responseopenai.ChatCompletion.create(modelgpt-4,messages[{role:user,content:你好}])文心一言的调用长这样importrequests responserequests.post(https://aip.baidubce.com/rpc/2.0/ai_custom/v1/wenxinworkshop/chat/completions,headers{Authorization:fBearer{baidu_token}},json{messages:[{role:user,content:你好}]})通义千问的调用又长这样…格式不统一、参数名不一样、错误处理各异。如果业务代码直接耦合某个模型的 SDK后面切换就是一场灾难。问题来了怎么设计一个万能接口让上层只关心我要调用大模型不关心底层是谁二、方案分析统一抽象层我一开始想的是直接写个 if-else 不就行了# 别这么干这是反面教材defcall_model(model_name,prompt):ifmodel_nameopenai:returncall_openai(prompt)elifmodel_namewenxin:returncall_wenxin(prompt)elifmodel_nameqwen:returncall_qwen(prompt)# ... 每次加模型都要改这里但这有几个问题开闭原则 violation加新模型要改核心代码参数映射散落各处温度、最大长度等参数各家命名不同异常处理不统一有的抛异常有的返回错误码扩展性差加流式输出、函数调用等新特性时到处都要改换个思路抽象出统一接口每个模型做适配器。就像 JDBC 统一了各种数据库的访问咱们也来个 “LLM-JDBC”。三、实现过程Step by Step第一步定义统一接口先想清楚上层业务到底需要什么# 统一接口所有模型适配器必须实现这个接口classLLMProvider(ABC):abstractmethoddefchat(self,messages:List[Message],config:ChatConfigNone)-ChatResponse:非流式对话passabstractmethoddefchat_stream(self,messages:List[Message],config:ChatConfigNone)-Iterator[ChatChunk]:流式对话打字机效果passabstractmethoddeffunction_call(self,messages:List[Message],functions:List[Function],config:ChatConfigNone)-FunctionCallResult:函数调用让模型决定调用哪个工具pass关键设计决策Message统一格式内部再转成各家需要的格式ChatConfig封装通用参数温度、最大 token、超时等返回结构统一包含内容、用量统计、原始响应调试用第二步统一数据模型各家 API 的字段名不一样但概念是一样的。咱们统一dataclassclassMessage:role:str# system, user, assistant, functioncontent:strname:Optional[str]None# function call 时用dataclassclassChatConfig:temperature:float0.7max_tokens:Optional[int]Nonetop_p:Optional[float]Nonetimeout:int30# 扩展各家特有的参数放这里extra_params:Dict[str,Any]NonedataclassclassChatResponse:content:strmodel:strusage:TokenUsage# prompt_tokens, completion_tokens, total_tokensraw_response:Any# 原始响应调试用finish_reason:str# stop, length, function_calldataclassclassTokenUsage:prompt_tokens:intcompletion_tokens:inttotal_tokens:int# 成本估算用estimated_cost:float0.0为什么要保留raw_response血泪教训上线后出问题发现是某家模型返回了奇怪的格式。保留原始响应排查问题快 10 倍。第三步实现 OpenAI 适配器以 OpenAI 为例看看适配器怎么写classOpenAIProvider(LLMProvider):def__init__(self,api_key:str,base_url:strNone,model:strgpt-4):self.clientOpenAI(api_keyapi_key,base_urlbase_url)self.modelmodeldefchat(self,messages:List[Message],config:ChatConfigNone)-ChatResponse:configconfigorChatConfig()# 1. 统一格式 → OpenAI 格式openai_messages[{role:m.role,content:m.content,**({name:m.name}ifm.nameelse{})}forminmessages]# 2. 调用 APIresponseself.client.chat.completions.create(modelself.model,messagesopenai_messages,temperatureconfig.temperature,max_tokensconfig.max_tokens,top_pconfig.top_p,**(config.extra_paramsor{}))# 3. OpenAI 格式 → 统一格式choiceresponse.choices[0]returnChatResponse(contentchoice.message.content,modelresponse.model,usageTokenUsage(prompt_tokensresponse.usage.prompt_tokens,completion_tokensresponse.usage.completion_tokens,total_tokensresponse.usage.total_tokens,estimated_costself._calc_cost(response.usage,response.model)),raw_responseresponse,finish_reasonchoice.finish_reason)defchat_stream(self,messages:List[Message],config:ChatConfigNone):# 流式输出把 SSE 流转成统一的 Chunk 迭代器configconfigorChatConfig()openai_messages[...]# 同上streamself.client.chat.completions.create(modelself.model,messagesopenai_messages,streamTrue,# 关键开流式**params)forchunkinstream:deltachunk.choices[0].deltayieldChatChunk(contentdelta.contentor,modelchunk.model,is_finishedchunk.choices[0].finish_reasonisnotNone)def_calc_cost(self,usage,model:str)-float:# 成本计算不同模型单价不同pricing{gpt-4:{prompt:0.03,completion:0.06},# 每 1K tokensgpt-3.5-turbo:{prompt:0.0015,completion:0.002},}ppricing.get(model,pricing[gpt-4])return(usage.prompt_tokens*p[prompt]usage.completion_tokens*p[completion])/1000关键点三层转换统一格式 → 厂商格式 → 统一格式成本计算内置调用完就知道花了多少钱流式输出用生成器内存友好响应快第四步实现其他厂商适配器文心一言、通义千问的适配器结构一样只是转换逻辑不同classWenxinProvider(LLMProvider):def__init__(self,api_key:str,secret_key:str,model:strernie-bot-4):self.api_keyapi_key self.secret_keysecret_key self.modelmodel self.access_tokenself._get_access_token()defchat(self,messages:List[Message],config:ChatConfigNone)-ChatResponse:# 文心一言的特殊点# 1. 需要先用 AK/SK 换 access_token# 2. 参数名不一样temperature → temperature这次碰巧一样# 3. 返回结构不一样wenxin_messagesself._convert_messages(messages)responserequests.post(fhttps://aip.baidubce.com/rpc/2.0/ai_custom/v1/wenxinworkshop/chat/{self.model}?access_token{self.access_token},json{messages:wenxin_messages,temperature:config.temperatureifconfigelse0.7,# 文心特有的参数...}).json()# 错误处理文心返回的错误格式也不一样iferror_codeinresponse:raiseLLMError(fWenxin error:{response[error_msg]})returnChatResponse(contentresponse[result],modelself.model,usageTokenUsage(prompt_tokensresponse.get(usage,{}).get(prompt_tokens,0),completion_tokensresponse.get(usage,{}).get(completion_tokens,0),total_tokensresponse.get(usage,{}).get(total_tokens,0)),raw_responseresponse,finish_reasonstop# 文心没有 finish_reason默认 stop)坑点提醒文心一言的 access_token 有有效期需要自动刷新通义千问的流式输出是 SSE 格式但事件名和 OpenAI 不一样有些模型不支持 system message要自动转成 user message第五步工厂模式 配置化适配器写好了怎么创建用工厂模式 配置文件classLLMFactory:根据配置创建对应的模型提供者_providers{openai:OpenAIProvider,wenxin:WenxinProvider,qwen:QwenProvider,ollama:OllamaProvider,# 本地部署}classmethoddefcreate(cls,name:str,**kwargs)-LLMProvider:ifnamenotincls._providers:raiseValueError(fUnknown provider:{name}. Available:{list(cls._providers.keys())})returncls._providers[name](**kwargs)classmethoddefregister(cls,name:str,provider_class:Type[LLMProvider]):注册新的模型提供者扩展用cls._providers[name]provider_class# 配置文件 config.yamlllm:default:openaiproviders:openai:api_key:${OPENAI_API_KEY}model:gpt-4wenxin:api_key:${WENXIN_API_KEY}secret_key:${WENXIN_SECRET_KEY}model:ernie-bot-4ollama:# 本地 Llamabase_url:http://localhost:11434model:llama3:8b使用方式# 加载配置configload_config(config.yaml)providerLLMFactory.create(openai,**config.llm.providers.openai)# 业务代码完全无感messages[Message(roleuser,content帮我写个 Python 快速排序)]responseprovider.chat(messages)print(response.content)print(f花了{response.usage.estimated_cost:.4f}美元)第六步多模型路由与降级企业级场景往往需要更复杂的策略classMultiLLMRouter:多模型路由根据策略选择模型支持降级def__init__(self,config:RouterConfig):self.providers{}# name - LLMProviderself.strategyconfig.strategy# priority, round_robin, cost_basedself.fallback_chainconfig.fallback_chain# [gpt-4, claude-3, qwen]defchat(self,messages:List[Message],config:ChatConfigNone)-ChatResponse:# 按优先级尝试失败自动降级forprovider_nameinself.fallback_chain:try:providerself.providers[provider_name]returnprovider.chat(messages,config)exceptLLMErrorase:logger.warning(f{provider_name}failed:{e}, trying next...)continueraiseLLMError(All providers failed!)defchat_with_routing(self,messages:List[Message],task_type:strNone)-ChatResponse:# 智能路由根据任务类型选模型# 简单任务 → 便宜的小模型# 复杂任务 → 强力大模型iftask_typesimple_qa:returnself.providers[gpt-3.5-turbo].chat(messages)eliftask_typecode_generation:returnself.providers[claude-3].chat(messages)eliftask_typecreative_writing:returnself.providers[gpt-4].chat(messages)# ...实际应用场景成本优化简单问题用 GPT-3.5复杂问题才用 GPT-4高可用主模型挂了自动切备用模型合规路由敏感数据走私有化部署普通问题走公有 API四、验证跑个完整的例子咱们来验证一下这套设计# 完整示例多模型无缝切换fromllm_coreimportLLMFactory,Message,ChatConfig# 1. 创建两个模型实例openaiLLMFactory.create(openai,api_keysk-xxx,modelgpt-3.5-turbo)wenxinLLMFactory.create(wenxin,api_keyak-xxx,secret_keysk-xxx)# 2. 同样的业务代码换模型只需改一行provideropenai# 或 wenxinmessages[Message(rolesystem,content你是一位专业的技术顾问),Message(roleuser,content解释一下微服务架构的优缺点)]# 3. 调用完全一样的代码responseprovider.chat(messages,ChatConfig(temperature0.7))print(f回答{response.content})print(f模型{response.model})print(fToken 用量{response.usage.total_tokens})print(f预估成本${response.usage.estimated_cost:.4f})# 4. 流式输出也是同样的接口print(\n--- 流式输出 ---)forchunkinprovider.chat_stream(messages):print(chunk.content,end,flushTrue)ifchunk.is_finished:print(\n[完成])运行结果回答微服务架构是一种将应用拆分为多个小型、独立服务的架构风格... 模型gpt-3.5-turbo-0125 Token 用量342 预估成本$0.0005 --- 流式输出 --- 微服务架构是一种将应用拆分为多个小型、独立服务的架构风格...[完成]验证通过切换模型只需要改provider openai这一行业务代码完全不用动。五、进阶加上函数调用Function Calling函数调用是让 AI 动起来的关键能力。统一封装后用法也很简洁# 定义工具查询天气weather_functionFunction(nameget_weather,description获取指定城市的天气,parameters{type:object,properties:{city:{type:string,description:城市名},date:{type:string,description:日期格式 YYYY-MM-DD}},required:[city]})# AI 决定调用哪个函数messages[Message(roleuser,content北京明天天气怎么样)]resultprovider.function_call(messages,functions[weather_function])ifresult.typefunction_call:print(fAI 要调用{result.function_name})print(f参数{result.arguments})# 实际执行函数...weatherget_weather(**result.arguments)# 把结果返回给 AI让它组织语言回答各家模型的函数调用格式差异很大但上层业务代码完全感知不到。六、小结今天咱们实现了一套企业级大模型统一接入层核心设计设计要点解决的问题统一接口LLMProvider业务代码与具体模型解耦统一数据模型格式转换集中在适配器不散落在业务代码工厂模式 配置化加新模型不改动核心代码符合开闭原则多模型路由成本优化、高可用、合规路由内置成本计算每次调用都知道花了多少钱这套代码不是玩具是可以在生产环境用的骨架。实际项目中你可能还需要重试机制带指数退避请求/响应日志限流熔断异步调用支持这些咱们在后面的性能优化篇再展开。你接入大模型时遇到过哪些坑比如某个模型的奇葩返回格式欢迎交流