🎨 DeepAgents 定制化

"""
模型选择 - 使用模型字符串
"""
from deepagents import create_deep_agent

# 使用 GPT-4o
agent_gpt = create_deep_agent(model="gpt-4o")

# 使用 Claude Opus
agent_opus = create_deep_agent(model="claude-opus-4-5-20251101")

# 使用 Gemini
agent_gemini = create_deep_agent(model="gemini-2.0-flash-exp")

"""
模型选择 - 使用 LangChain 模型对象（更精细控制）
"""
from langchain import init_chat_model
from deepagents import create_deep_agent

# 初始化模型对象，配置参数
model = init_chat_model(
    model="gpt-4o",
    model_provider="openai",
    temperature=0.7,          # 控制随机性
    max_tokens=4096,          # 最大输出长度
    timeout=60.0              # 超时设置
)

# 创建 DeepAgent
agent = create_deep_agent(model=model)

# 调用 Agent
result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "分析当前 AI 技术趋势"
    }]
})

"""
系统提示词定制 - 研究助手
"""
from deepagents import create_deep_agent

research_instructions = """你是一位专业的研究助手。你的任务是进行深入研究，然后撰写精炼的报告。

## 核心职责

1. **信息收集**: 使用网络搜索工具收集权威、最新的信息
2. **深度分析**: 对收集到的信息进行批判性分析和综合
3. **结构化输出**: 生成清晰、有条理的研究报告

## 工作流程

1. **理解需求**: 仔细分析用户的研究问题
2. **制定计划**: 使用 write_todos 工具分解研究任务
3. **执行搜索**: 进行多次有针对性的搜索，覆盖不同角度
4. **信息管理**: 使用文件系统工具保存和组织大量数据
5. **撰写报告**: 综合信息，生成结构化报告

## 输出格式

# [主题] 研究报告

## 执行摘要
[3-5 句话总结核心发现]

## 详细分析
[分段展开，包含数据和引用]

## 关键发现
- 发现 1
- 发现 2
- ...

## 参考来源
- 来源 1
- 来源 2
- ...

## 注意事项

- 优先使用权威来源
- 交叉验证重要信息
- 标注信息来源
- 保持客观中立
"""

# 创建研究助手 Agent
research_agent = create_deep_agent(
    model="gpt-4o",
    system_prompt=research_instructions
)

# 使用 Agent
result = research_agent.invoke({
    "messages": [{
        "role": "user",
        "content": "研究 2024 年 AI Agent 的最新进展"
    }]
})

print(result["messages"][-1].content)

"""
系统提示词定制 - 代码审查助手
"""
from deepagents import create_deep_agent

code_review_instructions = """你是一位资深的代码审查专家，专注于 Python 代码质量。

## 核心能力

1. **代码分析**: 深入理解代码逻辑和架构
2. **问题识别**: 发现潜在的 bug、性能问题和安全隐患
3. **最佳实践**: 推荐符合 Python 最佳实践的改进方案
4. **文档评估**: 检查代码注释和文档的完整性

## 审查标准

### 1. 代码质量
- 可读性和可维护性
- 命名规范（PEP 8）
- 函数和类的设计
- 复杂度控制

### 2. 性能优化
- 算法效率
- 数据结构选择
- 资源使用（内存、CPU）
- 并发和异步处理

### 3. 安全性
- 输入验证
- SQL 注入防护
- XSS 防护
- 敏感数据处理

### 4. 测试覆盖
- 单元测试
- 边界条件
- 错误处理

## 输出格式

# 代码审查报告

## 总体评分
- 代码质量: [1-10]
- 性能: [1-10]
- 安全性: [1-10]
- 可维护性: [1-10]

## 主要问题
1. **[严重性]** 问题描述
   - 位置: 文件:行号
   - 原因: ...
   - 建议: ...

## 优点
- ...

## 改进建议
1. ...
2. ...

## 工作流程

1. 使用 read_file 读取代码文件
2. 系统分析代码结构和逻辑
3. 识别问题并分类
4. 生成详细的审查报告
5. 如需深入分析，使用 task 工具创建专门的子 Agent
"""

# 创建代码审查 Agent
code_review_agent = create_deep_agent(
    model="claude-sonnet-4-5-20250929",
    system_prompt=code_review_instructions
)

# 使用 Agent 审查代码
result = code_review_agent.invoke({
    "messages": [{
        "role": "user",
        "content": "请审查 src/main.py 文件中的代码"
    }]
})

"""
添加自定义工具 - 网络搜索
"""
import os
from typing import Literal
from tavily import TavilyClient
from deepagents import create_deep_agent

# 初始化 Tavily 客户端
tavily_client = TavilyClient(api_key=os.environ["TAVILY_API_KEY"])

def internet_search(
    query: str,
    max_results: int = 5,
    topic: Literal["general", "news", "finance"] = "general",
    include_raw_content: bool = False,
) -> dict:
    """
    执行网络搜索

    Args:
        query: 搜索查询字符串
        max_results: 返回的最大结果数（1-10）
        topic: 搜索主题类型
        include_raw_content: 是否包含网页原始内容

    Returns:
        包含搜索结果的字典
    """
    return tavily_client.search(
        query=query,
        max_results=max_results,
        include_raw_content=include_raw_content,
        topic=topic,
    )

# 创建包含自定义工具的 Agent
agent = create_deep_agent(
    model="gpt-4o",
    tools=[internet_search],  # 添加自定义工具
    system_prompt="你是研究助手，可以使用网络搜索收集信息。"
)

# 使用 Agent
result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "搜索 LangChain 的最新功能"
    }]
})

"""
多个自定义工具 - 数据分析 Agent
"""
import os
import sqlite3
from typing import Literal
from tavily import TavilyClient
from deepagents import create_deep_agent

# 工具 1: 网络搜索
tavily_client = TavilyClient(api_key=os.environ["TAVILY_API_KEY"])

def internet_search(query: str, max_results: int = 5) -> dict:
    """搜索网络信息"""
    return tavily_client.search(query, max_results=max_results)

# 工具 2: 数据库查询
def query_database(sql: str) -> list:
    """
    执行 SQL 查询

    Args:
        sql: SQL 查询语句（只读，SELECT 语句）

    Returns:
        查询结果列表
    """
    # 验证是否为只读查询
    if not sql.strip().upper().startswith("SELECT"):
        return {"error": "仅支持 SELECT 查询"}

    try:
        conn = sqlite3.connect("data/analytics.db")
        cursor = conn.cursor()
        cursor.execute(sql)
        results = cursor.fetchall()
        conn.close()
        return results
    except Exception as e:
        return {"error": str(e)}

# 工具 3: 数据可视化
def create_chart(data: list, chart_type: Literal["bar", "line", "pie"]) -> str:
    """
    创建数据图表

    Args:
        data: 图表数据
        chart_type: 图表类型

    Returns:
        图表文件路径
    """
    # 简化示例：实际会使用 matplotlib 或 plotly
    import json
    chart_path = f"charts/chart_{chart_type}.json"

    # 保存图表数据
    with open(chart_path, "w") as f:
        json.dump({"type": chart_type, "data": data}, f)

    return chart_path

# 工具 4: 发送通知
def send_notification(message: str, channel: Literal["email", "slack"] = "slack") -> dict:
    """
    发送通知消息

    Args:
        message: 通知内容
        channel: 通知渠道

    Returns:
        发送状态
    """
    # 简化示例：实际会调用邮件或 Slack API
    print(f"[{channel.upper()}] {message}")
    return {"status": "sent", "channel": channel}

# 创建数据分析 Agent
data_analyst_prompt = """你是一位数据分析专家。你可以：

1. 使用 query_database 查询数据库
2. 使用 internet_search 搜索外部信息
3. 使用 create_chart 创建可视化图表
4. 使用 send_notification 发送分析报告

工作流程：
1. 理解用户的分析需求
2. 从数据库查询相关数据
3. 必要时搜索外部信息进行对比
4. 创建可视化图表
5. 生成分析报告
6. 发送通知
"""

data_agent = create_deep_agent(
    model="claude-sonnet-4-5-20250929",
    tools=[
        internet_search,
        query_database,
        create_chart,
        send_notification
    ],
    system_prompt=data_analyst_prompt
)

# 使用 Agent 进行数据分析
result = data_agent.invoke({
    "messages": [{
        "role": "user",
        "content": """分析最近 30 天的销售数据：
        1. 查询数据库获取销售数据
        2. 创建趋势图表
        3. 与去年同期对比
        4. 将分析报告发送到 Slack"""
    }]
})

print(result["messages"][-1].content)

"""
完整定制示例 - 技术支持 Agent
"""
import os
from typing import Literal
from langchain import init_chat_model
from tavily import TavilyClient
from deepagents import create_deep_agent

# ========== 1. 定义自定义工具 ==========

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

def search_documentation(query: str, product: Literal["langchain", "langgraph"] = "langchain") -> dict:
    """搜索产品文档"""
    domain_filter = "docs.langchain.com" if product == "langchain" else "langchain-ai.github.io"

    # 限定在官方文档域名内搜索
    return tavily_client.search(
        query=f"{query} site:{domain_filter}",
        max_results=3
    )

def check_system_status() -> dict:
    """检查系统状态"""
    # 简化示例：实际会调用监控 API
    return {
        "status": "operational",
        "services": {
            "api": "up",
            "database": "up",
            "cache": "up"
        },
        "response_time_ms": 45
    }

def create_support_ticket(
    title: str,
    description: str,
    priority: Literal["low", "medium", "high"] = "medium"
) -> dict:
    """创建支持工单"""
    # 简化示例：实际会调用工单系统 API
    ticket_id = f"TICKET-{hash(title) % 10000}"
    return {
        "ticket_id": ticket_id,
        "status": "created",
        "priority": priority
    }

# ========== 2. 配置模型 ==========

model = init_chat_model(
    model="gpt-4o",
    model_provider="openai",
    temperature=0.3,  # 较低温度确保回答准确
    max_tokens=2048
)

# ========== 3. 编写系统提示词 ==========

support_prompt = """你是一位专业的技术支持专家，专注于 LangChain 和 LangGraph 产品。

## 核心职责

1. **问题诊断**: 准确理解用户遇到的技术问题
2. **解决方案**: 提供清晰、可执行的解决步骤
3. **文档引导**: 引用官方文档支持你的建议
4. **升级管理**: 必要时创建支持工单

## 可用工具

- `search_documentation`: 搜索 LangChain/LangGraph 官方文档
- `check_system_status`: 检查系统服务状态
- `create_support_ticket`: 创建技术支持工单

## 工作流程

1. **理解问题**
   - 询问必要的上下文信息
   - 确认问题的具体表现

2. **诊断分析**
   - 检查系统状态（如适用）
   - 搜索相关文档和已知问题

3. **提供解决方案**
   - 给出分步骤的解决方案
   - 引用官方文档链接
   - 提供代码示例（如适用）

4. **升级处理**
   - 如果问题复杂或需要开发团队介入，创建工单
   - 告知用户工单编号和预期响应时间

## 输出格式

### 问题诊断
[简要总结用户的问题]

### 建议方案
1. 步骤 1
2. 步骤 2
...

### 参考文档
- [文档标题](链接)

### 后续跟进
[需要用户做什么，或已创建的工单信息]

## 注意事项

- 保持友好和耐心
- 避免技术术语过于复杂
- 如果不确定，明确说明并建议升级
- 始终引用官方文档
"""

# ========== 4. 创建定制化 DeepAgent ==========

support_agent = create_deep_agent(
    model=model,
    tools=[
        search_documentation,
        check_system_status,
        create_support_ticket
    ],
    system_prompt=support_prompt
)

# ========== 5. 使用 Agent ==========

result = support_agent.invoke({
    "messages": [{
        "role": "user",
        "content": """我在使用 LangChain 1.0 的 create_agent() 时遇到错误：

        TypeError: create_agent() got an unexpected keyword argument 'memory'

        这是什么问题？如何解决？"""
    }]
})

print("=" * 60)
print("技术支持响应：")
print("=" * 60)
print(result["messages"][-1].content)

# Agent 会自动：
# 1. 使用 search_documentation 搜索相关文档
# 2. 发现这是 0.x 到 1.0 迁移问题
# 3. 提供使用 checkpointer 替代 memory 的解决方案
# 4. 引用官方迁移文档
# 5. 如需要，创建支持工单

# 测试定制化 Agent 的流程

# 1. 单元测试工具
def test_internet_search():
    result = internet_search("LangChain", max_results=2)
    assert "results" in result
    assert len(result["results"]) <= 2

# 2. 测试系统提示词效果
test_cases = [
    "简单问题",
    "复杂的多步骤任务",
    "边界情况",
    "错误输入"
]

for test_case in test_cases:
    result = agent.invoke({"messages": [{"role": "user", "content": test_case}]})
    # 验证输出质量

# 3. 迭代优化
# - 分析失败案例
# - 调整系统提示词
# - 添加或修改工具
# - 重新测试

# 创建多个 Agent 实例
agent_fast = create_deep_agent(model="gemini-2.0-flash-exp", ...)
agent_powerful = create_deep_agent(model="claude-opus-4-5", ...)

# 根据任务选择
if task_complexity == "simple":
    result = agent_fast.invoke(...)
else:
    result = agent_powerful.invoke(...)

import os

# 启用 LangSmith 追踪
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-api-key"

# 所有工具调用和响应会自动记录
result = agent.invoke({...})

# 在 LangSmith UI 中查看：
# - 工具调用序列
# - 每步的输入输出
# - Token 使用情况
# - 执行时间