litellm 自定义提供商从0到1一个类接入任意 LLM【免费下载链接】litellmThe fastest, litest AI Gateway. Rust core with Python SDK. Call 100 LLM APIs in OpenAI (or native) format with cost tracking, guardrails, load balancing, and logging [Bedrock, Azure, OpenAI, Anthropic, OpenAI, VertexAI, vLLM, Nvidia NIM]项目地址: https://gitcode.com/GitHub_Trending/li/litellm你的项目里多了一家模型供应商接口格式和 OpenAI 对不上手动写适配层要改好几个文件写 litellm 自定义提供商其实没那么重加一个继承 BaseLLM 的类实现四个方法就能让litellm.completion()直接调它。这篇文章带你从最小骨架一路跑到首次调用成功。一张图看懂请求链路把 litellm 想象成机场值机柜台每家航司provider的票面格式都不同但柜台会把它们统一成同一张登机牌。一次请求进来链路固定四步入口从模型名前缀比如my_provider/quickstart-1解析出该走哪个 provider 类你的类把 OpenAI 风格的参数翻译成远端 API 要的样子发出请求拿回原始响应最后把响应转回统一的ModelResponse。想通 litellm 自定义 LLM 集成这条闭环后面的代码只是往格子里填肉。最小可跑骨架仓库里自带一份模板 litellm/llms/custom_llm.py四个方法的签名都替你想好了。照着它起个类from litellm.llms.base import BaseLLM from litellm.types.utils import GenericStreamingChunk from litellm.utils import ModelResponse class MyProvider(BaseLLM): def completion(self, model, messages, api_base, api_key, optional_params, **kw) - ModelResponse: ... # 发请求把响应包成 ModelResponse def streaming(self, model, messages, api_base, api_key, optional_params, **kw): ... # 逐块 yield GenericStreamingChunk async def acompletion(self, model, messages, api_base, api_key, optional_params, **kw) - ModelResponse: ... async def astreaming(self, model, messages, api_base, api_key, optional_params, **kw): ... # 注册模型名前缀 my_provider/ 即可被 litellm 路由到这个类为什么是这三样必须继承 litellm/llms/base.py 里的BaseLLM因为调度器按类去查找 handlerlitellm BaseLLM 继承是整个扩展开发的挂载点四个方法缺一不可同步/异步 × 非流式/流式是四条真实路由custom_chat_llm_router靠它们分派返回值必须是ModelResponse或 chunk因为下游的成本计算和 usage 统计全靠这些字段。四个核心方法真实仓库里每个 provider 都独占一个目录比如 litellm/llms/ollama/参数转换逻辑都拆在同目录的transformation.py里抄它的目录结构改就行。让 completion 把响应包成 litellm 认识的样子它负责同步、非流式的一次生成。最常踩的坑直接返回原始 dict——后面resp.choices[0].message.content一访问就炸。正确写法是参数从 OpenAI 格式转过去这件事在函数内部就近完成def completion(self, model, messages, api_base, api_key, optional_params, **kwargs) - ModelResponse: # OpenAI 风格参数已被筛进 optional_params别自己再解析一遍 payload { model: model, prompt: messages[-1][content], max_tokens: optional_params.get(max_tokens, 100), } resp client.post(f{api_base}/generate, jsonpayload, headers{Authorization: fBearer {api_key}}) # 转回统一格式下游成本、日志都读这些字段 return ModelResponse( idresp[id], modelmodel, choices[{index: 0, finish_reason: stop, message: {role: assistant, content: resp[text]}}], )acompletion 就是它的异步版逻辑一行都不用重写。流式 chunk 怎么解析streaming 负责同步流式输出是四个方法里踩坑最多的。两个硬约束必须是生成器返回 list 会一次性吐完再报错每个 yield 都必须是 OpenAI 流式格式的 chunk。litellm 流式响应适配的核心就在 SSE 逐行解析上def streaming(self, model, messages, api_base, api_key, optional_params, **kwargs): with client.stream(POST, f{api_base}/generate, jsonpayload) as resp: for line in resp.iter_lines(): if not line.startswith(data:): continue chunk json.loads(line.removeprefix(data:)) # 每块都要长成 OpenAI 流式 chunk 的样子 yield GenericStreamingChunk( choices[{delta: {content: chunk[token]}}], )astreaming 只是把这套循环换成异步版本。acompletion复用同步逻辑它负责异步非流式调用唯一区别是换用异步 HTTP client。最大的坑是在 async 函数里误用同步 client会把整个事件循环卡住async def acompletion(self, model, messages, api_base, api_key, optional_params, **kwargs) - ModelResponse: resp await client.post(f{api_base}/generate, jsonpayload, headersself._headers(api_key)) return self._to_model_response(resp, model) # 与 completion 共用转换函数把响应转换抽成私有方法同步异步各调一次省掉一半重复代码。astreamingasync for 逐行读它负责异步流式逻辑与 streaming 相同只是遍历方式变了——用aiter_lines()写成iter_lines()会拿到一个迭代器对象而非行内容async def astreaming(self, model, messages, api_base, api_key, optional_params, **kwargs): async with client.stream(POST, url, jsonpayload) as resp: async for line in resp.aiter_lines(): # 与 streaming 相同的 chunk 拼装逻辑 yield GenericStreamingChunk(choices[{delta: {...}}])四个方法写完custom_chat_llm_router会按调用方式自动分派你不需要再注册路由。注册与首次调用类落盘后litellm 按模型名前缀路由my_provider/开头的模型名都会进你的类。密钥按约定放进环境变量然后跑最小验证脚本import os, litellm os.environ[MY_PROVIDER_API_KEY] sk-test-xxx resp litellm.completion( modelmy_provider/quickstart-1, messages[{role: user, content: 你好}], ) print(resp.choices[0].message.content)能打印出模型的回答说明你的自定义 provider 就已经上线了。想加自动化用例照着 tests/ 里的现成测试抄结构即可。排错速查表症状大概率原因去哪看文件/方法resp.choices报NoneTypecompletion 返回的不是 ModelResponse字段没填全你的 provider 类completionstreamTrue只出一次全量内容streaming 返回了 list 而非生成器streaming流式第一帧就 TypeErroryield 的 chunk 不是 OpenAI 流式 dict 结构streaming/astreaming异步接口阻塞事件循环async 函数里混用了同步 clientacompletion401 或 KeyError 取不到密钥环境变量名不符合PROVIDER_API_KEY约定litellm/secret_managers/远端报错被吞、日志只剩 500没有抛带status_code的自定义异常custom_llm.py的CustomLLMError写法再往上走工具调用在 completion 里转换tools参数看 litellm/llms/ollama/ 各 provider 怎么处理。成本跟踪把新模型价格补进 model_prices_and_context_window.jsonproxy 会自动算 spend。图片生成模板里已留好image_generation方法照着填即可。路由与负载均衡provider 注册后可直接挂进 litellm/llms/init.py 的成本与路由体系配合 Router 做故障转移。上线后的可观测每个请求都会进审计日志效果长这样—— 把你写好的 provider 提个 PR 给社区同时盯住目标 API 的 changelog参数一变就改你的 transformation。【免费下载链接】litellmThe fastest, litest AI Gateway. Rust core with Python SDK. Call 100 LLM APIs in OpenAI (or native) format with cost tracking, guardrails, load balancing, and logging [Bedrock, Azure, OpenAI, Anthropic, OpenAI, VertexAI, vLLM, Nvidia NIM]项目地址: https://gitcode.com/GitHub_Trending/li/litellm创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 litellm 自定义提供商从0到1：一个类接入任意 LLM
litellm 自定义提供商从0到1一个类接入任意 LLM【免费下载链接】litellmThe fastest, litest AI Gateway. Rust core with Python SDK. Call 100 LLM APIs in OpenAI (or native) format with cost tracking, guardrails, load balancing, and logging [Bedrock, Azure, OpenAI, Anthropic, OpenAI, VertexAI, vLLM, Nvidia NIM]项目地址: https://gitcode.com/GitHub_Trending/li/litellm你的项目里多了一家模型供应商接口格式和 OpenAI 对不上手动写适配层要改好几个文件写 litellm 自定义提供商其实没那么重加一个继承 BaseLLM 的类实现四个方法就能让litellm.completion()直接调它。这篇文章带你从最小骨架一路跑到首次调用成功。一张图看懂请求链路把 litellm 想象成机场值机柜台每家航司provider的票面格式都不同但柜台会把它们统一成同一张登机牌。一次请求进来链路固定四步入口从模型名前缀比如my_provider/quickstart-1解析出该走哪个 provider 类你的类把 OpenAI 风格的参数翻译成远端 API 要的样子发出请求拿回原始响应最后把响应转回统一的ModelResponse。想通 litellm 自定义 LLM 集成这条闭环后面的代码只是往格子里填肉。最小可跑骨架仓库里自带一份模板 litellm/llms/custom_llm.py四个方法的签名都替你想好了。照着它起个类from litellm.llms.base import BaseLLM from litellm.types.utils import GenericStreamingChunk from litellm.utils import ModelResponse class MyProvider(BaseLLM): def completion(self, model, messages, api_base, api_key, optional_params, **kw) - ModelResponse: ... # 发请求把响应包成 ModelResponse def streaming(self, model, messages, api_base, api_key, optional_params, **kw): ... # 逐块 yield GenericStreamingChunk async def acompletion(self, model, messages, api_base, api_key, optional_params, **kw) - ModelResponse: ... async def astreaming(self, model, messages, api_base, api_key, optional_params, **kw): ... # 注册模型名前缀 my_provider/ 即可被 litellm 路由到这个类为什么是这三样必须继承 litellm/llms/base.py 里的BaseLLM因为调度器按类去查找 handlerlitellm BaseLLM 继承是整个扩展开发的挂载点四个方法缺一不可同步/异步 × 非流式/流式是四条真实路由custom_chat_llm_router靠它们分派返回值必须是ModelResponse或 chunk因为下游的成本计算和 usage 统计全靠这些字段。四个核心方法真实仓库里每个 provider 都独占一个目录比如 litellm/llms/ollama/参数转换逻辑都拆在同目录的transformation.py里抄它的目录结构改就行。让 completion 把响应包成 litellm 认识的样子它负责同步、非流式的一次生成。最常踩的坑直接返回原始 dict——后面resp.choices[0].message.content一访问就炸。正确写法是参数从 OpenAI 格式转过去这件事在函数内部就近完成def completion(self, model, messages, api_base, api_key, optional_params, **kwargs) - ModelResponse: # OpenAI 风格参数已被筛进 optional_params别自己再解析一遍 payload { model: model, prompt: messages[-1][content], max_tokens: optional_params.get(max_tokens, 100), } resp client.post(f{api_base}/generate, jsonpayload, headers{Authorization: fBearer {api_key}}) # 转回统一格式下游成本、日志都读这些字段 return ModelResponse( idresp[id], modelmodel, choices[{index: 0, finish_reason: stop, message: {role: assistant, content: resp[text]}}], )acompletion 就是它的异步版逻辑一行都不用重写。流式 chunk 怎么解析streaming 负责同步流式输出是四个方法里踩坑最多的。两个硬约束必须是生成器返回 list 会一次性吐完再报错每个 yield 都必须是 OpenAI 流式格式的 chunk。litellm 流式响应适配的核心就在 SSE 逐行解析上def streaming(self, model, messages, api_base, api_key, optional_params, **kwargs): with client.stream(POST, f{api_base}/generate, jsonpayload) as resp: for line in resp.iter_lines(): if not line.startswith(data:): continue chunk json.loads(line.removeprefix(data:)) # 每块都要长成 OpenAI 流式 chunk 的样子 yield GenericStreamingChunk( choices[{delta: {content: chunk[token]}}], )astreaming 只是把这套循环换成异步版本。acompletion复用同步逻辑它负责异步非流式调用唯一区别是换用异步 HTTP client。最大的坑是在 async 函数里误用同步 client会把整个事件循环卡住async def acompletion(self, model, messages, api_base, api_key, optional_params, **kwargs) - ModelResponse: resp await client.post(f{api_base}/generate, jsonpayload, headersself._headers(api_key)) return self._to_model_response(resp, model) # 与 completion 共用转换函数把响应转换抽成私有方法同步异步各调一次省掉一半重复代码。astreamingasync for 逐行读它负责异步流式逻辑与 streaming 相同只是遍历方式变了——用aiter_lines()写成iter_lines()会拿到一个迭代器对象而非行内容async def astreaming(self, model, messages, api_base, api_key, optional_params, **kwargs): async with client.stream(POST, url, jsonpayload) as resp: async for line in resp.aiter_lines(): # 与 streaming 相同的 chunk 拼装逻辑 yield GenericStreamingChunk(choices[{delta: {...}}])四个方法写完custom_chat_llm_router会按调用方式自动分派你不需要再注册路由。注册与首次调用类落盘后litellm 按模型名前缀路由my_provider/开头的模型名都会进你的类。密钥按约定放进环境变量然后跑最小验证脚本import os, litellm os.environ[MY_PROVIDER_API_KEY] sk-test-xxx resp litellm.completion( modelmy_provider/quickstart-1, messages[{role: user, content: 你好}], ) print(resp.choices[0].message.content)能打印出模型的回答说明你的自定义 provider 就已经上线了。想加自动化用例照着 tests/ 里的现成测试抄结构即可。排错速查表症状大概率原因去哪看文件/方法resp.choices报NoneTypecompletion 返回的不是 ModelResponse字段没填全你的 provider 类completionstreamTrue只出一次全量内容streaming 返回了 list 而非生成器streaming流式第一帧就 TypeErroryield 的 chunk 不是 OpenAI 流式 dict 结构streaming/astreaming异步接口阻塞事件循环async 函数里混用了同步 clientacompletion401 或 KeyError 取不到密钥环境变量名不符合PROVIDER_API_KEY约定litellm/secret_managers/远端报错被吞、日志只剩 500没有抛带status_code的自定义异常custom_llm.py的CustomLLMError写法再往上走工具调用在 completion 里转换tools参数看 litellm/llms/ollama/ 各 provider 怎么处理。成本跟踪把新模型价格补进 model_prices_and_context_window.jsonproxy 会自动算 spend。图片生成模板里已留好image_generation方法照着填即可。路由与负载均衡provider 注册后可直接挂进 litellm/llms/init.py 的成本与路由体系配合 Router 做故障转移。上线后的可观测每个请求都会进审计日志效果长这样—— 把你写好的 provider 提个 PR 给社区同时盯住目标 API 的 changelog参数一变就改你的 transformation。【免费下载链接】litellmThe fastest, litest AI Gateway. Rust core with Python SDK. Call 100 LLM APIs in OpenAI (or native) format with cost tracking, guardrails, load balancing, and logging [Bedrock, Azure, OpenAI, Anthropic, OpenAI, VertexAI, vLLM, Nvidia NIM]项目地址: https://gitcode.com/GitHub_Trending/li/litellm创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考