# DeepAgents 技术架构层次
┌──────────────────────────────────┐
│        应用层 (Applications)       │
│  对话系统 │ 数据分析 │ 自动化工作流  │
├──────────────────────────────────┤
│       代理层 (Agent Runtime)       │
│  Agent  │ Pipeline  │ Router      │
├──────────────────────────────────┤
│      编排层 (LangGraph Engine)     │
│  StateGraph │ Node │ Edge │ State │
├──────────────────────────────────┤
│     工具层 (LangChain Tools)       │
│  Search │ Calculator │ Code │ ... │
├──────────────────────────────────┤
│      模型层 (LLM Providers)        │
│  OpenAI │ Anthropic │ Local LLM   │
└──────────────────────────────────┘

# 安装 DeepAgents
pip install deepagents langchain langgraph

# 创建第一个智能代理
from deepagents import Agent, LLM, Tool

# 定义工具
@Tool.define
def get_weather(city: str) -> str:
    """获取指定城市的天气信息"""
    return f"{city}今天晴，温度25°C"

# 创建代理
agent = Agent(
    name="weather_assistant",
    llm=LLM(model="gpt-4"),
    tools=[get_weather],
    system_prompt="你是一个友好的天气助手"
)

# 运行代理
response = agent.run("北京今天天气怎么样？")
print(response)
# 输出：北京今天天气晴朗，温度25°C，适合出门活动！

# 多轮对话
response2 = agent.chat("那上海呢？")
print(response2)
# 输出：上海今天晴，温度25°C...（代理自动理解上下文）

# 传统编程：硬编码所有逻辑
def process_order(order):
    if order.type == "standard":
        return standard_shipping(order)
    elif order.type == "express":
        return express_shipping(order)
    elif order.type == "international":
        return international_shipping(order)
    else:
        return unknown_order_handler(order)
    # 每新增一种订单类型，都要修改这个函数

# 代理编程：让 AI 自主决策
agent = Agent(
    name="order_processor",
    llm=LLM(model="gpt-4"),
    tools=[standard_shipping, express_shipping,
           international_shipping, ask_customer],
    system_prompt="你处理各种订单，自主选择最佳处理方式"
)

# 无需硬编码所有分支，代理根据上下文自主决策
result = agent.run("处理这个国际快递订单，需要加急")
# 代理自动选择 international_shipping + express 选项

# Agent ReAct 循环伪代码
class Agent:
    def run(self, user_input):
        state = {"input": user_input, "history": []}

        while not state.get("finished"):
            # 1. 思考：分析当前状态
            thought = self.llm.think(
                state["input"],
                state["history"],
                available_tools=self.tools
            )

            # 2. 决策：选择动作
            action = self.llm.decide(thought)

            if action.type == "tool_call":
                # 3. 执行：调用工具
                result = self.tools[action.name].run(
                    action.arguments
                )
                state["history"].append({
                    "thought": thought,
                    "action": action,
                    "observation": result
                })
            elif action.type == "final_answer":
                # 4. 输出：返回最终答案
                state["finished"] = True
                return action.content

        return "无法完成任务"

from langgraph.graph import StateGraph, END

# 定义状态结构
class AgentState(TypedDict):
    messages: list[Message]
    current_tool: str | None
    tool_result: Any
    iteration: int
    max_iterations: int

# 构建状态图
graph = StateGraph(AgentState)

# 添加节点
graph.add_node("think", think_node)      # 思考
graph.add_node("decide", decide_node)    # 决策
graph.add_node("execute", execute_node)  # 执行
graph.add_node("respond", respond_node)  # 回复

# 添加边（状态转换）
graph.add_edge("think", "decide")
graph.add_conditional_edges(
    "decide",
    should_continue,
    {
        "tool_call": "execute",
        "final_answer": "respond",
        "max_iterations": "respond",
    }
)
graph.add_edge("execute", "think")  # 执行后回到思考
graph.add_edge("respond", END)       # 回复后结束

# 设置入口
graph.set_entry_point("think")

# 编译并运行
app = graph.compile()
result = app.invoke({"messages": [user_message]})

from langchain.chains import SequentialChain, RouterChain

# 1. 顺序链：步骤按固定顺序执行
qa_pipeline = SequentialChain(
    chains=[
        question_analyzer,   # 分析问题类型和意图
        context_retriever,   # 检索相关上下文（RAG）
        answer_generator,    # 基于上下文生成答案
        fact_checker,        # 验证答案的事实准确性
    ],
    input_variables=["question"],
    output_variables=["answer", "confidence", "sources"]
)

# 2. 路由链：根据输入类型分发
router = RouterChain(
    destinations={
        "code": code_helper_chain,
        "math": math_solver_chain,
        "creative": creative_writing_chain,
        "general": general_qa_chain,
    },
    default_chain=general_qa_chain,
    llm=LLM(model="gpt-4")
)

# 3. 自定义链：组合任意逻辑
class ValidationChain(Chain):
    def _call(self, inputs):
        answer = inputs["answer"]
        # 交叉验证：用另一个 LLM 验证答案
        validation = self.validator_llm.run(
            f"验证以下答案是否准确：{answer}"
        )
        if validation.score < 0.7:
            return {"answer": "需要更多研究", "retry": True}
        return {"answer": answer, "retry": False}

from deepagents import Agent, LLM, Tool, Router

# 定义客服工具集
@Tool.define
def query_order(order_id: str) -> dict:
    """查询订单状态"""
    return {"id": order_id, "status": "配送中", "eta": "明天"}

@Tool.define
def process_refund(order_id: str, reason: str) -> dict:
    """处理退款请求"""
    return {"refund_id": "R001", "amount": 299, "status": "处理中"}

@Tool.define
def recommend_products(category: str, budget: int) -> list:
    """推荐产品"""
    return [{"name": "产品A", "price": 199}, {"name": "产品B", "price": 259}]

@Tool.define
def escalate_to_human(issue: str) -> str:
    """转接人工客服"""
    return f"已转接人工客服，问题描述：{issue}"

# 创建路由：根据意图分发到不同代理
router = Router(
    intent_classifier=LLM(model="gpt-4"),
    routes={
        "order_query": Agent(
            name="订单查询专家",
            tools=[query_order]
        ),
        "refund": Agent(
            name="退款处理专家",
            tools=[process_refund, query_order]
        ),
        "recommendation": Agent(
            name="产品推荐专家",
            tools=[recommend_products]
        ),
    },
    fallback=Agent(
        name="通用助手",
        tools=[escalate_to_human]
    )
)

# 运行客服系统
response = router.handle("我昨天买的耳机想退货")
# 路由到退款处理专家
print(response)  # "好的，我来帮您处理退款..."

from deepagents import Agent, LLM, Tool
import pandas as pd

# 定义数据分析工具
@Tool.define
def execute_sql(query: str) -> str:
    """执行 SQL 查询并返回结果"""
    result = db.execute(query)
    return pd.DataFrame(result).to_markdown()

@Tool.define
def create_chart(data: str, chart_type: str) -> str:
    """根据数据生成图表"""
    df = pd.read_csv(data)
    fig = df.plot(kind=chart_type).get_figure()
    fig.savefig("chart.png")
    return "图表已保存到 chart.png"

@Tool.define
def explain_data(summary_stats: str) -> str:
    """用自然语言解释数据统计结果"""
    return f"数据分析结果：{summary_stats}"

# 创建数据分析代理
analyst = Agent(
    name="data_analyst",
    llm=LLM(model="gpt-4"),
    tools=[execute_sql, create_chart, explain_data],
    system_prompt="""你是一个数据分析专家。
    收到分析请求后：
    1. 先理解用户想知道什么
    2. 编写 SQL 查询数据
    3. 选择合适的图表类型
    4. 用简洁的语言解释结果"""
)

# 执行分析
result = analyst.run(
    "分析上个月各产品类别的销售额，找出增长最快的类别"
)
print(result)
# 代理自动：编写SQL → 执行查询 → 生成柱状图 → 用自然语言解释趋势

from deepagents import Agent, LLM, Memory

# 短期记忆：对话历史（默认）
agent_with_short_memory = Agent(
    name="assistant",
    llm=LLM(model="gpt-4"),
    memory=Memory(type="conversation", window=10),
    # 只保留最近 10 轮对话
)

# 长期记忆：向量数据库
agent_with_long_memory = Agent(
    name="assistant",
    llm=LLM(model="gpt-4"),
    memory=Memory(
        type="vector_store",
        provider="chromadb",
        collection="user_preferences",
        # 自动提取和检索重要信息
    ),
)

# 组合使用：短期 + 长期
agent = Agent(
    name="personal_assistant",
    llm=LLM(model="gpt-4"),
    memory=[
        Memory(type="conversation", window=10),
        Memory(type="vector_store", provider="chromadb"),
        Memory(type="summary", frequency=5),
        # 每 5 轮对话自动生成摘要
    ],
)

# 测试记忆效果
agent.chat("我叫张三，是一名软件工程师")
agent.chat("我喜欢用 Python 做数据分析")
agent.chat("我正在学习 LangChain")

# 50 轮对话后...
response = agent.chat("根据我的背景，推荐一些学习资源")
# 代理记住了：张三 + Python + 数据分析 + LangChain
# 会推荐：LangChain 实战 + Python AI 开发相关资源

from deepagents import Agent, LLM, Tool
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings

# 第一步：构建知识库
documents = load_documents("./knowledge_base/")
chunks = text_splitter.split_documents(
    documents, chunk_size=500, chunk_overlap=50
)
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=OpenAIEmbeddings(),
    persist_directory="./chroma_db"
)

# 第二步：创建 RAG 检索工具
@Tool.define
def search_knowledge_base(query: str) -> str:
    """在知识库中搜索与查询相关的文档"""
    results = vectorstore.similarity_search(query, k=3)
    return "\n\n".join([
        f"[来源{i+1}] {doc.page_content}"
        for i, doc in enumerate(results)
    ])

# 第三步：创建 RAG 代理
rag_agent = Agent(
    name="knowledge_assistant",
    llm=LLM(model="gpt-4"),
    tools=[search_knowledge_base],
    system_prompt="""你是知识库助手。回答时先搜索相关内容，
    基于搜索结果给出答案。始终标注信息来源。"""
)

response = rag_agent.run("公司年假政策是什么？")

# 层级编排：主编排器 + 专业子代理
from deepagents import Agent, LLM, Orchestrator

# 创建主编排器
orchestrator = Orchestrator(
    name="project_manager",
    llm=LLM(model="gpt-4"),
    system_prompt="""你是项目管理代理。
    收到任务后：
    1. 分析任务需求
    2. 分解为子任务
    3. 分配给合适的专业代理
    4. 综合结果返回最终答案"""
)

# 注册专业子代理
orchestrator.register_agent(
    "researcher",
    Agent(name="研究员", tools=["search", "scraper", "arxiv"])
)
orchestrator.register_agent(
    "analyst",
    Agent(name="分析师", tools=["calculator", "statistics"])
)
orchestrator.register_agent(
    "writer",
    Agent(name="撰稿人", tools=["editor", "formatter"])
)

# 运行层级编排
result = orchestrator.run(
    "撰写一份关于 2024 年 AI Agent 技术趋势的深度报告"
)
# 编排器自动：
# 1. 分解为：文献调研 → 数据分析 → 报告撰写
# 2. 分配给：researcher → analyst → writer
# 3. 综合输出最终报告

# 辩论模式：两个代理就同一问题展开辩论
from deepagents import DebateAgent, Judge

pro_agent = DebateAgent(
    name="支持方",
    stance="支持在医疗领域大规模使用 AI",
    llm=LLM(model="gpt-4"),
    tools=["search"]
)

con_agent = DebateAgent(
    name="反对方",
    stance="对医疗 AI 保持谨慎态度",
    llm=LLM(model="gpt-4"),
    tools=["search"]
)

judge = Judge(
    name="裁判",
    llm=LLM(model="gpt-4"),
    max_rounds=3  # 最多辩论 3 轮
)

# 开始辩论
result = judge.run(
    topic="医疗 AI 是否应该获得自主诊断权限？",
    agents=[pro_agent, con_agent]
)

# result 包含：
# - 每轮辩论的观点和论据
# - 最终综合结论（综合双方观点）
# - 推荐的行动方案（带条件）
print(result.consensus)
# "在严格监管框架下，允许 AI 辅助诊断，
#  但最终决策权必须由持证医生保留..."

# 生产级错误处理框架
from deepagents import Agent, RetryPolicy, CircuitBreaker

# 配置重试策略
retry_policy = RetryPolicy(
    max_attempts=3,
    backoff="exponential",  # 指数退避：1s, 2s, 4s
    retry_on=[LLMRateLimitError, LLMAPIError],
    fallback_model="gpt-3.5-turbo"  # 主模型失败时降级
)

# 配置熔断器
circuit_breaker = CircuitBreaker(
    failure_threshold=5,   # 连续 5 次失败后熔断
    recovery_timeout=60,   # 60 秒后尝试恢复
    half_open_requests=3   # 半开状态允许 3 次测试请求
)

# 创建带完整错误处理的代理
production_agent = Agent(
    name="production_assistant",
    llm=LLM(model="gpt-4"),
    retry_policy=retry_policy,
    circuit_breaker=circuit_breaker,
    timeout=30,              # 单次请求超时 30 秒
    max_iterations=10,       # 最多 10 个 ReAct 循环
    graceful_degradation=True # 优雅降级而非抛出异常
)

# 监控和告警
@production_agent.on_error
def handle_error(error, context):
    logger.error(f"代理错误: {error}", extra=context)
    metrics.increment("agent.errors")
    if isinstance(error, CircuitBreakerOpenError):
        alert.send("代理熔断器已触发，请检查服务状态")

# 执行任务
try:
    result = production_agent.run("分析季度销售数据")
except CircuitBreakerOpenError:
    result = "服务暂时不可用，已切换到备用方案"
    fallback_result = simple_rule_based_analysis(data)

# === 完整项目：AI 研究助手 ===
from deepagents import (
    Agent, LLM, Tool, Pipeline, Router,
    Memory, Orchestrator
)

# 1. 定义工具集
@Tool.define
def search_papers(query: str, max_results: int = 10) -> list:
    """搜索学术论文"""
    return arxiv_client.search(query, max_results)

@Tool.define
def summarize_paper(paper_url: str) -> dict:
    """总结论文核心内容"""
    paper = fetch_paper(paper_url)
    return {
        "title": paper.title,
        "summary": paper.abstract,
        "key_findings": extract_findings(paper),
        "methodology": extract_methodology(paper)
    }

@Tool.define
def compare_papers(paper_ids: list) -> str:
    """对比多篇论文的方法和结论"""
    papers = [fetch_paper(pid) for pid in paper_ids]
    return generate_comparison_table(papers)

# 2. 创建专业代理
researcher = Agent(
    name="文献检索员",
    llm=LLM(model="gpt-4"),
    tools=[search_papers],
    system_prompt="你是学术文献检索专家，擅长找到最相关的论文"
)

analyzer = Agent(
    name="论文分析师",
    llm=LLM(model="gpt-4"),
    tools=[summarize_paper, compare_papers],
    system_prompt="你是学术论文分析专家，擅长提炼核心观点和方法论"
)

writer = Agent(
    name="报告撰写员",
    llm=LLM(model="gpt-4"),
    tools=[],
    system_prompt="你是学术写作专家，擅长将分析结果组织成结构清晰的报告"
)

# 3. 编排工作流
pipeline = Pipeline([
    researcher,   # 第一步：检索相关论文
    analyzer,     # 第二步：分析和对比论文
    writer        # 第三步：撰写研究报告
])

# 4. 执行研究任务
result = pipeline.run(
    "调研 2024 年 RAG（检索增强生成）技术的最新进展和趋势"
)

# 5. 输出结果
print(f"检索论文数: {result.papers_found}")
print(f"分析论文数: {result.papers_analyzed}")
print(f"报告字数: {len(result.report)}")
print(result.report)

# 进阶：带质量保证的研究系统
from deepagents import (
    Orchestrator, DebateAgent, Judge,
    Agent, LLM, QualityGate
)

# 质量门：确保每个阶段输出达标
quality_gate = QualityGate(
    criteria={
        "completeness": 0.8,  # 完整性 ≥ 80%
        "accuracy": 0.85,     # 准确性 ≥ 85%
        "relevance": 0.9      # 相关性 ≥ 90%
    },
    on_fail="retry_with_feedback",  # 不达标则重试
    max_retries=3
)

# 创建编排器
orchestrator = Orchestrator(
    name="研究总监",
    llm=LLM(model="gpt-4"),
    quality_gate=quality_gate
)

# 注册辩论模式代理组
orchestrator.register_debate_group(
    topic="技术路线选择",
    agents=[
        DebateAgent(name="保守派", stance="优先稳定性"),
        DebateAgent(name="创新派", stance="优先前沿性"),
    ],
    judge=Judge(max_rounds=2)
)

# 运行带质量保证的研究
result = orchestrator.run(
    task="评估 Transformer 架构 vs Mamba 架构在长文本处理中的优劣",
    quality_threshold=0.85
)

# 如果质量不达标，系统会自动：
# 1. 分析质量缺陷
# 2. 补充搜索缺失信息
# 3. 重新生成不达标部分
# 4. 最多重试 3 次

# 生产部署配置示例
from deepagents import Agent, LLM, DeploymentConfig

# 1. 多模型路由：按任务复杂度选择模型
model_router = ModelRouter({
    "simple": LLM(model="gpt-3.5-turbo", cost_per_1k=0.5),
    "medium": LLM(model="gpt-4", cost_per_1k=30),
    "complex": LLM(model="gpt-4-turbo", cost_per_1k=60),
})

# 2. 速率限制和成本控制
agent = Agent(
    name="production_agent",
    llm=model_router,
    rate_limit={
        "requests_per_minute": 60,
        "tokens_per_day": 1000000
    },
    cost_limit={
        "daily_budget": 50.0,  # 每日预算上限
        "alert_threshold": 0.8  # 80% 时告警
    }
)

# 3. 优雅降级配置
agent.add_fallback(
    condition=RateLimitExceeded,
    fallback_llm=LLM(model="claude-3-haiku"),
    max_fallbacks=3
)

# 4. 监控和追踪
agent.enable_tracing(
    provider="langsmith",
    project="production-agent",
    tags=["v2.0", "customer-facing"]
)

# 5. 健康检查
@app.health_check
def agent_health():
    return {
        "status": "healthy",
        "model": agent.current_model,
        "requests_today": agent.metrics.requests,
        "cost_today": agent.metrics.cost,
        "avg_latency": agent.metrics.avg_latency
    }