摘要:2026年开年王炸!DeepSeek-V3.2 及其 Speciale 版本横空出世,以不到 GPT-5 四分之一的价格,在数学、编程及 Agent 性能上实现全面领跑。本文将带你从零开始,实战 DeepSeek-V3.2 的 API 调用、Function Calling 原理,并深入 LangChain 框架开发 Agent。最重要的是,本文将独家公开如何解决 LangChain 1.x 版本无法完美适配 DeepSeek-V3.2-Reasoner(推理模型)的痛点,手把手教你重写 ChatModel,实现带“思考链”的智能 Agent。

关键词:DeepSeek-V3.2, Agent, LangChain, Function Calling, Python, 大模型开发, 思考链


前言:DeepSeek-V3.2 的技术变革

距离 DeepSeek-V3.2-EXP 发布仅两个月,正式版 DeepSeek-V3.2 已震撼上线。这不仅仅是一次版本号的更迭,更是底层架构的革新。

  • DSA 稀疏注意力机制:在维持性能的同时,推理和训练成本暴降 50%。

  • GRPO 训练框架:结合大规模合成 Agent 数据集,突破性能极限。

  • Speciale 模型(推理版):解除思考链长度限制,引入自验证数学推理(Self-verification),在 HLE(人类最后测试)及 Agent 任务中追平 Gemini 3.0 Pro。

最令人心动的是它的性价比——价格仅为 Gemini 3.0 的 1/5,GPT-5.1 的 1/4。对于开发者而言,这意味着我们可以在生产环境中以更低的成本部署更智能的 Agent。

本文将分为三个部分带领大家实战:

  1. 基础篇:API 快速上手与多模态交互。

  2. 进阶篇:深入理解 Function Calling(工具调用)底层原理。

  3. 高阶篇:LangChain 集成与 Reasoner 模型适配。


第一部分:DeepSeek-V3.2 快速上手

在开始 Agent 开发前,我们需要确保环境配置正确。

1.1 环境准备与 API 配置

首先,前往 DeepSeek 开放平台申请 API Key。
为了安全起见,严禁在代码中硬编码 Key。我们使用 .env 文件来管理环境变量。

项目结构:

project/
├── .env
├── main.py
└── requirements.txt

.env 文件内容:

DEEPSEEK_API_KEY=sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx  # 请替换为你的实际 Key
OPENWEATHER_API_KEY=xxxxxxxxxxxxxxxxxxxxxxxxxx      # 后续天气查询需要

加载环境变量:

import os
from dotenv import load_dotenv

# 加载环境变量,override=True 确保覆盖系统变量
load_dotenv(override=True)

DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY")

if not DEEPSEEK_API_KEY:
    raise ValueError("未检测到 DEEPSEEK_API_KEY,请检查 .env 文件")

1.2 基础对话模型调用 (deepseek-chat)

我们直接使用兼容 OpenAI 协议的 SDK 进行连接测试。

from openai import OpenAI

client = OpenAI(
    api_key=DEEPSEEK_API_KEY, 
    base_url="https://api.deepseek.com"
)

response = client.chat.completions.create(
    model="deepseek-chat",
    messages=[
        {"role": "system", "content": "你是乐于助人的助手"},
        {"role": "user", "content": "你好,请介绍一下 DeepSeek-V3.2 的特点。"},
    ],
)

print(response.choices[0].message.content)

输出示例:

 能力方面:纯文本模型,支持 128K 上下文,擅长代码生成与逻辑分析...
 服务理念:完全免费使用,API 价格极具竞争力...

1.3 推理模型调用 (deepseek-reasoner)

DeepSeek-V3.2-Speciale(即 deepseek-reasoner)是本次更新的重头戏。它在输出最终答案前,会先进行一段“思维链”(CoT)推理。

response_reasoner = client.chat.completions.create(
    model="deepseek-reasoner",
    messages=[
        {"role": "user", "content": "如果有3个苹果,吃掉了1个,还剩几个?请通过思考回答。"},
    ],
)

# 获取思考过程
reasoning = response_reasoner.choices[0].message.reasoning_content
# 获取最终回答
content = response_reasoner.choices[0].message.content

print(f"🧠 [思考过程]:\n{reasoning}\n")
print(f"🗣️ [最终回答]:\n{content}")

注意:reasoning_content 是 DeepSeek 特有的字段,标准的 OpenAI SDK 对象中可能不包含此字段提示,但实际上响应 JSON 中是存在的。后续在 LangChain 中我们会详细解决这个字段的解析问题。


第二部分:原生 Function Calling 深度实战

Agent 的核心在于使用工具。DeepSeek-V3.2 首次引入了“思考模式下工具调用”,极大地提升了复杂任务的一致性。

2.1 定义外部工具函数

我们以一个经典的“查询天气”为例。这里需要调用 OpenWeatherMap 的 API。

import requests
import json

def get_weather(loc):
    """
    查询即时天气函数
    :param loc: 城市名称(英文),如 'Beijing'
    :return: JSON 格式的天气数据
    """
    api_key = os.getenv("OPENWEATHER_API_KEY")
    if not api_key:
        return json.dumps({"error": "Missing OpenWeather API Key"})

    url = "https://api.openweathermap.org/data/2.5/weather"
    params = {
        "q": loc,
        "appid": api_key,
        "units": "metric",
        "lang": "zh_cn"
    }
    
    try:
        response = requests.get(url, params=params)
        data = response.json()
        return json.dumps(data)
    except Exception as e:
        return json.dumps({"error": str(e)})

# 注册工具描述(JSON Schema)
tools = [
    {
        "type": "function",
        "function": {
            'name': 'get_weather',
            'description': '查询即时天气函数,需输入城市英文名',
            'parameters': {
                'type': 'object',
                'properties': {
                    'loc': {
                        'description': "城市名称,如 'Beijing', 'Shanghai'",
                        'type': 'string'
                    }
                },
                'required': ['loc']
            }
        }
    }
]

2.2 手撸 Function Calling 执行流

理解这个流程对于 debug 复杂的 Agent 至关重要。一个完整的工具调用循环包含:

  1. 用户提问。

  2. 模型第一轮响应:返回 tool_calls(告诉我们需要调用的函数名和参数)。

  3. 代码执行函数:本地运行函数,获取结果。

  4. 模型第二轮响应:将函数运行结果(role: tool)回传给模型,模型生成最终自然语言回复。

实战代码(支持多步调用):

# 初始化消息列表
messages = [{"role": "user", "content": "请问北京和杭州今天天气如何?"}]
available_functions = {"get_weather": get_weather}

# --- 第一轮交互 ---
response = client.chat.completions.create(
    model="deepseek-reasoner", # 尝试使用推理模型进行工具规划
    messages=messages,
    tools=tools,
)

response_msg = response.choices[0].message
tool_calls = response_msg.tool_calls

if tool_calls:
    print(f"🤖 模型决定调用工具: {len(tool_calls)} 次")
    # 将模型的回复(包含 tool_calls)加入历史记录
    messages.append(response_msg.model_dump())
    
    # --- 执行工具 ---
    for tool_call in tool_calls:
        func_name = tool_call.function.name
        func_args = json.loads(tool_call.function.arguments)
        
        print(f"🔧 执行函数: {func_name} 参数: {func_args}")
        
        # 动态调用函数
        function_to_call = available_functions[func_name]
        func_result = function_to_call(**func_args)
        
        # 将结果作为 tool 类型的消息加入历史
        messages.append({
            "role": "tool",
            "content": func_result,
            "tool_call_id": tool_call.id
        })

    # --- 第二轮交互(汇总结果) ---
    final_response = client.chat.completions.create(
        model="deepseek-reasoner",
        messages=messages,
        tools=tools,
    )
    
    print("💡 最终回答:")
    print(final_response.choices[0].message.content)

运行结果分析:
DeepSeek-V3.2 会精准地识别出需要分别查询 "Beijing" 和 "Hangzhou",并发起两个 tool_call,这展示了其强大的意图识别和多任务拆解能力。


第三部分:LangChain 集成开发

手动管理 messages 列表非常繁琐,接下来我们将 DeepSeek 接入 LangChain 框架。

3.1 LangChain 基础接入

pip install langchain-deepseek langchain-community
from langchain_deepseek import ChatDeepSeek

# 基础对话模型
chat_model = ChatDeepSeek(
    model="deepseek-chat",
    api_key=os.getenv("DEEPSEEK_API_KEY")
)

# 推理模型
reasoner_model = ChatDeepSeek(
    model="deepseek-reasoner",
    api_key=os.getenv("DEEPSEEK_API_KEY")
)

res = chat_model.invoke("你好,介绍下你自己")
print(res.content)

3.2 使用 LangChain 创建 Agent

我们使用 LangChain 的 @tool 装饰器重写工具,并创建一个能够查询天气写入文件的智能体。

from langchain.agents import create_agent, AgentExecutor
from langchain.tools import tool
from datetime import datetime

# 定义工具 1:查询天气
@tool
def get_weather_tool(loc: str) -> str:
    """查询指定城市的实时天气。参数 loc 为城市英文名,如 Beijing"""
    return get_weather(loc) # 复用之前的逻辑

# 定义工具 2:写入文件
@tool
def write_file_tool(content: str) -> str:
    """将内容写入本地文件,文件名自动生成"""
    try:
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        filename = f"output_{timestamp}.txt"
        with open(filename, "w", encoding="utf-8") as f:
            f.write(content)
        return f"✅ 文件已保存至: {os.path.abspath(filename)}"
    except Exception as e:
        return f"❌ 写入失败: {e}"

# 创建 Agent
# 注意:这里暂时使用 deepseek-chat,因为 reasoner 模型在 LangChain 1.x 有坑
model = ChatDeepSeek(model="deepseek-chat")
tools = [get_weather_tool, write_file_tool]

# 这里使用 create_tool_calling_agent 是更现代的写法
from langchain.agents import create_tool_calling_agent
from langchain.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一名高效的助手,可以调用工具解决问题。"),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

agent = create_tool_calling_agent(model, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# 执行任务
agent_executor.invoke({"input": "查询上海和深圳的天气,并写个对比报告保存到本地。"})

执行效果:
Agent 会先并行调用两次天气查询,获取数据后进行对比分析,最后调用 write_file_tool 将结果保存。整个过程行云流水。


第四部分:填坑!适配 DeepSeek-V3.2-Reasoner 模型

如果你尝试将上述代码中的模型换成 deepseek-reasoner,在 LangChain 1.x 环境下,你极大概率会遇到以下报错:

AttributeError, NameError 或工具调用死循环。

核心原因:

  1. Thinking 字段丢失:LangChain 标准对象不包含 reasoning_content,导致思考过程丢失。

  2. 消息格式不兼容:推理模型在输出 tool_calls 时,其内部处理逻辑与标准 deepseek-chat 或 OpenAI 略有不同,导致 LangChain 的解析器无法正确提取工具参数。

  3. 异步/同步执行器冲突:如文中报错所示,tool_node 在处理并发工具调用时,与 DeepSeek Reasoner 的流式输出配合存在 Bug。

终极解决方案:自定义 DeepSeekReasonerChatModel

为了解决这个问题,我们需要通过继承 BaseChatModel,手写一个完全适配 DeepSeek 特性的 ChatModel 类。这个类将实现:

  • ✅ 完美保留 reasoning_content。

  • ✅ 修复工具调用的参数传递问题。

  • ✅ 兼容 LangChain 的 Agent 框架。

4.1 核心代码实现

新建文件 deepseek_reasoner_model.py:

from typing import Optional, List, Dict, Any
from langchain_core.language_models.chat_models import BaseChatModel
from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, BaseMessage, ToolMessage
from langchain_core.outputs import ChatResult, ChatGeneration
from openai import OpenAI

class DeepSeekReasonerChatModel(BaseChatModel):
    """
    自定义 DeepSeek Reasoner 模型适配器
    修复了 LangChain 对 reasoning_content 的兼容性问题
    """
    api_key: str
    base_url: str = "https://api.deepseek.com"
    model_name: str = "deepseek-reasoner"
    temperature: float = 0.7
    bound_tools: Optional[List[Dict]] = None
    _client: Any = None

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self._client = OpenAI(api_key=self.api_key, base_url=self.base_url)

    @property
    def _llm_type(self) -> str:
        return "deepseek-reasoner-custom"

    def bind_tools(self, tools: List[Any], **kwargs):
        """核心修复:正确绑定工具格式"""
        openai_tools = []
        for tool in tools:
            # 兼容 LangChain Tool 对象
            if hasattr(tool, "args_schema"):
                schema = tool.args_schema.model_json_schema()
                openai_tools.append({
                    "type": "function",
                    "function": {
                        "name": tool.name,
                        "description": tool.description,
                        "parameters": schema
                    }
                })
        
        # 返回带有绑定工具的新实例
        new_instance = self.copy()
        new_instance.bound_tools = openai_tools
        return new_instance

    def _convert_messages(self, messages: List[BaseMessage]) -> List[Dict]:
        """将 LangChain 消息转换为 OpenAI 格式,保留 reasoning_content"""
        openai_msgs = []
        for msg in messages:
            if isinstance(msg, HumanMessage):
                openai_msgs.append({"role": "user", "content": msg.content})
            elif isinstance(msg, SystemMessage):
                openai_msgs.append({"role": "system", "content": msg.content})
            elif isinstance(msg, ToolMessage):
                openai_msgs.append({
                    "role": "tool", 
                    "content": msg.content, 
                    "tool_call_id": msg.tool_call_id
                })
            elif isinstance(msg, AIMessage):
                msg_dict = {"role": "assistant", "content": msg.content or ""}
                # 关键:处理 tool_calls
                if msg.tool_calls:
                    msg_dict["tool_calls"] = [
                        {
                            "id": tc["id"],
                            "type": "function",
                            "function": {
                                "name": tc["name"],
                                "arguments": json.dumps(tc["args"])
                            }
                        } for tc in msg.tool_calls
                    ]
                # 关键:恢复 reasoning_content(如果历史记录里有)
                if "reasoning_content" in msg.additional_kwargs:
                    # 注意:OpenAI API 发送时通常不需要发送 reasoning_content,
                    # 但为了保持上下文一致性,部分场景可能需要,此处暂存。
                    pass 
                openai_msgs.append(msg_dict)
        return openai_msgs

    def _generate(self, messages: List[BaseMessage], stop=None, run_manager=None, **kwargs) -> ChatResult:
        """核心生成逻辑"""
        openai_messages = self._convert_messages(messages)
        
        params = {
            "model": self.model_name,
            "messages": openai_messages,
            "temperature": self.temperature,
        }
        
        if self.bound_tools:
            params["tools"] = self.bound_tools

        # 调用 API
        response = self._client.chat.completions.create(**params)
        choice = response.choices[0]
        message = choice.message

        # 构造返回结果
        additional_kwargs = {}
        # 捕获 reasoning_content
        if hasattr(message, 'reasoning_content') and message.reasoning_content:
            additional_kwargs['reasoning_content'] = message.reasoning_content

        # 解析 tool_calls
        tool_calls_list = []
        if message.tool_calls:
            for tc in message.tool_calls:
                tool_calls_list.append({
                    "name": tc.function.name,
                    "args": json.loads(tc.function.arguments),
                    "id": tc.id
                })

        ai_message = AIMessage(
            content=message.content or "",
            tool_calls=tool_calls_list,
            additional_kwargs=additional_kwargs
        )

        return ChatResult(generations=[ChatGeneration(message=ai_message)])
4.2 使用自定义模型运行 Agent

现在,我们可以使用这个“满血版”的模型来运行我们的 Agent 了。

from deepseek_reasoner_model import DeepSeekReasonerChatModel

# 1. 初始化自定义模型
custom_model = DeepSeekReasonerChatModel(
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    model_name="deepseek-reasoner" # 使用 Specilae 模型
)

# 2. 绑定工具并创建 Agent
tools = [get_weather_tool]
agent = create_tool_calling_agent(custom_model, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# 3. 执行并打印思考过程
result = agent_executor.invoke({"input": "北京今天比沈阳冷吗?"})

# 4. 提取思考过程 (Magic Time!)
for msg in result['messages']: # 注意:根据你的 Agent 返回结构调整
    if isinstance(msg, AIMessage) and 'reasoning_content' in msg.additional_kwargs:
        print(f"\n🧠 深度思考:\n{msg.additional_kwargs['reasoning_content'][:200]}...\n")

成功输出示例:

🧠 深度思考:
用户想要比较北京和沈阳的气温。我需要分别查询这两个城市的天气。
Step 1: 调用 get_weather 查询 Beijing。
Step 2: 调用 get_weather 查询 Shenyang。
Step 3: 对比温度数据。
...

AI: 根据查询结果,北京当前温度 -4°C,沈阳当前温度 -13°C。沈阳明显比北京更冷,温差约为 9 度...

总结与展望

通过本文,我们不仅学会了如何使用 DeepSeek-V3.2 的基础 API,还深入到了 Agent 开发的核心——Function Calling。最重要的是,我们通过自定义 BaseChatModel,成功攻克了 LangChain 适配 DeepSeek-V3.2-Reasoner 的难题。

核心知识点回顾:

  1. DeepSeek-V3.2:高性价比、DSA 架构、Agent 性能强悍。

  2. Function Calling:Agent 与世界交互的桥梁,理解 Input/Output 循环是关键。

  3. LangChain 适配:对于非标准 OpenAI 字段(如 reasoning_content),继承并重写 _generate 是最灵活的解决方案。

DeepSeek 的出现,让“强推理 + 低成本”的 Agent 落地成为可能。 

Logo

有“AI”的1024 = 2048,欢迎大家加入2048 AI社区

更多推荐