DeepSeek Agent 版本对比与使用指南

📋 概述

本文档详细对比了DeepSeek Agent在不同LangChain版本下的实现差异，并提供了现代化的使用方法和最佳实践。

📊 新老版本差异对比

1. 导入方式的演变

版本类型	导入方式	状态	说明
老版本	from langchain.schema import HumanMessage, SystemMessage	❌ 已弃用	会产生警告信息
新版本	from langchain_core.messages import HumanMessage, SystemMessage	✅ 推荐	官方推荐路径

2. API调用方式

版本类型	调用方式	状态	说明
老版本	response = llm(messages)	⚠️ 弃用警告	仍可用但不推荐
新版本	response = llm.invoke(messages)	✅ 推荐	新的标准调用方式

3. 配置参数变化

参数功能	老版本参数名	新版本参数名	说明
API端点	openai_api_base	base_url	更简洁的命名
API密钥	openai_api_key	api_key	统一的参数名

4. 工具调用方式对比

🔴 手动方式 (当前使用)

# 手动JSON解析方式
tools_description = """
Available tools:
- get_weather(city: str): Get weather information for a given city

When you need to use a tool, respond with a JSON object in this format:
{"action": "get_weather", "parameters": {"city": "city_name"}}
"""

# 手动解析JSON响应
if '{"action":' in response_content:
    tool_call = json.loads(json_str)
    action = tool_call.get("action")
    parameters = tool_call.get("parameters", {})

优点:

• 兼容性好，适用于各种LangChain版本
• 逻辑清晰，易于理解和调试
• 对模型要求较低

缺点:

• 代码冗长，需要手动处理JSON解析
• 错误处理复杂
• 不支持复杂的工具链

🟢 现代化方式 (推荐)

# 使用@tool装饰器
@tool
def get_weather(city: str) -> str:
    """Get weather for a given city."""
    return f"It's always sunny in {city}!"

# 使用create_agent
agent = create_agent(
    model=llm,
    tools=[get_weather],
    system_prompt="You are a helpful assistant."
)

优点:

• 代码简洁，自动处理工具调用
• 内置错误处理和类型检查
• 支持复杂的工具链和并行调用
• 更好的调试和监控支持

缺点:

• 依赖较新的LangChain版本
• 学习曲线稍陡

⚠️ 常见问题和解决方案

1. 依赖版本冲突

问题:

ImportError: cannot import name 'model_validator' from 'pydantic'

原因: pydantic版本过旧，与新版LangChain不兼容

解决方案:

pip install --upgrade pydantic>=2.0.0

2. 导入路径变更

问题:

ModuleNotFoundError: No module named 'langchain.schema'

解决方案:

# 旧版本
from langchain.schema import HumanMessage, SystemMessage

# 新版本
from langchain_core.messages import HumanMessage, SystemMessage

3. Agent创建方法不支持

问题:

ImportError: cannot import name 'create_tool_calling_agent'

原因: LangChain版本虽然较新，但某些特性未包含

解决方案: 使用兼容的 create_agent 方法或手动实现

🚀 推荐的新版本使用方法

1. 环境准备

# 创建虚拟环境
python -m venv venv

# 激活环境 (Windows)
.\venv\Scripts\activate

# 激活环境 (Linux/Mac)  
source venv/bin/activate

2. 安装最新版本库

# 升级pip
python -m pip install --upgrade pip

# 安装核心包
pip install langchain>=0.1.0
pip install langchain-openai>=0.1.0
pip install langchain-core>=0.1.0
pip install openai>=1.0.0

# 可选：安装其他工具
pip install langchain-community
pip install langsmith  # 用于调试和监控
pip install python-dotenv  # 环境变量管理

3. 现代化代码模板

"""
DeepSeek Agent - 集成智谱Web Search MCP服务
核心功能：让大模型具备实时联网搜索能力，获取最新信息
"""
import os
import requests
import json
from datetime import datetime
from dotenv import load_dotenv  # 加载环境变量
from sseclient import SSEClient  # 解析SSE流式响应
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage
from langchain.agents import create_agent

# 加载.env文件中的环境变量（优先于硬编码）
load_dotenv()

# -------------------------- 1. 定义基础工具 --------------------------
@tool
def get_weather(city: str) -> str:
    """Get weather information for a given city (e.g., "Beijing")."""
    # 示例：返回模拟天气（实际可替换为真实天气API）
    return f"It's always sunny in {city}!"


@tool
def get_time() -> str:
    """Get current system time (format: YYYY-MM-DD HH:MM:SS)."""
    return f"Current time: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"


# -------------------------- 2. 核心：封装智谱Web Search MCP工具 --------------------------
@tool
def zhipu_websearch(query: str) -> str:
    """
    Search the web using Zhipu Web Search MCP (SSE streaming service).
    核心用途：获取实时信息、最新新闻、网页内容检索（大模型知识库未覆盖的内容）
    输入示例："2024 Olympics closing date"、"最新AI发展趋势"、"2024诺贝尔物理学奖得主"
    """
    # 步骤1：校验MCP密钥是否配置
    mcp_api_key = os.getenv("DASHSCOPE_API_KEY")
    if not mcp_api_key:
        return "❌ 错误：未配置DASHSCOPE_API_KEY环境变量，请在.env文件中添加阿里云百炼密钥。"

    # 步骤2：构造MCP服务请求参数
    mcp_url = "https://dashscope.aliyuncs.com/api/v1/mcps/zhipu-websearch/sse"  # MCP服务端点
    headers = {
        "Authorization": f"Bearer {mcp_api_key}",  # 密钥认证
        "Content-Type": "application/json"
    }
    payload = {
        "query": query,          # 用户搜索关键词
        "max_results": 3         # 控制返回结果数量（可调整为1-5）
    }

    try:
        # 步骤3：发送POST请求（开启流式传输stream=True）
        response = requests.post(
            mcp_url,
            headers=headers,
            json=payload,
            stream=True,          # 关键：启用流式响应（SSE依赖）
            timeout=60            # 延长超时（搜索可能耗时较长）
        )
        response.raise_for_status()  # 捕获HTTP错误（如401密钥无效、403权限不足）

        # 步骤4：解析SSE流式数据
        client = SSEClient(response)  # 初始化SSE客户端
        search_results = []           # 存储所有搜索结果

        for event in client.events():
            # 每个event.data是JSON字符串，需解析为字典
            event_data = json.loads(event.data)

            # 提取有效结果（根据MCP官方返回格式调整，避免空数据）
            if "results" in event_data and isinstance(event_data["results"], list):
                for result in event_data["results"]:
                    # 提取结果核心字段（标题、链接、摘要）
                    title = result.get("title", "无标题")
                    url = result.get("url", "无链接")
                    summary = result.get("summary", "无摘要")
                    # 格式化结果（便于大模型阅读和用户理解）
                    formatted_result = f"- **{title}**\n  链接：{url}\n  摘要：{summary}\n"
                    search_results.append(formatted_result)

        # 步骤5：结果去重+整合（避免重复结果）
        unique_results = list(dict.fromkeys(search_results))  # 保持插入顺序去重

        if unique_results:
            return f"🔍 搜索结果 for '{query}'：\n\n" + "\n".join(unique_results)
        else:
            return f"🔍 搜索 '{query}' 未找到相关结果（可能关键词过于模糊或网络异常）。"

    # 捕获网络异常（如超时、连接失败）
    except requests.exceptions.RequestException as e:
        return f"❌ MCP搜索服务调用失败：{str(e)}（请检查网络连接或MCP服务状态）。"
    # 捕获JSON解析或其他未知异常
    except Exception as e:
        return f"❌ 搜索结果解析异常：{str(e)}（可能MCP返回格式更新，请检查代码适配）。"


# -------------------------- 3. 创建集成MCP的DeepSeek Agent --------------------------
def create_deepseek_agent():
    """创建包含MCP搜索工具的Agent"""
    # 步骤1：配置DeepSeek大模型（兼容OpenAI接口格式）
    llm = ChatOpenAI(
        model="deepseek-chat",       # DeepSeek聊天模型（支持工具调用）
        base_url="https://api.deepseek.com/v1",  # DeepSeek API端点
        api_key=os.getenv("DEEPSEEK_API_KEY"),   # 从环境变量获取密钥
        temperature=0.1,             # 降低随机性（工具调用需精准）
        max_tokens=2000,             # 增加最大 tokens（容纳长搜索结果）
        timeout=30
    )

    # 步骤2：整合所有工具（基础工具+MCP搜索工具）
    tools = [get_weather, get_time, zhipu_websearch]

    # 步骤3：创建Agent（自定义系统提示词，指导Agent正确选择工具）
    agent = create_agent(
        model=llm,
        tools=tools,
        system_prompt="""You are a helpful assistant with access to 3 tools. 
        Strictly choose the appropriate tool based on user questions:
        1. Weather query: Use get_weather (input: city name, e.g., "Shanghai").
        2. Current time query: Use get_time (no input needed).
        3. Real-time information/web search: Use zhipu_websearch (input: clear search query, e.g., "2024 Nobel Prize in Physics winner").
        Rules:
        - Always use tools for questions that require real-time data or web content (don't rely on your internal knowledge).
        - After getting search results, summarize them clearly and cite sources (include links if possible).
        - If a tool call fails, inform the user of the error and suggest alternative questions.
        - For questions unrelated to weather, time, or real-time info, answer directly with your knowledge.
        """
    )

    return agent


# -------------------------- 4. 安全调用Agent（含错误处理） --------------------------
def safe_invoke(agent, user_input: str) -> str:
    """安全调用Agent，捕获所有异常并返回友好提示"""
    try:
        # 构造Agent输入（符合LangChain Agent的消息格式）
        result = agent.invoke({
            "messages": [HumanMessage(content=user_input)]
        })
        # 返回Agent最后一条消息（即最终回答）
        return result["messages"][-1].content
    except Exception as e:
        return f"⚠️ Agent调用失败：{str(e)}（请检查大模型密钥、网络或工具配置）。"


# -------------------------- 5. 主函数（测试集成效果） --------------------------
def main():
    print("🚀 DeepSeek Agent（集成智谱Web Search MCP）启动中...")
    print("💡 支持功能：天气查询、时间查询、实时联网搜索（如新闻、赛事结果、科技趋势）\n")

    # 创建Agent
    agent = create_deepseek_agent()

    # 测试用例（覆盖不同工具调用场景）
    test_cases = [
        "What's the weather in Guangzhou?",  # 调用get_weather
        "What time is it now?",              # 调用get_time
        "Who won the 2024 Nobel Prize in Physics?",  # 调用zhipu_websearch（实时信息）
        "What's the latest news about AI in 2024?",  # 调用zhipu_websearch（最新新闻）
        "Tell me a joke about programming."  # 不调用工具（大模型直接回答）
    ]

    # 执行测试
    for i, question in enumerate(test_cases, 1):
        print(f"--- 测试用例 {i} ---")
        print(f"❓ 用户问题：{question}")
        response = safe_invoke(agent, question)
        print(f"🤖 Agent回答：{response}\n")


if __name__ == "__main__":
    main()

🛠️ 最佳实践建议

1. 安全性

# 使用环境变量管理敏感信息
import os
from dotenv import load_dotenv

load_dotenv()
api_key = os.getenv("DEEPSEEK_API_KEY")

# .env 文件内容
# DEEPSEEK_API_KEY=your-actual-api-key-here

2. 依赖管理

# 生成requirements.txt
pip freeze > requirements.txt

# 或者手动创建clean版本
cat > requirements.txt << EOF
langchain>=0.1.0
langchain-openai>=0.1.0  
langchain-core>=0.1.0
openai>=1.0.0
python-dotenv>=1.0.0
EOF

3. 开发工具

# 安装开发工具
pip install langsmith  # LangChain调试工具
pip install jupyter    # 交互式开发
pip install black      # 代码格式化
pip install pytest     # 单元测试

4. 错误处理模式

def robust_agent_call(agent, user_input: str, max_retries: int = 3) -> str:
    """带重试机制的Agent调用"""
    for attempt in range(max_retries):
        try:
            result = agent.invoke({"messages": [HumanMessage(content=user_input)]})
            return result['messages'][-1].content
        except Exception as e:
            if attempt == max_retries - 1:
                return f"❌ 调用失败: {e}"
            print(f"⚠️ 第{attempt + 1}次尝试失败，重试中...")
            time.sleep(2 ** attempt)  # 指数退避

📈 性能优化建议

1. 连接池配置

llm = ChatOpenAI(
    model="deepseek-chat",
    base_url="https://api.deepseek.com/v1",
    api_key="your-api-key-here",
    temperature=0,
    max_retries=3,
    timeout=30,
    # 连接池配置
    http_client=httpx.Client(
        limits=httpx.Limits(max_connections=10, max_keepalive_connections=5)
    )
)

2. 缓存机制

from langchain_core.caches import InMemoryCache
from langchain_core.globals import set_llm_cache

# 启用缓存
set_llm_cache(InMemoryCache())

📋 版本选择建议

✅ 推荐使用新版本的原因

1. 更好的类型提示和IDE支持
2. 更稳定的API，减少破坏性变更
3. 更丰富的功能，如内置错误处理
4. 更好的性能和内存管理
5. 官方长期支持