Agent-10-10.长任务与外部计算调度

发表于 | 分类于
Agent 接入分钟/小时级外部任务的模式：异步 submit、轮询、回调与方案对比。

系列：00 索引 · 上一篇：09 LangGraph · 下一篇：11 引用报告

1. 行业常见问题

场景 为何不能同步 tool
GPU 训练/推理 分钟~小时
批量 BLAST/折叠 队列排队
工作流引擎（Airflow/Argo/K8s Job） 异步 Pod
MCP/HTTP 超时 通常 ≤ 几分钟

若 Agent 进程内 sleep 等待，会 占连接、占 worker、无法扩展。

2. 该技术如何解决

薄 Agent + 厚 Worker：

  1. Tool/MCP：submit(job) → 返回 job_id
  2. Tool/MCP：get_status(job_id) / get_logs
  3. Agent 或编排层：轮询 / 事件驱动 / 用户 HITL 后再查
  4. 结果落 对象存储或共享路径，Agent 只传 URI

3. 核心原理

3.1 三 Tool 兼容模式（无原生 Task 协议时）

Tool 作用
submit_* 创建任务
get_*_status 查询阶段
get_*_result 取结果 URI 或摘要

3.2 MCP 长任务注意

  • Host 可能不支持超长单次 tools/call
  • 应 快速返回 job 标识，观测另调
  • 默认值链：上层勿用 None 覆盖下层合法默认（如任务名前缀）

3.3 与 HITL 结合

submit 前确认 参数与预估成本；失败时保留 workflow id 供排错。

4. 典型实现与代码示例

长任务由 mcp/mcp_argo_sirna（FastMCP）提供 submit/status/wait 三件套，agent/analyst/artifacts.py 在 LangGraph analyst 节点按 plan 顺序调用，并与 HITL（Agent-10-08）联动。

4.1 MCP Server：薄接口、快返回

1
2
3
4
5
6
7
8
9
10
11
12
# mcp/mcp_argo_sirna/server.py
@mcp.tool()
def submit_template_tool(template_name, generate_name="", parameters_json="{}") -> dict:
    return submit_template(template_name, generate_name=generate_name, parameters=params)

@mcp.tool()
def get_workflow_status_tool(workflow_name: str) -> dict:
    return get_workflow_status(workflow_name)

@mcp.tool()
def wait_workflow_tool(workflow_name: str, timeout_seconds: int = 3600) -> dict:
    return wait_workflow(workflow_name, timeout_seconds=timeout_seconds)

默认 ARGO_MCP_MODE=mock 可离线验收；live 模式对接真实 Argo Workflow。

4.2 Analyst：plan 驱动 + HITL 闸门

1
2
3
4
5
6
7
8
9
10
11
12
13
14
# agent/analyst/artifacts.py — 读取 research_plan.v1 的 steps[]
def execute_plan_steps(plan, *, mcp_call, require_hitl=True, hitl_approved=False):
    # plan 字段 runtime 作用：
    #   steps[].template  → submit_template_tool(template_name=...)
    #   steps[].params    → parameters_json
    #   steps[].id        → generate_name 前缀、execution_manifest.workflows[].step_id
    #   goal / run_id     → 写入 execution_manifest.v1
    if require_hitl and not hitl_approved:
        return {"status": "blocked_hitl", ...}  # status 阻断下游 reporter 当作成功

    # execution_manifest.v1 输出字段：
    #   status            — completed / partial / blocked_hitl
    #   workflows[]       — 每步 workflow_name、phase、outputs（报告引用计算结果 URI）
    ...

LangGraph 在 interrupt_before=["analyst"] 处暂停，用户 approve 后 Command(resume=True) 续跑。

4.3 工程验收

1
2
3
4
5
6
7
8
9
# 仅 MCP mock 模式（stdio）
uv run python mcp/mcp_argo_sirna/server.py

# HITL 中断 → 批准 → Analyst 提交（mock Argo）
uv run python -m agent.orchestrator.cli start-hitl --symbol BRCA1
uv run python -m agent.orchestrator.cli resume <run_id> --approve

# 单次跑通 full 流水线（HITL 已批准）
uv run python -m agent.orchestrator.cli run --symbol BRCA1 --full --hitl --approve --mock-llm

5. 替代方案与优缺点

方案 优点 缺点
三 Tool 轮询 简单、Host 兼容广 延迟、需退避
MCP Tasks（原生，若 Host 支持） 标准进度/取消 生态仍在演进
Webhook 回调 + 消息队列 实时 公网回调复杂
Temporal Activity 可靠长事务 基础设施
同步阻塞 tool 代码最少 生产不可用

6. 自检题

  1. 为什么 submit 应快速返回而不是 wait 完成？
  2. 轮询间隔如何权衡体验与负载？
  3. 失败任务 Agent 应向用户暴露哪些字段？

7. 延伸阅读

-------------本文结束感谢您的阅读-------------