
5分钟零改造!让任何Deepseek API支持工具调用
当主流Agent框架(如LangChain、5ire-MCP)遇上Deepseek API,工具调用不兼容的难题让无数开发者抓狂。笔者通过一个巧妙的"中间层魔法",让所有不兼容的API秒变OpenAI工具调用专家!在litellm的server config配置中增加自定义的model_info, use_proxy 来决定是否启用hook。这样当用户使用litellm暴露的API时就可以顺利的使
🎉 手把手教你解锁Deepseek API隐藏技能!5分钟实现工具调用兼容。
🔥 还在为Deepseek API不支持工具调用发愁?这个开源解决方案让所有Agent框架都能丝滑调用工具!
🚀 突破性解决方案原理揭秘
当主流Agent框架(如LangChain、5ire-MCP)遇上Deepseek API,工具调用不兼容的难题让无数开发者抓狂。笔者通过一个巧妙的"中间层魔法",让所有不兼容的API秒变OpenAI工具调用专家!
核心黑科技拆解:
1️⃣ 智能请求拦截用LiteLLM搭建代理网关,实时嗅探API流量
2️⃣ 语义翻译官将OpenAI标准的tools参数转化为System Prompt
3️⃣ 角色扮演术把tool角色响应伪装成user内容,完成协议兼容
具体实现代码如下:
import litellm
import os
from litellm.integrations.custom_logger import CustomLogger
from jinja2 import Environment, FileSystemLoader, select_autoescape
from litellm.proxy.proxy_server import UserAPIKeyAuth, DualCache, proxy_config
from litellm.types.utils import ModelResponse,ChatCompletionMessageToolCall
from litellm._logging import verbose_proxy_logger
from litellm.caching.caching import Cache
from typing import Optional, Literal
class ToolHandler(CustomLogger):
def __init__(self, message_logging = True):
super().__init__(message_logging)
# Set up Jinja2 template environment
template_path = os.path.join(os.path.dirname(__file__), 'templates')
self.env = Environment(
loader=FileSystemLoader(template_path),
autoescape=select_autoescape()
)
self.function_cache = Cache(type="local", default_in_memory_ttl=180.0)
def support_proxy(self, data: dict) ->bool:
model_list = proxy_config.config['model_list']
for model in model_list:
if model['model_name'] == data['model']:
if 'use_proxy' in model['model_info']:
return model['model_info']['use_proxy']
return False
async def async_pre_call_hook(self,
user_api_key_dict: UserAPIKeyAuth,
cache: DualCache,
data: dict,
call_type: Literal[
"completion",
"text_completion",
"embeddings",
"image_generation",
"moderation",
"audio_transcription",
]):
if not self.support_proxy(data) or call_type != "completion":
return data
verbose_proxy_logger.debug(f"use function proxy to hook data: {data}")
# Check if tools are present
if 'tools' in data and isinstance(data['tools'], list) and len(data['tools']) > 0:
# Render template
template = self.env.get_template('prompt_template.j2')
rendered_prompt = template.render(
tools=data['tools']
)
# Add tools prompt to system message
if 'messages' in data:
for message in data['messages']:
if message.get('role') == 'system':
message['content'] = f"{message['content']}\n\n{rendered_prompt}"
data["tools"] = None
# test role for ai tools
if 'messages' in data:
for message in data['messages']:
if 'tool_calls' in message:
del message['tool_calls']
if message.get('role') == 'tool':
id = message.get('tool_call_id')
if id is not None:
tool_call = await self.function_cache.async_get_cache(cache_key=id)
if tool_call is not None and isinstance(tool_call, dict):
function_name = tool_call['function']['name']
message['role'] = 'user'
content = message['content']
message['content'] = f"Function: {function_name}, Result: {content}"
del message['tool_call_id']
return data
async def async_post_call_success_hook(
self,
data: dict,
user_api_key_dict: UserAPIKeyAuth,
response: ModelResponse,
):
if not self.support_proxy(data):
return
verbose_proxy_logger.debug(f"use function proxy to hook response: {response}")
choice = response.choices[0]
content = choice.message.content
tool_calls = []
try:
json_blocks = content.split('```json')
for block in json_blocks[1:]:
json_str = block.split('```')[0].strip()
import json
json_data = json.loads(json_str)
if isinstance(json_data, list):
for item in json_data:
if 'name' in item and 'arguments' in item:
tool_call = ChatCompletionMessageToolCall(function=item)
tool_calls.append(tool_call)
key = tool_call.id
await self.function_cache.async_add_cache(tool_call, cache_key=key)
else:
if 'name' in json_data and 'arguments' in json_data:
tool_call = ChatCompletionMessageToolCall(function=json_data)
tool_calls.append(tool_call)
key = tool_call.id
await self.function_cache.async_add_cache(tool_call, cache_key=key)
if len(tool_calls)>0:
choice.finish_reason = 'tool_calls'
choice.message.tool_calls = tool_calls
response.choices = [choice]
except Exception as e:
verbose_proxy_logger.error(f"Error parsing JSON Markdown: {e}")
tool_call_handler_instance = ToolHandler()
其中使用了jinjia2 template实现sytem_prompt的工具调用组合后。具体的模板内容如下
{% if tools|length > 0 %}
你可以用如下工具解决问题:
{% for tool in tools %}{% if tool['type'] == 'function' %}
### {{ tool['function']['name'] }}
```json
{{ tool['function'] | tojson(indent=4) }}
```
{% endif %}{% endfor %}
## 注意:
1. 在调用上述函数时,请使用 Json 格式表示调用的参数。
2. 用尽量少的函数调用解决问题。
3. 调用JSON格式参考如下:
```json
{
"name": "send_email",
"arguments": {
"to_email": "Recipient's email address",
"title": "The title of the email",
"body": "The Body of the email"
}
}
}
```
{% endif %}
在litellm的server config配置中增加自定义的model_info, use_proxy 来决定是否启用hook。这样当用户使用litellm暴露的API时就可以顺利的使用Agent工具。
配置文件参考如下:
# config.yaml
model_list:
- model_name: deepseek-r1-250120
litellm_params:
model: openai/deepseek-r1-250120
api_base: https://ark.cn-beijing.volces.com/api/v3/
model_info:
use_proxy: true
- model_name: deepseek-v3-241226
litellm_params:
model: openai/deepseek-v3-241226
api_base: https://ark.cn-beijing.volces.com/api/v3/
model_info:
use_proxy: true
- model_name: doubao-1-5-pro-32k-250115
litellm_params:
model: openai/doubao-1-5-pro-32k-250115
api_base: https://ark.cn-beijing.volces.com/api/v3/
- model_name: sf-deepseek-v3
litellm_params:
model: openai/deepseek-ai/DeepSeek-V3
api_base: https://api.siliconflow.cn/v1
model_info:
use_proxy: true
litellm_settings:
callbacks: agentbr.proxy.tool_call_handler_instance
克隆代码后启动litellm
litellm --config config.yaml
笔者使用langchain的工具调用方法,对豆包的Deepseek API,测试通过。
已开源完整解决方案在 https://github.com/shanyou/deepseek-tools-call-example
💡 这个方案不仅适用于Deepseek,还可快速移植到其他大模型API!如果觉得有用记得点赞收藏。
更多推荐
所有评论(0)