🎉 手把手教你解锁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!如果觉得有用记得点赞收藏。

Logo

欢迎加入DeepSeek 技术社区。在这里,你可以找到志同道合的朋友,共同探索AI技术的奥秘。

更多推荐