长任务恢复与 Deep Research 续跑

REPORT_STAGES = (
    ReportStage(key="plan_research",   title="规划研究问题",   receipt_key="deepresearch:plan:v1",          tool_name="llm.plan",         ...),
    ReportStage(key="search_web",      title="检索公开网页",   receipt_key="deepresearch:web_search:v1",    tool_name="web.search",       ...),
    ReportStage(key="fetch_sources",   title="抓取来源正文",   receipt_key="deepresearch:web_fetch:v1",    tool_name="web.fetch",         ...),
    ReportStage(key="screen_evidence", title="筛选证据并去重", receipt_key="deepresearch:evidence_screen:v1", tool_name="evidence.screen", ...),
    ReportStage(key="analyze_findings",title="交叉分析发现",   receipt_key="deepresearch:analysis:v1",     tool_name="llm.analyze",      ...),
    ReportStage(key="critic_review",    title="批判性质检",     receipt_key="deepresearch:critic:v1",        tool_name="llm.critic",       ...),
    ReportStage(key="write_report",    title="生成研究报告",   receipt_key="deepresearch:report:v1",       tool_name="workspace.write",  ...),
)

def _compile_graph(checkpointer: Any) -> Any:
    graph = StateGraph(ReportState)
    for stage in REPORT_STAGES:
        graph.add_node(stage.key, _stage_node(stage))
    graph.add_node("finalize_report", finalize_report)
    graph.add_node("reflection", _run_reflection_node)
    graph.add_edge(START, "plan_research")
    graph.add_edge("plan_research", "search_web")
    graph.add_edge("search_web", "fetch_sources")
    graph.add_edge("fetch_sources", "screen_evidence")
    graph.add_edge("screen_evidence", "reflection")
    graph.add_conditional_edges(
        "reflection",
        _reflection_router,
        {"search_web": "search_web", "analyze_findings": "analyze_findings"},
    )
    graph.add_conditional_edges(
        "analyze_findings",
        lambda state: "critic_review" if state.get("requires_critic_review", True) else "write_report",
        {"critic_review": "critic_review", "write_report": "write_report"},
    )
    graph.add_edge("critic_review", "write_report")
    graph.add_edge("write_report", "finalize_report")
    graph.add_edge("finalize_report", END)
    return graph.compile(checkpointer=checkpointer)

def _reflection_router(state: ReportState) -> str:
    loop_count = int(state.get("research_loop_count") or 0)
    cfg = ResearchConfig.from_state(state)
    if state.get("is_sufficient") or loop_count >= cfg.max_research_loops:
        return "analyze_findings"
    return "search_web"

def _build_web_search_subgraph():
    def fan_out_queries(state: WebSearchState):
        queries = state.get("queries") or [state.get("query", "")]
        return [Send("query_agent", {"query": state.get("query", ""), "queries": [query]}) for query in queries]

    async def query_agent(state: WebSearchState) -> WebSearchState:
        query = (state.get("queries") or [state.get("query", "")])[0]
        max_results = int(state.get("max_results") or ResearchConfig.from_env().search_results_per_query)
        results = await _web_search(query, max_results=max_results)
        return {"search_packets": [{"agent": "web_search_agent", "query": query, "results": results}]}

    def reduce_queries(state: WebSearchState) -> WebSearchState:
        packets = state.get("search_packets") or []
        digest = "；".join(f"{packet['query']}={len(packet.get('results') or [])} 条" for packet in packets)
        return {"search_digest": digest}

    graph = StateGraph(WebSearchState)
    graph.add_node("query_agent", query_agent)
    graph.add_node("reduce_queries", reduce_queries)
    graph.add_conditional_edges(START, fan_out_queries)
    graph.add_edge("query_agent", "reduce_queries")
    graph.add_edge("reduce_queries", END)
    return graph.compile(name="web_search_subgraph")

def fan_out_reviewers(state: AnalysisState):
    reviewers = state.get("reviewers") or ["fact_reviewer", "implication_reviewer", "risk_reviewer"]
    return [
        Send("analysis_reviewer", {
            "query": state.get("query", ""),
            "evidence_table": state.get("evidence_table", []),
            "reviewers": [reviewer],
        })
        for reviewer in reviewers
    ]

async for update in graph.astream(
    None,
    config=config,
    context=self.build_native_context(payload.get("platform_context")),
    stream_mode="updates",
):
    stage = self._stage_from_update(update)
    if stage is None:
        continue
    # 只处理 checkpoint 之后的节点 update,checkpoint 前的节点不会重跑
    ...

@staticmethod
def _checkpoint_config_from(config: dict[str, Any]) -> dict[str, Any]:
    configurable = dict((config or {}).get("configurable") or {})
    thread_id = str(configurable.get("thread_id") or "").strip()
    checkpoint_id = str(configurable.get("checkpoint_id") or "").strip()
    if not thread_id:
        return dict(config or {})
    return {
        "configurable": {
            "thread_id": thread_id,
            "checkpoint_ns": str(configurable.get("checkpoint_ns") or ""),
            "checkpoint_id": checkpoint_id,
        }
    }

for stage in REPORT_STAGES:
    yield {"type": "tool_call", "tool_name": stage.tool_name, ...}
    await asyncio.sleep(_stage_delay_seconds())

    if _should_fail_stage(stage, "start"):
        raise RuntimeError(f"simulated_tool_failure:{stage.key}")

    values, framework_ref, checkpoint_id, graph_config = await self._invoke_graph_step(
        graph, graph_input=graph_input, config=graph_config, stage=stage, payload=payload,
    )
    graph_input = None
    artifact_path = _artifact_path_for_state(values, stage)

    for runtime_event in self._stage_runtime_events(values, stage=stage, run_id=run_id):
        yield runtime_event

    # checkpoint 先于 tool_result:平台建 tool_receipt 时会从 session 查
    # latest checkpoint with matching run_id,必须先写 checkpoint,receipt 才能关联到正确 checkpoint_id。
    yield {"type": "checkpoint", "metadata": _checkpoint_metadata_for_ref(stage=stage, run_id=run_id, framework_ref=framework_ref, artifact_path=artifact_path)}
    yield {"type": "tool_result", "tool_name": stage.tool_name, "tool_output": {"ok": True, "receipt_key": stage.receipt_key, "artifact_path": artifact_path, "langgraph_checkpoint_id": checkpoint_id}, ...}

from ksadk.toolsets.workspace import write_workspace_file, write_workspace_files

def _write_workspace_report_files(state, report_markdown, report_html):
    base_dir = _workspace_artifact_dir(state)
    files = [
        {"path": f"{base_dir}/deepresearch-report.md", "content": report_markdown},
        {"path": f"{base_dir}/deepresearch-report.html", "content": report_html},
        {"path": f"{base_dir}/sources.json", "content": _build_sources_json(state)},
        {"path": f"{base_dir}/evidence.json", "content": _build_evidence_json(state, report_markdown)},
        {"path": f"{base_dir}/research-state.json", "content": _build_research_state_json(state, report_markdown, report_html)},
    ]
    result = write_workspace_files(files, overwrite=True)
    if not isinstance(result, dict) or not result.get("ok"):
        raise RuntimeError(f"write_workspace_files failed for {base_dir}: {result}")
    written = [str(item.get("path", "")) for item in result.get("written", []) if item.get("path")]
    return written, result

from ksadk.toolsets.workspace import write_workspace_file, write_workspace_files
from ksadk.toolsets.web import web_fetch as _ksadk_web_fetch
from ksadk.toolsets.web import web_search as _ksadk_web_search

async def _web_search(query: str, *, max_results: int = 5) -> list[dict[str, Any]]:
    # 优先 ksadk 内置 web_search(ksyun provider),失败再走 metaso → configured_api → 公开搜索降级
    ksadk_results = await _search_with_ksadk_web_search(query, max_results=max_results)
    if ksadk_results:
        return ksadk_results
    ...

name: long-task-resume
framework: langgraph
entry_point: agent.py
agent_variable: root_agent
runner_class: agent.LongTaskE2ERunner
ui_profile: custom
ui_path: /research
ui_bundle_path: research-ui/dist
resources:
  cpu: "1"
  memory: "2Gi"