DeepAgents 深度解析：从概念到代码验证的完整指南

uv add deepagents tavily-python python-dotenv langchain-deepseek

DEEPSEEK_API_KEY=sk-xxx
TAVILY_API_KEY=tvly-dev-xxx

import os
from typing import Literal
from tavily import TavilyClient
from deepagents import create_deep_agent
from langchain_deepseek import ChatDeepSeek

# 加载环境变量
from dotenv import load_dotenv
load_dotenv()

tavily_client = TavilyClient(api_key=os.environ["TAVILY_API_KEY"])

# 初始化 DeepSeek 模型
deepseek_model = ChatDeepSeek(
    model="deepseek-chat",  # 使用 DeepSeek-V3，支持工具调用和结构化输出
    api_key=os.environ["DEEPSEEK_API_KEY"],
    temperature=0.1,
)

def internet_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,
    )

# System prompt to steer the agent to be an expert researcher
research_instructions = """You are an expert researcher. Your job is to conduct thorough research and then write a polished report.

You have access to an internet search tool as your primary means of gathering information.

## `internet_search`

Use this to run an internet search for a given query. You can specify the max number of results to return, the topic, and whether raw content should be included.
"""

agent = create_deep_agent(
    tools=[internet_search],
    system_prompt=research_instructions,
    model=deepseek_model  # 使用 DeepSeek 模型
)

result = agent.invoke({"messages": [{"role": "user", "content": "What is langgraph?"}]})

# Print the agent's response
print(result["messages"][-1].content)

Based on my research, here's a comprehensive overview of what LangGraph is:

## What is LangGraph?

**LangGraph** is an open-source framework within the LangChain ecosystem designed for building **stateful, multi-actor applications** using Large Language Models (LLMs). It provides a structured approach to creating complex agent and multi-agent workflows with fine-grained control over both the flow and state of applications.

## Key Features and Capabilities

### 1. **State Management**
- Maintains persistent state across multiple agent interactions
- Enables agents to remember context and build upon previous interactions
- Supports complex, long-running workflows that require memory

### 2. **Multi-Agent Coordination**
- Orchestrates multiple AI agents working together
- Manages parallel execution of agents
- Coordinates information sharing between agents through centralized state

### 3. **Graph-Based Architecture**
- Models workflows as directed graphs where nodes represent agents or operations
- Supports conditional workflows that alter execution paths based on agent outputs
- Provides fine-grained control over workflow logic

### 4. **Human-in-the-Loop Integration**
- Allows human intervention and oversight in agent workflows
- Supports real-time monitoring and debugging
- Enables manual approval or modification of agent decisions

## Primary Use Cases

LangGraph is particularly well-suited for:

- **Complex multi-agent systems** where multiple specialized agents need to collaborate
- **Stateful applications** that require memory and context across interactions
- **Custom agent architectures** with unique logic and workflow requirements
- **Production-ready agent systems** that need reliability and monitoring
- **Task automation tools** with complex, conditional workflows

## Technical Foundation

- Built on top of LangChain's ecosystem
- Available in both Python and JavaScript
- Supports integration with various LLM providers
- Provides APIs and tools for deployment and monitoring
- Used by companies like GitLab, Elastic, Cisco, and Vodafone in production

## Why Use LangGraph?

LangGraph addresses the challenges of building sophisticated LLM applications by providing:
- **Structured coordination** of multiple agents
- **State persistence** across complex workflows
- **Fine-grained control** over agent behavior
- **Production-ready** deployment capabilities
- **Flexible architecture** for custom requirements

In essence, LangGraph moves beyond simple LLM integrations to enable the creation of truly sophisticated, stateful agent systems capable of handling complex, real-world tasks that require coordination, memory, and conditional logic

import os
import logging
import json
from typing import Literal
from tavily import TavilyClient
from deepagents import create_deep_agent
from langchain_deepseek import ChatDeepSeek
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.messages import BaseMessage
from langchain_core.outputs import LLMResult

# 加载环境变量
from dotenv import load_dotenv
load_dotenv()

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('deepagents.log'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

# 自定义回调处理器来记录模型交互
class DeepSeekLoggingCallback(BaseCallbackHandler):
    """自定义回调处理器，用于记录 DeepSeek 模型的完整交互过程"""
    
    def on_llm_start(self, serialized, prompts, **kwargs):
        """记录模型调用开始"""
        logger.info("=== DeepSeek 模型调用开始 ===")
        try:
            logger.info(f"序列化信息: {json.dumps(serialized, ensure_ascii=False, indent=2, default=str)}")
        except Exception as e:
            logger.info(f"序列化信息: {str(serialized)} (序列化失败: {e})")
        for i, prompt in enumerate(prompts):
            logger.info(f"输入提示 {i+1}: {prompt}")
        if kwargs:
            try:
                logger.info(f"额外参数: {json.dumps(kwargs, ensure_ascii=False, indent=2, default=str)}")
            except Exception as e:
                logger.info(f"额外参数: {str(kwargs)} (序列化失败: {e})")
    
    def on_llm_end(self, response: LLMResult, **kwargs):
        """记录模型调用结束"""
        logger.info("=== DeepSeek 模型调用结束 ===")
        for i, generation in enumerate(response.generations):
            for j, gen in enumerate(generation):
                output_text = gen.text
                logger.info(f"输出结果 {i+1}-{j+1}: {output_text}")
                
                # 记录生成信息（如果可用）
                if hasattr(gen, 'generation_info') and gen.generation_info:
                    try:
                        logger.info(f"生成信息 {i+1}-{j+1}: {json.dumps(gen.generation_info, ensure_ascii=False, indent=2, default=str)}")
                    except Exception as e:
                        logger.info(f"生成信息 {i+1}-{j+1}: {str(gen.generation_info)}")
        
        # 记录使用的 token 数量（如果可用）
        if hasattr(response, 'llm_output') and response.llm_output:
            try:
                logger.info(f"LLM 输出信息: {json.dumps(response.llm_output, ensure_ascii=False, indent=2, default=str)}")
            except Exception as e:
                logger.info(f"LLM 输出信息: {str(response.llm_output)}")
        
        if kwargs:
            try:
                logger.info(f"额外参数: {json.dumps(kwargs, ensure_ascii=False, indent=2, default=str)}")
            except Exception as e:
                logger.info(f"额外参数: {str(kwargs)}")
    
    def on_llm_error(self, error, **kwargs):
        """LLM 出错时调用"""
        logger.error(f"=== DeepSeek 模型调用出错 ===")
        logger.error(f"错误信息: {str(error)}")
        logger.error(f"错误类型: {type(error).__name__}")
        if kwargs:
            logger.error(f"额外参数: {json.dumps(kwargs, ensure_ascii=False, indent=2)}")
    
    def on_llm_new_token(self, token: str, **kwargs):
        """记录新生成的 token"""
        logger.info(f"新生成 token: '{token}'")
        if kwargs:
            logger.info(f"Token 额外参数: {json.dumps(kwargs, ensure_ascii=False, indent=2)}")
    
    def on_chain_start(self, serialized, inputs, **kwargs):
        """记录链调用开始"""
        logger.info("=== 链调用开始 ===")
        logger.info(f"链序列化信息: {json.dumps(serialized, ensure_ascii=False, indent=2)}")
        logger.info(f"链输入: {json.dumps(inputs, ensure_ascii=False, indent=2)}")
        if kwargs:
            logger.info(f"链额外参数: {json.dumps(kwargs, ensure_ascii=False, indent=2)}")
    
    def on_chain_end(self, outputs, **kwargs):
        """记录链调用结束"""
        logger.info("=== 链调用结束 ===")
        logger.info(f"链输出: {json.dumps(outputs, ensure_ascii=False, indent=2)}")
        if kwargs:
            logger.info(f"链额外参数: {json.dumps(kwargs, ensure_ascii=False, indent=2)}")
    
    def on_chain_error(self, error, **kwargs):
        """记录链调用出错"""
        logger.error(f"=== 链调用出错 ===")
        logger.error(f"错误信息: {str(error)}")
        logger.error(f"错误类型: {type(error).__name__}")
        if kwargs:
            logger.error(f"链额外参数: {json.dumps(kwargs, ensure_ascii=False, indent=2)}")
    
    def on_tool_start(self, serialized, input_str, **kwargs):
        """记录工具调用开始"""
        logger.info("=== 工具调用开始 ===")
        logger.info(f"工具序列化信息: {json.dumps(serialized, ensure_ascii=False, indent=2)}")
        logger.info(f"工具输入: {input_str}")
        if kwargs:
            logger.info(f"工具额外参数: {json.dumps(kwargs, ensure_ascii=False, indent=2)}")
    
    def on_tool_end(self, output, **kwargs):
        """记录工具调用结束"""
        logger.info("=== 工具调用结束 ===")
        logger.info(f"工具输出: {output}")
        if kwargs:
            logger.info(f"工具额外参数: {json.dumps(kwargs, ensure_ascii=False, indent=2)}")
    
    def on_tool_error(self, error, **kwargs):
        """记录工具调用出错"""
        logger.error(f"=== 工具调用出错 ===")
        logger.error(f"错误信息: {str(error)}")
        logger.error(f"错误类型: {type(error).__name__}")
        if kwargs:
            logger.error(f"工具额外参数: {json.dumps(kwargs, ensure_ascii=False, indent=2)}")
    
    def on_text(self, text, **kwargs):
        """记录文本输出"""
        logger.info(f"=== 文本输出 ===")
        logger.info(f"文本内容: {text}")
        if kwargs:
            logger.info(f"文本额外参数: {json.dumps(kwargs, ensure_ascii=False, indent=2)}")
    
    def on_retry(self, retry_state, **kwargs):
        """记录重试"""
        logger.info(f"=== 重试操作 ===")
        logger.info(f"重试状态: {retry_state}")
        if kwargs:
            logger.info(f"重试额外参数: {json.dumps(kwargs, ensure_ascii=False, indent=2)}")
    
    def on_custom_event(self, name, data, **kwargs):
        """记录自定义事件"""
        logger.info(f"=== 自定义事件: {name} ===")
        logger.info(f"事件数据: {json.dumps(data, ensure_ascii=False, indent=2)}")
        if kwargs:
            logger.info(f"事件额外参数: {json.dumps(kwargs, ensure_ascii=False, indent=2)}")

# 创建回调处理器实例
deepseek_callback = DeepSeekLoggingCallback()

tavily_client = TavilyClient(api_key=os.environ["TAVILY_API_KEY"])

# 初始化 DeepSeek 模型
logger.info("初始化 DeepSeek 模型...")
deepseek_model = ChatDeepSeek(
    model="deepseek-chat",  # 使用 DeepSeek-V3，支持工具调用和结构化输出
    api_key=os.environ["DEEPSEEK_API_KEY"],
    temperature=0.1,
    callbacks=[deepseek_callback]  # 添加回调处理器
)
logger.info(f"DeepSeek 模型初始化完成: model=deepseek-chat, temperature=0.1")

def internet_search(
    query: str,
    max_results: int = 5,
    topic: Literal["general", "news", "finance"] = "general",
    include_raw_content: bool = False,
):
    """Run a web search"""
    logger.info("=== Tavily 网络搜索开始 ===")
    logger.info(f"搜索参数:")
    logger.info(f"  - 查询内容: '{query}'")
    logger.info(f"  - 最大结果数: {max_results}")
    logger.info(f"  - 搜索主题: '{topic}'")
    logger.info(f"  - 包含原始内容: {include_raw_content}")
    
    try:
        result = tavily_client.search(
            query,
            max_results=max_results,
            include_raw_content=include_raw_content,
            topic=topic,
        )
        
        logger.info("=== Tavily 网络搜索结果 ===")
        results_count = len(result.get('results', []))
        logger.info(f"搜索完成，返回 {results_count} 个结果")
        
        # 记录每个搜索结果的详细信息
        for i, search_result in enumerate(result.get('results', []), 1):
            logger.info(f"结果 {i}:")
            logger.info(f"  - 标题: {search_result.get('title', 'N/A')}")
            logger.info(f"  - URL: {search_result.get('url', 'N/A')}")
            logger.info(f"  - 评分: {search_result.get('score', 'N/A')}")
            
            # 记录内容摘要
            content = search_result.get('content', '')
            if content:
                logger.info(f"  - 内容摘要: {content}")
            else:
                logger.info(f"  - 内容摘要: 无内容")
            
            # 如果包含原始内容，也记录完整内容
        if include_raw_content and 'raw_content' in search_result:
            raw_content = search_result.get('raw_content', '')
            if raw_content:
                logger.info(f"  - 原始内容: {raw_content}")
            else:
                logger.info(f"  - 原始内容: 无原始内容")
        
        # 记录搜索查询信息（如果可用）
        if 'query' in result:
            logger.info(f"实际执行的查询: '{result['query']}'")
        
        logger.info("=== Tavily 网络搜索完成 ===")
        return result
        
    except Exception as e:
        logger.error(f"=== Tavily 网络搜索失败 ===")
        logger.error(f"错误信息: {str(e)}")
        logger.error(f"搜索参数: query='{query}', max_results={max_results}, topic='{topic}'")
        raise

# System prompt to steer the agent to be an expert researcher
research_instructions = """You are an expert researcher. Your job is to conduct thorough research and then write a polished report.

You have access to an internet search tool as your primary means of gathering information.

## `internet_search`

Use this to run an internet search for a given query. You can specify the max number of results to return, the topic, and whether raw content should be included.
"""

logger.info("=== 创建 DeepAgent ===")
agent = create_deep_agent(
    tools=[internet_search],
    system_prompt=research_instructions,
    model=deepseek_model  # 使用 DeepSeek 模型
)
logger.info("DeepAgent 创建完成")

user_query = "What is langgraph?"
logger.info(f"=== 开始执行用户查询 ===")
logger.info(f"用户问题: '{user_query}'")

result = agent.invoke({"messages": [{"role": "user", "content": user_query}]})

logger.info("=== 用户查询执行完成 ===")
logger.info("=== Agent 最终响应 ===")

# 记录完整的对话历史
for i, message in enumerate(result["messages"]):
    role = message.role if hasattr(message, 'role') else getattr(message, 'type', 'unknown')
    content = message.content if hasattr(message, 'content') else str(message)
    
    logger.info(f"消息 {i+1} ({role}):")
    logger.info(f"  内容: {content}")

# Print the agent's response
final_response = result["messages"][-1].content
logger.info("=== 输出最终结果给用户 ===")
print("\n" + "="*50)
print("DeepAgent 响应:")
print("="*50)
print(final_response)
print("="*50)

System: You are an expert researcher. Your job is to conduct thorough research and then write a polished report.

You have access to an internet search tool as your primary means of gathering information.

## `internet_search`

Use this to run an internet search for a given query. You can specify the max number of results to return, the topic, and whether raw content should be included.


In order to complete the objective that the user asks of you, you have access to a number of standard tools.

## `write_todos`

You have access to the `write_todos` tool to help you manage and plan complex objectives.
Use this tool for complex objectives to ensure that you are tracking each necessary step and giving the user visibility into your progress.
This tool is very helpful for planning complex objectives, and for breaking down these larger complex objectives into smaller steps.

It is critical that you mark todos as completed as soon as you are done with a step. Do not batch up multiple steps before marking them as completed.
For simple objectives that only require a few steps, it is better to just complete the objective directly and NOT use this tool.
Writing todos takes time and tokens, use it when it is helpful for managing complex many-step problems! But not for simple few-step requests.

## Important To-Do List Usage Notes to Remember
- The `write_todos` tool should never be called multiple times in parallel.
- Don't be afraid to revise the To-Do list as you go. New information may reveal new tasks that need to be done, or old tasks that are irrelevant.

## Filesystem Tools `ls`, `read_file`, `write_file`, `edit_file`

You have access to a filesystem which you can interact with using these tools.
All file paths must start with a /.

- ls: list all files in the filesystem
- read_file: read a file from the filesystem
- write_file: write to a file in the filesystem
- edit_file: edit a file in the filesystem

## `task` (subagent spawner)

You have access to a `task` tool to launch short-lived subagents that handle isolated tasks. These agents are ephemeral — they live only for the duration of the task and return a single result.

When to use the task tool:
- When a task is complex and multi-step, and can be fully delegated in isolation
- When a task is independent of other tasks and can run in parallel
- When a task requires focused reasoning or heavy token/context usage that would bloat the orchestrator thread
- When sandboxing improves reliability (e.g. code execution, structured searches, data formatting)
- When you only care about the output of the subagent, and not the intermediate steps (ex. performing a lot of research and then returned a synthesized report, performing a series of computations or lookups to achieve a concise, relevant answer.)

Subagent lifecycle:
1. **Spawn** → Provide clear role, instructions, and expected output
2. **Run** → The subagent completes the task autonomously
3. **Return** → The subagent provides a single structured result
4. **Reconcile** → Incorporate or synthesize the result into the main thread

When NOT to use the task tool:
- If you need to see the intermediate reasoning or steps after the subagent has completed (the task tool hides them)
- If the task is trivial (a few tool calls or simple lookup)
- If delegating does not reduce token usage, complexity, or context switching
- If splitting would add latency without benefit

## Important Task Tool Usage Notes to Remember
- Whenever possible, parallelize the work that you do. This is true for both tool_calls, and for tasks. Whenever you have independent steps to complete - make tool_calls, or kick off tasks (subagents) in parallel to accomplish them faster. This saves time for the user, which is incredibly important.
- Remember to use the `task` tool to silo independent tasks within a multi-part objective.
- You should use the `task` tool whenever you have a complex task that will take multiple steps, and is independent from other tasks that the agent needs to complete. These agents are highly competent and efficient.
Human: What is langgraph?