DeepAgents · 深度智能体

一、DeepAgents 深度研究智能体应用

DeepAgents 是 LangChain 团队开源的一款高性能智能体框架，专为长周期、高复杂度任务设计。基于 LangChain 和 LangGraph 构建，通过内置的任务规划、文件系统、子智能体等能力，让智能体能够更高效地完成复杂、多步骤的任务，而无需开发者从零搭建底层逻辑。

技术架构

本文基于 LangChain DeepAgents + Tavily Search API，快速构建一个可深度研究的智能体应用。其中 Tavily Search API 采用封装为 MCP Server 的方式提供给智能体调用。

实验配置：

• 模型：自建模型（兼容 OpenAI API 格式）
• 搜索 API：Tavily Search API（封装为 MCP Server）
• 整体流程：DeepAgents → MCP Tools → Tavily API

相关资源

• 官方文档：https://docs.langchain.com/oss/python/deepagents/quickstart
• Github 地址：https://github.com/langchain-ai/deepagents
• Tavily Search API 文档：https://docs.tavily.com/documentation/quickstart

主要依赖版本

tavily-python==0.7.12
mcp==1.9.2
langchain==1.1.3
langchain-mcp-adapters==0.1.4
langgraph==1.0.5
fastapi==0.115.14
python-dotenv==1.0.0

环境配置

创建 .env 文件，配置自建模型信息：

OPENAI_BASE_URL=https://your-api-base-url.com/v1
OPENAI_API_KEY=your-api-key
OPENAI_MODEL=your-model-name

关键点说明：

• OPENAI_BASE_URL：自建模型的 API 基础地址（兼容 OpenAI API 格式）
• OPENAI_API_KEY：API 密钥
• OPENAI_MODEL：模型名称
• 使用 python-dotenv 库加载环境变量，避免硬编码敏感信息

二、Tavily Search API 封装 MCP Server

2.1 MCP Server 实现

将 Tavily Search API 封装为 MCP Server，主要封装两个能力：

• Web 网络搜索：web_search 工具
• 网页内容查看：extract 工具

代码实现：

from mcp.server.fastmcp import FastMCP
from typing import Literal
from tavily import TavilyClient

mcp = FastMCP("Web-Search-Server")

tavily_client = TavilyClient(api_key="tvly-xxxx")  # 修改为你的key

@mcp.tool()
def web_search(query: str, max_results: int = 5,
               topic: Literal["general", "news", "finance"] = "general",
               include_raw_content: bool = False):
    """Run a web search"""
    return tavily_client.search(
        query,
        max_results=max_results,
        include_raw_content=include_raw_content,
        topic=topic,
    )
    
@mcp.tool()
def extract(url: str):
    """Extract web page content from URL."""
    return tavily_client.extract(url)

if __name__ == "__main__":
    mcp.settings.port = 6030
    mcp.run("sse")

2.2 启动 MCP Server

启动 MCP Server 后，服务运行在 http://localhost:6030/sse，提供 SSE（Server-Sent Events）传输方式。

三、深度智能体搭建

3.1 创建深度智能体

使用 deepagents 下的 create_deep_agent 创建智能体，参数和使用方式与 LangChain 之前的 Agent 创建几乎相同。MCP 工具可以直接使用 MultiServerMCPClient 抽取的 MCP Tools。

完整实现代码：

import os
from dotenv import load_dotenv
from langchain_core.messages import AIMessageChunk

# 加载 .env 文件中的环境变量
load_dotenv()

# 配置自建模型（必须从 .env 文件中读取）
BASE_URL = os.getenv("OPENAI_BASE_URL")
API_KEY = os.getenv("OPENAI_API_KEY")
MODEL_NAME = os.getenv("OPENAI_MODEL")

# 检查必需的配置项是否存在
if not BASE_URL:
    raise ValueError("OPENAI_BASE_URL 未在 .env 文件中配置")
if not API_KEY:
    raise ValueError("OPENAI_API_KEY 未在 .env 文件中配置")
if not MODEL_NAME:
    raise ValueError("OPENAI_MODEL 未在 .env 文件中配置")

from deepagents import create_deep_agent
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.chat_models import init_chat_model
import asyncio

system_prompt = """
你是一位研究专家。你的工作是根据用户的要求进行彻底的研究，然后写一份润色的报告。
## 工作流程
1. 理解核心需求，识别关键要素。
2. 制定任务规划，明确待办事项。
3. 逐步研究，记录发现，验证结论。
4. 整体结果和审查, 必要时重新制定解决链路。
5. 输出详细研究报告。
"""

async def main():
    ## 初始化模型
    # 创建模型实例，使用自建模型（兼容 OpenAI API 格式）
    llm = init_chat_model(
        model=MODEL_NAME,
        model_provider="openai",
        temperature=0,
        base_url=BASE_URL,
        api_key=API_KEY,
    )
    
    ## 定义MCP Server, 执行前面创建的
    client = MultiServerMCPClient(
        {
            "web-search": {
                "url": "http://localhost:6030/sse",
                "transport": "sse"
            }
        }
    )
    
    ## 获取 MCP Tools
    tools = await client.get_tools()
    
    ## 创建深度智能体
    agent = create_deep_agent(
        model=llm,
        tools=tools,
        system_prompt=system_prompt
    )
    
    ## 执行深度研究，以流式的方式输出
    async for stream_type, chunk in agent.astream(
            input={
                "messages": [
                    {"role": "user", "content": "埃菲尔铁塔与最高建筑相比有多高？"}
                ]
            },
            stream_mode=["updates", "messages"]
    ):
        if stream_type == "messages" and type(chunk[0]) is AIMessageChunk:
            content = chunk[0].content
            if not content:
                continue
            print(content, end="", flush=True)
        elif stream_type == "updates":
            if "model" in chunk:
                model = chunk["model"]
                if "messages" in model:
                    messages = model["messages"]
                    for message in messages:
                        tool_calls = message.tool_calls
                        for tool_call in tool_calls:
                            name = tool_call["name"]
                            args = tool_call["args"]
                            if name == "write_todos":
                                todos = args["todos"]
                                print_todos = []
                                for todo in todos:
                                    todo_content = todo["content"]
                                    status = todo["status"]
                                    print_todos.append(f"> {todo_content}  -->  {status}")
                                if print_todos:
                                    print_todos = "\n".join(print_todos)
                                    print(f"\n> TODO: \n{print_todos}\n\n")
                            else:
                                print(f"\n> Call MCP: {name}, args: {args}\n")

            if "tools" in chunk:
                tools = chunk["tools"]
                if "messages" in tools:
                    messages = tools["messages"]
                    for message in messages:
                        content = message.content
                        if "Updated todo list" in content:
                            continue
                        print("\nMCP Result: \n", content, "\n")

if __name__ == '__main__':
    asyncio.run(main())

关键点说明：

• load_dotenv() 从 .env 文件加载环境变量，避免硬编码敏感信息
• init_chat_model() 使用 model_provider="openai" 指定使用 OpenAI 兼容的 API 格式
• base_url 和 api_key 参数用于连接自建模型服务
• 配置检查确保必需的配置项存在，避免运行时错误

3.2 执行流程说明

智能体的执行流程：

1. 初始化阶段：

   • 从 .env 文件加载模型配置（BASE_URL、API_KEY、MODEL_NAME）
   • 初始化 LLM 模型（自建模型，兼容 OpenAI API 格式）
   • 连接 MCP Server（web-search）
   • 获取 MCP Tools

2. 任务规划阶段：

   • 智能体根据用户问题生成待办事项（TODO list）
   • 每个待办事项标记状态（in_progress、pending、completed）

3. 执行阶段：

   • 逐步执行每个待办事项
   • 调用 MCP Tools（web_search、extract）获取信息
   • 更新待办事项状态

4. 报告生成阶段：

   • 整合所有研究结果
   • 生成详细的研究报告

3.3 执行效果示例

测试问题：埃菲尔铁塔与最高建筑相比有多高？

执行过程：

• 智能体先制定初步的待办事项
• 逐步执行每个待办，调用 MCP Server 进行搜索
• 最后生成详细的研究报告

四、封装为 OpenAI 协议接口

4.1 FastAPI 实现

使用 FastAPI 封装为 OpenAI 协议接口，便于在客户端（如 Cherry Studio、OpenWebUI）使用。

注意：这里主要为实现功能，api_key 写死在程序中为：sk-da4b6cb4a41e4cascascasc9508deb556942（随机生成的），后续使用客户端连接时，需要填写该 api_key。

完整实现代码：

import os
from dotenv import load_dotenv
from langchain_core.messages import AIMessageChunk

# 加载 .env 文件中的环境变量
load_dotenv()

# 配置自建模型（必须从 .env 文件中读取）
BASE_URL = os.getenv("OPENAI_BASE_URL")
API_KEY = os.getenv("OPENAI_API_KEY")
MODEL_NAME = os.getenv("OPENAI_MODEL")

# 检查必需的配置项是否存在
if not BASE_URL:
    raise ValueError("OPENAI_BASE_URL 未在 .env 文件中配置")
if not API_KEY:
    raise ValueError("OPENAI_API_KEY 未在 .env 文件中配置")
if not MODEL_NAME:
    raise ValueError("OPENAI_MODEL 未在 .env 文件中配置")

from fastapi import FastAPI, HTTPException, Header, Depends
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from typing import List, Dict, Any, Optional, Generator
from langchain.chat_models import init_chat_model
from langchain_mcp_adapters.client import MultiServerMCPClient
from deepagents import create_deep_agent
from datetime import datetime
import time, uuid

app = FastAPI(title="OpenAI Compatible Chat API")
api_key = "sk-da4b6cb4a41e4cascascasc9508deb556942"
agent = None

system_prompt = """
你是一位研究专家。你的工作是根据用户的要求进行彻底的研究，然后写一份润色的报告。
## 工作流程
1. 理解核心需求，识别关键要素。
2. 制定任务规划，明确待办事项。
3. 逐步研究，记录发现，验证结论。
4. 整体结果和审查, 必要时重新制定解决链路。
5. 输出详细研究报告。
"""

# 创建模型实例，使用自建模型（兼容 OpenAI API 格式）
llm = init_chat_model(
    model=MODEL_NAME,
    model_provider="openai",
    temperature=0,
    base_url=BASE_URL,
    api_key=API_KEY,
)

**关键点说明：**
- 模型配置从 `.env` 文件读取，便于不同环境使用不同配置
- `init_chat_model()` 支持自建模型，只需提供兼容 OpenAI API 格式的服务端点
- 全局变量 `llm` 在应用启动时初始化，避免重复创建


class ChatCompletionRequest(BaseModel):
    model: str
    messages: List[Dict[str, Any]]
    temperature: Optional[float] = 1.0
    max_tokens: Optional[int] = None
    stream: Optional[bool] = False


class ChatCompletionChoice(BaseModel):
    index: int
    message: List[Dict[str, Any]]
    finish_reason: str


class ChatCompletionResponse(BaseModel):
    id: str
    object: str = "chat.completion"
    created: int
    model: str
    choices: List[ChatCompletionChoice]
    usage: Dict[str, int]


class ChatCompletionChunk(BaseModel):
    id: str
    object: str = "chat.completion.chunk"
    created: int
    model: str
    choices: List[Dict[str, Any]]


async def create_agent():
    '''创建MCP Agent'''
    client = MultiServerMCPClient(
        {
            "web-search": {
                "url": "http://localhost:6030/sse",
                "transport": "sse"
            }
        }
    )
    tools = await client.get_tools()
    return create_deep_agent(
        model=llm,
        tools=tools,
        system_prompt=system_prompt
    )


async def invoke_agent(messages: []):
    global agent
    if not agent:
        agent = await create_agent()
    async for stream_type, chunk in agent.astream({"messages": messages}, stream_mode=["updates", "messages"]):
        if stream_type == "messages" and type(chunk[0]) is AIMessageChunk:
            content = chunk[0].content
            if not content:
                continue
            yield content
        elif stream_type == "updates":
            if "model" in chunk:
                model = chunk["model"]
                if "messages" in model:
                    messages = model["messages"]
                    for message in messages:
                        tool_calls = message.tool_calls
                        for tool_call in tool_calls:
                            name = tool_call["name"]
                            args = tool_call["args"]
                            if name == "write_todos":
                                todos = args["todos"]
                                print_todos = []
                                for todo in todos:
                                    todo_content = todo["content"]
                                    status = todo["status"]
                                    print_todos.append(f"> {todo_content}  -->  {status}")
                                if print_todos:
                                    print_todos = "\n".join(print_todos)
                                    yield f"\n> TODO: \n{print_todos}\n\n"
                            else:
                                yield f"\n> Call MCP Server: {name}, args: {args}\n\n"
            # 客户端暂时不输出工具的输出，如果需要，放开注释即可
            # if "tools" in chunk:
            #     tools = chunk["tools"]
            #     if "messages" in tools:
            #         messages = tools["messages"]
            #         for message in messages:
            #             content = message.content
            #             if "Updated todo list" in content:
            #                 continue
            #             yield f"> MCP Response: \n\n```json\n{content}\n```\n"


async def handle_stream_response(messages: [],
                                 model: str, request_id: str) -> Generator[str, None, None]:
    # 执行 agent
    async for msg in invoke_agent(messages):
        chunk_data = ChatCompletionChunk(
            id=request_id,
            created=int(time.time()),
            model=model,
            choices=[{
                "index": 0,
                "delta": {
                    "content": msg
                },
                "finish_reason": None
            }]
        )
        yield f"data: {chunk_data.model_dump_json()}\n\n"

    final_chunk = ChatCompletionChunk(
        id=request_id,
        created=int(time.time()),
        model=model,
        choices=[{
            "index": 0,
            "delta": {},
            "finish_reason": "stop"
        }]
    )
    yield f"data: {final_chunk.model_dump_json()}\n\n"
    yield "data: [DONE]\n\n"


async def verify_auth(authorization: Optional[str] = Header(None)) -> bool:
    '''验证token'''
    if not authorization:
        return False
    if authorization.startswith("Bearer "):
        token = authorization[7:]
    else:
        token = authorization
    return token == api_key


@app.post("/v1/chat/completions")
async def chat_completions(request: ChatCompletionRequest, auth_result: bool = Depends(verify_auth)):
    # 检查身份验证结果
    if not auth_result:
        raise HTTPException(
            status_code=401,
            detail={
                "error": {
                    "message": "Invalid authentication credentials",
                    "type": "invalid_request_error",
                    "param": None,
                    "code": "invalid_api_key"
                }
            },
            headers={"WWW-Authenticate": "Bearer"}
        )
    ## 暂不支持非流式返回
    if not request.stream:
        raise HTTPException(
            status_code=400,
            detail={
                "error": {
                    "message": "Streaming responses are not implemented in this mock API",
                    "type": "invalid_request_error",
                    "param": "stream",
                    "code": "invalid_parameter"
                }
            }
        )
    try:
        # 触发 agent 并流式返回
        request_id = f"chatcmpl-{uuid.uuid4().hex[:8]}"
        return StreamingResponse(
            handle_stream_response(request.messages, request.model, request_id),
            media_type="text/plain",
            headers={
                "Cache-Control": "no-cache",
                "Connection": "keep-alive",
                "Content-Type": "text/event-stream"
            }
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.get("/v1/models")
async def list_models():
    """列出可用模型"""
    return {
        "object": "list",
        "data": [
            {
                "id": "agent_model",
                "object": "model",
                "created": int(time.time()),
                "owned_by": "agent"
            }
        ]
    }


@app.get("/health")
async def health_check():
    """健康检查"""
    return {"status": "healthy", "timestamp": datetime.now().isoformat()}


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

4.2 关键实现点

1. API Key 验证：通过 verify_auth 函数验证请求的 API Key
2. 流式响应：使用 StreamingResponse 实现流式输出
3. OpenAI 协议兼容：返回格式符合 OpenAI Chat Completions API 规范
4. 工具输出控制：工具输出内容较多，默认不输出给客户端（可通过注释控制）

4.3 启动服务

python api.py

服务启动在 http://0.0.0.0:8000，提供以下端点：

• POST /v1/chat/completions：聊天完成接口
• GET /v1/models：列出可用模型
• GET /health：健康检查

五、使用 Cherry Studio 连接测试

5.1 配置连接

使用任何支持 OpenAI 协议的客户端（如 Cherry Studio、OpenWebUI）连接测试。

Cherry Studio 配置步骤：

1. 添加 OpenAI 连接
2. Base URL：http://localhost:8000/v1
3. API Key：sk-da4b6cb4a41e4cascascasc9508deb556942
4. Model：agent_model

5.2 测试案例

案例 1：埃菲尔铁塔与最高建筑相比有多高？

执行过程：

• 智能体制定待办事项
• 逐步执行搜索任务
• 生成详细研究报告

研究报告包含：

• 埃菲尔铁塔高度：330米（2022年更新）
• 世界最高建筑哈利法塔：828米
• 高度对比分析：哈利法塔比埃菲尔铁塔高出498米，是埃菲尔铁塔高度的2.51倍
• 历史意义对比
• 详细的数据表格和分析

六、技术要点总结

6.1 DeepAgents 核心优势

1. 内置任务规划：自动生成待办事项，跟踪执行进度
2. 文件系统支持：内置虚拟文件系统，支持文件操作
3. 子智能体能力：可以派生子智能体处理复杂任务
4. 长周期任务支持：专为复杂、多步骤任务设计

6.2 MCP 协议优势

1. 标准化工具接口：统一的工具调用协议
2. 多服务器支持：可以连接多个 MCP Server
3. 灵活传输方式：支持 SSE、WebSocket 等传输方式
4. 易于扩展：可以轻松添加新的工具能力

6.3 整体架构优势

1. 模块化设计：MCP Server、DeepAgents、API 接口分离
2. 易于部署：可以独立部署各个组件
3. 客户端兼容：支持任何 OpenAI 协议兼容的客户端
4. 流式输出：实时展示智能体执行过程

七、总结

通过 LangChain DeepAgents + MCP 构建深度研究型智能体应用，实现了：

1. 强大的研究能力：能够进行深度、多步骤的研究任务
2. 灵活的工具集成：通过 MCP 协议轻松集成各种工具
3. 良好的用户体验：支持流式输出，实时展示执行过程
4. 标准化接口：通过 OpenAI 协议兼容各种客户端

这种架构为构建复杂的 AI 应用提供了强大的基础，特别适合需要深度研究、多步骤执行的任务场景。