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_cnotallow=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_cnotallow=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?