from deepagents import create_deep_agent


supervisor = create_deep_agent(
    model="openai:gpt-4.1",
    system_prompt="""
你是 AI 行业周报的主管理者。
先拆分任务，再把可独立完成的工作委派给 subagent；
只合并有证据的结论，最终输出 Markdown 周报。
""",
)

result = supervisor.invoke(
    {
        "messages": [
            {
                "role": "user",
                "content": "整理本周三个重要 AI 事件，并说明各自影响。",
            }
        ]
    }
)

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

from typing import Any

from deepagents import SubAgent, create_deep_agent
from langchain_core.tools import tool


DEMO_EVENTS = {
    "agent": [
        {
            "title": "Agent 平台发布异步任务能力",
            "source": "https://example.com/agent-platform",
            "summary": "长任务可在后台执行，并由主管理者查询状态。",
        }
    ],
    "model": [
        {
            "title": "新模型强调工具调用可靠性",
            "source": "https://example.com/model-release",
            "summary": "复杂工具链的成功率得到改善。",
        }
    ],
}


@tool
def search_news(topic: str) -> list[dict[str, str]]:
    """搜索演示新闻；生产环境应替换为真实搜索服务。"""
    return DEMO_EVENTS.get(topic.lower(), [])


@tool
def calculate_change(current: float, previous: float) -> dict[str, Any]:
    """计算数值及百分比变化。"""
    if previous == 0:
        return {"absolute": current, "percent": None, "note": "基期为 0"}
    return {
        "absolute": round(current - previous, 4),
        "percent": round((current - previous) / previous * 100, 2),
    }


researcher: SubAgent = {
    "name": "researcher",
    "description": "检索 AI 新闻并返回标题、摘要和来源；需要搜集事实时使用。",
    "system_prompt": """
你是研究员。每条结论必须附来源 URL；区分来源原文和你的推断。
如果没有结果，明确返回“未找到”，不得补写看似合理的新闻。
""",
    "tools": [search_news],
}

analyst: SubAgent = {
    "name": "data-analyst",
    "description": "计算变化率并解释指标；出现数字比较时使用。",
    "system_prompt": """
你是数据分析师。必须调用计算工具，不要心算；
返回输入、公式、结果和数据限制，不得把相关性写成因果关系。
""",
    "tools": [calculate_change],
}

fact_checker: SubAgent = {
    "name": "fact-checker",
    "description": "审核已有草稿中的事实、数字和来源，不承担初次写作。",
    "system_prompt": """
你是事实核查员。逐条输出 PASS、WARN 或 FAIL，并解释依据。
没有一手来源的关键断言不能标记为 PASS。
""",
    "tools": [search_news],
}

supervisor = create_deep_agent(
    model="openai:gpt-4.1",
    system_prompt="""
你是周报主管理者。先列出交付标准，再按需委派研究、分析和核查。
不要自己伪装成专家；专家结果冲突时，优先补充核查并保留不确定性。
最终输出：摘要、事件、数据解读、风险提示、来源。
""",
    subagents=[researcher, analyst, fact_checker],
)

response = supervisor.invoke(
    {
        "messages": [
            {
                "role": "user",
                "content": (
                    "以 agent 和 model 为关键词制作演示周报；"
                    "并计算某指标从 125 增长到 160 的变化率，最后核查关键事实。"
                ),
            }
        ]
    }
)

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

from deepagents import CompiledSubAgent, create_deep_agent
from langchain.agents import create_agent
from langchain_core.tools import tool


@tool
def style_rules(section: str) -> str:
    """返回指定栏目采用的编辑规则。"""
    rules = {
        "summary": "不超过 120 字，先结论后背景。",
        "event": "标题、事实、影响三段式；事实必须保留来源。",
        "risk": "明确区分已知事实、推断和未知项。",
    }
    return rules.get(section, "使用简洁、可核查的中文。")


# create_agent 返回可调用的已编译图；也可换成自己用 StateGraph 编译的图。
editor_graph = create_agent(
    model="openai:gpt-4.1-mini",
    tools=[style_rules],
    system_prompt="""
你是终稿编辑。只重组和压缩已有材料，不新增事实；
发现来源缺失时插入 [待核查]，不要自行补齐。
""",
)

editor = CompiledSubAgent(
    name="editor",
    description="把已核查的材料整理成最终 Markdown；只在事实核查后调用。",
    runnable=editor_graph,
)

supervisor = create_deep_agent(
    model="openai:gpt-4.1",
    system_prompt="先研究和核查，最后且仅最后调用 editor 生成终稿。",
    subagents=[editor],
)

result = supervisor.invoke(
    {
        "messages": [
            {
                "role": "user",
                "content": "把以下已核查材料编辑成周报：事件 A 已确认；影响仍属推断。",
            }
        ]
    }
)
print(result["messages"][-1].content)

from deepagents import create_deep_agent

# researcher、analyst、fact_checker、editor 来自前面的定义。
supervisor = create_deep_agent(
    model="openai:gpt-4.1",
    subagents=[researcher, analyst, fact_checker, editor],
    system_prompt="""
你是主管理者，请按以下状态机编排：

1. DISCOVER：凡是外部事实，交给 researcher；多个主题可独立委派。
2. ANALYZE：凡是数值比较，交给 data-analyst，并传入原始数值。
3. VERIFY：整理“断言—来源—证据”清单，交给 fact-checker。
4. REPAIR：存在 FAIL 时回到 DISCOVER；存在 WARN 时在终稿保留限定语。
5. PUBLISH：只有关键断言均无 FAIL，才把材料交给 editor。

每次委派只给最小必要上下文。最多执行两轮 REPAIR；两轮后仍失败，
停止扩张任务并向用户报告缺口。不得调用未注册的角色。
""",
)

from deepagents import SubAgent, create_deep_agent


def build_supervisor(*, can_read_internal: bool, language: str):
    agents: list[SubAgent | CompiledSubAgent] = [
        researcher,
        analyst,
        fact_checker,
        editor,
    ]

    if can_read_internal:
        internal_expert: SubAgent = {
            "name": "internal-knowledge-expert",
            "description": "查询已授权的内部材料；仅处理内部事实问题。",
            "system_prompt": "只返回当前调用者有权访问的内容，并保留文档标识。",
            "tools": [],  # 在真实应用中放入经过权限过滤的检索工具
        }
        agents.append(internal_expert)

    return create_deep_agent(
        model="openai:gpt-4.1",
        subagents=agents,
        system_prompt=f"用 {language} 输出；只能调用已注册专家。",
    )


supervisor = build_supervisor(can_read_internal=False, language="简体中文")

import asyncio
import json
import uuid
from dataclasses import asdict, dataclass, field
from datetime import datetime, timezone
from typing import Literal

from deepagents import create_deep_agent
from langchain_core.tools import tool


Status = Literal[
    "queued", "running", "succeeded", "failed", "cancelling", "cancelled"
]


def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()


@dataclass
class Job:
    id: str
    expert: str
    instruction: str
    status: Status = "queued"
    progress: int = 0
    result: str | None = None
    error: str | None = None
    updates: list[str] = field(default_factory=list)
    created_at: str = field(default_factory=utc_now)
    updated_at: str = field(default_factory=utc_now)


class AsyncExpertRegistry:
    """单进程演示注册表；一个应用实例创建一个 registry。"""

    def __init__(self, max_concurrency: int = 3) -> None:
        self.jobs: dict[str, Job] = {}
        self.tasks: dict[str, asyncio.Task[None]] = {}
        self._lock = asyncio.Lock()
        self._semaphore = asyncio.Semaphore(max_concurrency)

    async def start(self, expert: str, instruction: str) -> dict:
        if expert not in {"researcher", "analyst", "fact-checker"}:
            raise ValueError(f"unknown expert: {expert}")

        job = Job(id=uuid.uuid4().hex, expert=expert, instruction=instruction)
        async with self._lock:
            self.jobs[job.id] = job
            self.tasks[job.id] = asyncio.create_task(self._run(job.id))
        return {"task_id": job.id, "status": job.status}

    async def _run(self, task_id: str) -> None:
        job = self.jobs[task_id]
        try:
            async with self._semaphore:
                job.status = "running"
                job.updated_at = utc_now()

                # 模拟分阶段长任务；每一阶段都是取消点。
                for progress in (20, 50, 80):
                    await asyncio.sleep(0.2)
                    job.progress = progress
                    job.updated_at = utc_now()

                # 演示如何吸收 update；真实 worker 可将更新写入下一次模型调用。
                update_text = "；补充要求：" + "；".join(job.updates) if job.updates else ""
                await asyncio.sleep(0.2)
                job.result = (
                    f"{job.expert} 已完成：{job.instruction}{update_text}。"
                    "这是演示结果；生产环境应在此调用真正的专家 agent。"
                )
                job.progress = 100
                job.status = "succeeded"
                job.updated_at = utc_now()
        except asyncio.CancelledError:
            job.status = "cancelled"
            job.updated_at = utc_now()
            raise
        except Exception as exc:
            job.status = "failed"
            job.error = f"{type(exc).__name__}: {exc}"
            job.updated_at = utc_now()

    async def check(self, task_id: str) -> dict:
        job = self._get(task_id)
        return asdict(job)

    async def update(self, task_id: str, instruction: str) -> dict:
        job = self._get(task_id)
        if job.status not in {"queued", "running"}:
            return {"task_id": task_id, "accepted": False, "status": job.status}
        job.updates.append(instruction)
        job.updated_at = utc_now()
        return {"task_id": task_id, "accepted": True, "status": job.status}

    async def cancel(self, task_id: str) -> dict:
        job = self._get(task_id)
        task = self.tasks.get(task_id)
        if job.status not in {"queued", "running"} or task is None:
            return {"task_id": task_id, "cancelled": False, "status": job.status}

        job.status = "cancelling"
        job.updated_at = utc_now()
        task.cancel()
        try:
            await task
        except asyncio.CancelledError:
            pass
        return {"task_id": task_id, "cancelled": True, "status": job.status}

    async def list(self, status: Status | None = None) -> list[dict]:
        jobs = self.jobs.values()
        if status is not None:
            jobs = (job for job in jobs if job.status == status)
        return [
            {
                "task_id": job.id,
                "expert": job.expert,
                "status": job.status,
                "progress": job.progress,
                "updated_at": job.updated_at,
            }
            for job in jobs
        ]

    def _get(self, task_id: str) -> Job:
        try:
            return self.jobs[task_id]
        except KeyError as exc:
            raise ValueError(f"unknown task_id: {task_id}") from exc


registry = AsyncExpertRegistry(max_concurrency=3)


@tool("start")
async def start_tool(
    expert: Literal["researcher", "analyst", "fact-checker"],
    instruction: str,
) -> str:
    """启动后台专家任务。返回 task_id；不要把它当作任务结果。"""
    return json.dumps(await registry.start(expert, instruction), ensure_ascii=False)


@tool("check")
async def check_tool(task_id: str) -> str:
    """查询一个后台任务的状态、进度；成功时同时返回结果。"""
    return json.dumps(await registry.check(task_id), ensure_ascii=False)


@tool("update")
async def update_tool(task_id: str, instruction: str) -> str:
    """向 queued/running 任务补充要求；已结束的任务不会被修改。"""
    return json.dumps(
        await registry.update(task_id, instruction), ensure_ascii=False
    )


@tool("cancel")
async def cancel_tool(task_id: str) -> str:
    """请求取消 queued/running 任务。重复取消是安全的。"""
    return json.dumps(await registry.cancel(task_id), ensure_ascii=False)


@tool("list")
async def list_tool(status: str = "") -> str:
    """列出后台任务；status 为空表示全部。"""
    normalized = status or None
    return json.dumps(
        await registry.list(normalized),  # type: ignore[arg-type]
        ensure_ascii=False,
    )


async_supervisor = create_deep_agent(
    model="openai:gpt-4.1",
    tools=[start_tool, check_tool, update_tool, cancel_tool, list_tool],
    system_prompt="""
你是异步周报主管理者。

- 研究、分析、核查超过一个步骤时，用 start 启动后台专家。
- 独立任务可连续 start；保存每个 task_id 及其用途。
- start 只表示已受理，绝不把 task_id 当结果。
- 需要依赖结果时才 check；不要无间隔地高频轮询。
- 用户补充要求时，对仍在运行的对应任务调用 update。
- 任务失去价值、超时或用户撤回时调用 cancel。
- 不确定有哪些任务时调用 list。
- 只有 status=succeeded 才读取 result；failed/cancelled 不得编造结果。
""",
)


async def main() -> None:
    response = await async_supervisor.ainvoke(
        {
            "messages": [
                {
                    "role": "user",
                    "content": (
                        "启动研究员收集 agent 新闻，同时让分析师比较 125 和 160；"
                        "告诉我任务句柄和当前状态，不必等待全部完成。"
                    ),
                }
            ]
        }
    )
    print(response["messages"][-1].content)

    # 演示进程不能在后台任务完成前退出。
    await asyncio.sleep(1.0)
    print(json.dumps(await registry.list(), ensure_ascii=False, indent=2))


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

# 应用启动时创建，避免每个任务重复编译。
research_agent = create_deep_agent(
    model="openai:gpt-4.1-mini",
    tools=[search_news],
    system_prompt="你是研究员；每条结论必须附来源。",
)

expert_agents = {"researcher": research_agent}


async def invoke_real_expert(job: Job) -> str:
    agent = expert_agents[job.expert]
    extra = "\n补充要求：" + "；".join(job.updates) if job.updates else ""
    result = await agent.ainvoke(
        {
            "messages": [
                {"role": "user", "content": job.instruction + extra}
            ]
        }
    )
    return result["messages"][-1].content