引言
当 AI Agent 从单次问答走向持续运行的自主系统，一个核心挑战随之而来：如何让 Agent 可靠地执行跨越数分钟甚至数小时的复杂任务？单次 LLM 调用无法支撑多步骤、有依赖、需容错的长任务执行。本文从工程实践角度，系统拆解 AI Agent 长任务编排引擎的核心组件——任务分解、状态机管理、错误恢复与自校正机制，并结合实际代码示例展示可落地的架构设计。
1. 长任务的本质挑战
AI Agent 面临的"长任务"并非简单的多次调用串联，它涉及以下核心问题：
- 状态持久化：任务可能跨越进程重启、网络中断，必须能从断点恢复
- 依赖管理：子任务之间有复杂的 DAG 依赖关系，需要拓扑排序与并行调度
- 部分失败处理：某个子任务失败时，是重试、跳过还是补偿？
- 超时与资源控制：防止死循环和无限制的资源消耗
- 人机协作：长任务中可能需要人类介入审批或提供信息
业界常见的工程范式包括 LangChain 的 LCEL 链式调用、CrewAI 的角色协作、AutoGen 的多 Agent 对话，以及 OpenAI 的 Assistants API 线程模型。但这些框架在工程可靠性方面仍有不足，我们需要的是更接近 Temporal、Airflow、Cadence 这类分布式工作流引擎的保障能力。
2. 任务分解：从用户意图到可执行子任务
2.1 Plan-and-Execute 模式
Plan-and-Execute 是任务分解最经典的架构模式。其核心思想是将"规划"与"执行"解耦——先由 LLM 生成完整的执行计划，再按步骤执行并动态调整。
from dataclasses import dataclass, field
from enum import Enum
from typing import Any
import json, time, uuid
class TaskStatus(Enum):
PENDING = "pending"
RUNNING = "running"
COMPLETED = "completed"
FAILED = "failed"
SKIPPED = "skipped"
@dataclass
class SubTask:
id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
description: str = ""
action: str = "" # tool/function name
parameters: dict = field(default_factory=dict)
dependencies: list = field(default_factory=list) # list of task IDs
status: TaskStatus = TaskStatus.PENDING
result: Any = None
error: str = ""
retry_count: int = 0
max_retries: int = 3
timeout_seconds: int = 60
started_at: float = 0
completed_at: float = 0
@dataclass
class ExecutionPlan:
plan_id: str = field(default_factory=lambda: str(uuid.uuid4())[:12])
goal: str = ""
subtasks: list = field(default_factory=list) # List[SubTask]
context: dict = field(default_factory=dict)
status: TaskStatus = TaskStatus.PENDING2.2 LLM 驱动的任务分解
任务分解的质量直接决定整体执行成功率。好的分解需满足以下原则：DAG 依赖（无环）、粒度适中（单个子任务 30s~5min 完成）、可独立测试、失败可补偿。
class PlanGenerator:
"""基于 LLM 的任务分解器，生成结构化执行计划"""
SYSTEM_PROMPT = """你是一个任务分解专家。将用户目标分解为可执行的子任务列表。
输出 JSON 格式：
{
"subtasks": [
{
"description": "子任务描述",
"action": "工具/函数名",
"parameters": {参数键值对},
"dependencies": ["前置任务ID"],
"timeout_seconds": 60
}
]
}
规则：
1. 每个子任务应独立可验证
2. 明确标注依赖关系
3. 涉及外部 API 调用的任务 timeout_seconds 设为 120
4. 涉及文件操作的任务设为 30
5. 涉及人工审批的设为 300"""
def generate(self, goal: str, tools: list, context: dict = None) -> ExecutionPlan:
tool_descriptions = "\n".join(
f"- {t['name']}: {t['description']}" for t in tools
)
user_prompt = f"目标: {goal}\n\n可用工具:\n{tool_descriptions}"
if context:
user_prompt += f"\n\n上下文信息:\n{json.dumps(context, ensure_ascii=False)}"
response = self._call_llm(self.SYSTEM_PROMPT, user_prompt)
plan_data = self._parse_plan(response)
plan = ExecutionPlan(goal=goal, context=context or {})
for i, task_data in enumerate(plan_data["subtasks"]):
subtask = SubTask(
id=f"task_{i:03d}",
description=task_data["description"],
action=task_data["action"],
parameters=task_data.get("parameters", {}),
dependencies=task_data.get("dependencies", []),
timeout_seconds=task_data.get("timeout_seconds", 60),
)
plan.subtasks.append(subtask)
return plan
def _call_llm(self, system: str, user: str) -> str:
# 实际实现调用 LLM API
pass
def _parse_plan(self, response: str) -> dict:
# 提取并验证 JSON
try:
start = response.index('{')
end = response.rindex('}') + 1
return json.loads(response[start:end])
except (ValueError, json.JSONDecodeError):
raise ValueError(f"无法解析计划: {response[:200]}")3. 状态机驱动的任务编排引擎
长任务执行的核心是一个状态机。每个子任务经历 PENDING → RUNNING → COMPLETED/FAILED 的状态转换，而编排引擎负责驱动整个状态流转。
3.1 持久化状态存储
import sqlite3
import threading
from datetime import datetime
class PlanStore:
"""持久化任务执行计划到 SQLite，支持断点恢复"""
def __init__(self, db_path: str = "agent_plans.db"):
self.db_path = db_path
self._local = threading.local()
self._init_db()
@property
def conn(self):
if not hasattr(self._local, 'conn'):
self._local.conn = sqlite3.connect(self.db_path)
self._local.conn.row_factory = sqlite3.Row
return self._local.conn
def _init_db(self):
self.conn.executescript("""
CREATE TABLE IF NOT EXISTS plans (
plan_id TEXT PRIMARY KEY,
goal TEXT NOT NULL,
status TEXT DEFAULT 'pending',
context TEXT DEFAULT '{}',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE IF NOT EXISTS subtasks (
id TEXT PRIMARY KEY,
plan_id TEXT NOT NULL,
description TEXT,
action TEXT,
parameters TEXT DEFAULT '{}',
dependencies TEXT DEFAULT '[]',
status TEXT DEFAULT 'pending',
result TEXT,
error TEXT DEFAULT '',
retry_count INTEGER DEFAULT 0,
max_retries INTEGER DEFAULT 3,
timeout_seconds INTEGER DEFAULT 60,
started_at REAL,
completed_at REAL,
FOREIGN KEY (plan_id) REFERENCES plans(plan_id)
);
CREATE INDEX IF NOT EXISTS idx_subtasks_plan
ON subtasks(plan_id, status);
""")
self.conn.commit()
def save_plan(self, plan: ExecutionPlan):
self.conn.execute("""
INSERT OR REPLACE INTO plans (plan_id, goal, status, context, updated_at)
VALUES (?, ?, ?, ?, ?)
""", (plan.plan_id, plan.goal, plan.status.value,
json.dumps(plan.context, ensure_ascii=False), datetime.now().isoformat()))
for task in plan.subtasks:
self.conn.execute("""
INSERT OR REPLACE INTO subtasks
(id, plan_id, description, action, parameters, dependencies,
status, result, error, retry_count, max_retries, timeout_seconds,
started_at, completed_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", (task.id, plan.plan_id, task.description, task.action,
json.dumps(task.parameters, ensure_ascii=False),
json.dumps(task.dependencies),
task.status.value,
json.dumps(task.result, ensure_ascii=False) if task.result else None,
task.error, task.retry_count, task.max_retries,
task.timeout_seconds, task.started_at, task.completed_at))
self.conn.commit()
def load_plan(self, plan_id: str) -> ExecutionPlan:
"""从数据库恢复执行计划（断点续传）"""
row = self.conn.execute(
"SELECT * FROM plans WHERE plan_id = ?", (plan_id,)
).fetchone()
if not row:
raise ValueError(f"Plan {plan_id} not found")
plan = ExecutionPlan(
plan_id=row["plan_id"],
goal=row["goal"],
status=TaskStatus(row["status"]),
context=json.loads(row["context"])
)
task_rows = self.conn.execute(
"SELECT * FROM subtasks WHERE plan_id = ?", (plan_id,)
).fetchall()
for tr in task_rows:
task = SubTask(
id=tr["id"],
description=tr["description"],
action=tr["action"],
parameters=json.loads(tr["parameters"]),
dependencies=json.loads(tr["dependencies"]),
status=TaskStatus(tr["status"]),
result=json.loads(tr["result"]) if tr["result"] else None,
error=tr["error"],
retry_count=tr["retry_count"],
max_retries=tr["max_retries"],
timeout_seconds=tr["timeout_seconds"],
started_at=tr["started_at"] or 0,
completed_at=tr["completed_at"] or 0,
)
plan.subtasks.append(task)
return plan3.2 编排引擎核心
import concurrent.futures
import time
import logging
logger = logging.getLogger(__name__)
class OrchestrationEngine:
"""任务编排引擎：DAG 拓扑排序 + 并行调度 + 超时控制"""
def __init__(self, tool_registry: dict, store: PlanStore):
self.tool_registry = tool_registry # action_name -> callable
self.store = store
self._running = {}
def execute(self, plan: ExecutionPlan) -> ExecutionPlan:
"""执行整个计划"""
plan.status = TaskStatus.RUNNING
self.store.save_plan(plan)
# 拓扑排序确定执行顺序
execution_order = self._topological_sort(plan.subtasks)
logger.info(f"计划 {plan.plan_id}: {len(plan.subtasks)} 个子任务, "
f"{len(execution_order)} 个批次")
for batch_idx, batch in enumerate(execution_order):
logger.info(f"执行批次 {batch_idx + 1}/{len(execution_order)}: "
f"{[t.id for t in batch]}")
self._execute_batch(plan, batch)
# 持久化中间状态（支持断点恢复）
self.store.save_plan(plan)
# 检查是否有无法恢复的失败
if self._has_unrecoverable_failure(plan):
plan.status = TaskStatus.FAILED
self.store.save_plan(plan)
logger.error(f"计划 {plan.plan_id} 因不可恢复错误终止")
return plan
# 所有子任务完成
plan.status = TaskStatus.COMPLETED if all(
t.status == TaskStatus.COMPLETED for t in plan.subtasks
) else TaskStatus.FAILED
self.store.save_plan(plan)
return plan
def _topological_sort(self, tasks: list) -> list:
"""Kahn 算法：按 DAG 层级分批，同一批可并行"""
task_map = {t.id: t for t in tasks}
in_degree = {t.id: 0 for t in tasks}
dependents = {t.id: [] for t in tasks}
for t in tasks:
for dep in t.dependencies:
if dep in task_map:
in_degree[t.id] += 1
dependents[dep].append(t.id)
# 分层 BFS
batches = []
current = [tid for tid, deg in in_degree.items() if deg == 0]
while current:
batch = [task_map[tid] for tid in current]
batches.append(batch)
next_level = []
for tid in current:
for dep_tid in dependents[tid]:
in_degree[dep_tid] -= 1
if in_degree[dep_tid] == 0:
next_level.append(dep_tid)
current = next_level
return batches
def _execute_batch(self, plan: ExecutionPlan, batch: list):
"""并行执行一个批次内的子任务"""
with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
futures = {}
for task in batch:
if self._check_dependencies_failed(plan, task):
task.status = TaskStatus.SKIPPED
task.error = "依赖任务失败，跳过执行"
continue
future = executor.submit(self._run_task_with_timeout, task)
futures[future] = task
for future in concurrent.futures.as_completed(futures):
task = futures[future]
try:
task.result = future.result()
except concurrent.futures.TimeoutError:
task.status = TaskStatus.FAILED
task.error = f"执行超时 ({task.timeout_seconds}s)"
logger.warning(f"任务 {task.id} 超时")
except Exception as e:
task.status = TaskStatus.FAILED
task.error = str(e)
logger.error(f"任务 {task.id} 失败: {e}")
def _run_task_with_timeout(self, task: SubTask):
"""执行单个子任务，带超时控制"""
task.status = TaskStatus.RUNNING
task.started_at = time.time()
tool = self.tool_registry.get(task.action)
if not tool:
raise ValueError(f"未注册的工具: {task.action}")
# 实际工程中使用 signal / multiprocessing 实现硬超时
result = tool(**task.parameters)
task.completed_at = time.time()
task.status = TaskStatus.COMPLETED
return result
def _check_dependencies_failed(self, plan: ExecutionPlan, task: SubTask) -> bool:
"""检查是否有依赖任务已失败"""
task_map = {t.id: t for t in plan.subtasks}
for dep_id in task.dependencies:
dep_task = task_map.get(dep_id)
if dep_task and dep_task.status == TaskStatus.FAILED:
return True
return False
def _has_unrecoverable_failure(self, plan: ExecutionPlan) -> bool:
"""检查是否存在不可恢复的失败（重试耗尽）"""
for t in plan.subtasks:
if t.status == TaskStatus.FAILED and t.retry_count >= t.max_retries:
return True
return False4. 错误恢复与自校正机制
4.1 指数退避重试 + 降级策略
class RetryPolicy:
"""带指数退避和熔断的重试策略"""
def __init__(self, max_retries=3, base_delay=1.0, max_delay=60.0,
exponential_base=2.0, retryable_errors=None):
self.max_retries = max_retries
self.base_delay = base_delay
self.max_delay = max_delay
self.exponential_base = exponential_base
self.retryable_errors = retryable_errors or (TimeoutError, ConnectionError)
self._failure_counts = {} # 简易熔断计数器
def should_retry(self, task: SubTask, error: Exception) -> bool:
"""判断是否应重试"""
if task.retry_count >= self.max_retries:
return False
if isinstance(error, self.retryable_errors):
return True
# 对 LLM 输出的解析错误也允许重试
if isinstance(error, (json.JSONDecodeError, ValueError)):
return True
return False
def get_delay(self, retry_count: int) -> float:
"""指数退避延迟"""
import random
delay = self.base_delay * (self.exponential_base ** retry_count)
# 加 jitter 防止惊群
delay = delay * (0.5 + random.random() * 0.5)
return min(delay, self.max_delay)
def is_circuit_open(self, action: str) -> bool:
"""简易熔断：连续失败 5 次后熔断 30s"""
failures = self._failure_counts.get(action, [])
recent = [t for t in failures if time.time() - t < 30]
self._failure_counts[action] = recent
return len(recent) >= 5
def record_failure(self, action: str):
if action not in self._failure_counts:
self._failure_counts[action] = []
self._failure_counts[action].append(time.time())4.2 自校正：让 Agent 从失败中学习
class SelfCorrectionEngine:
"""自校正引擎：捕获执行失败并让 LLM 提出修复方案"""
CORRECTION_PROMPT = """你正在执行一个任务编排计划，以下子任务失败了：
任务描述: {task_description}
工具调用: {tool_name}({params})
错误信息: {error_message}
已完成的前置任务结果: {completed_results}
请分析失败原因并提供修复方案。输出格式：
{
"root_cause": "根本原因分析",
"fix_strategy": "retry_with_adjusted_params | substitute_tool | skip_with_compensation | abort",
"new_params": {修改后的参数},
"alternative_tool": "替代工具名（如适用）"
}"""
def __init__(self, llm_client, max_correction_rounds=2):
self.llm = llm_client
self.max_correction_rounds = max_correction_rounds
async def attempt_correction(self, failed_task: SubTask,
plan: ExecutionPlan) -> dict:
"""尝试自动校正失败任务"""
task_map = {t.id: t for t in plan.subtasks}
completed = {
t.id: t.result for t in plan.subtasks
if t.status == TaskStatus.COMPLETED
}
prompt = self.CORRECTION_PROMPT.format(
task_description=failed_task.description,
tool_name=failed_task.action,
params=json.dumps(failed_task.parameters, ensure_ascii=False),
error_message=failed_task.error,
completed_results=json.dumps(completed, ensure_ascii=False, default=str)[:2000]
)
for attempt in range(self.max_correction_rounds):
response = await self.llm.chat(prompt)
try:
correction = json.loads(self._extract_json(response))
if correction["fix_strategy"] == "abort":
return {"action": "abort", "reason": correction["root_cause"]}
elif correction["fix_strategy"] == "retry_with_adjusted_params":
return {
"action": "retry",
"new_params": correction.get("new_params", {})
}
elif correction["fix_strategy"] == "substitute_tool":
return {
"action": "substitute",
"new_tool": correction["alternative_tool"],
"new_params": correction.get("new_params", {})
}
elif correction["fix_strategy"] == "skip_with_compensation":
return {
"action": "skip",
"compensation_description": correction.get("compensation", "")
}
except (json.JSONDecodeError, KeyError) as e:
logger.warning(f"校正响应解析失败 (尝试 {attempt + 1}): {e}")
continue
return {"action": "abort", "reason": "自动校正失败，超出最大尝试次数"}
def _extract_json(self, text: str) -> str:
start = text.index('{')
end = text.rindex('}') + 1
return text[start:end]5. 人在回路：审批与协作机制
长任务中不可避免需要人类介入——金额审批、敏感操作确认、信息补充。工程实现需要解决异步通知和超时处理。
import asyncio
from enum import Enum
class ApprovalStatus(Enum):
PENDING = "pending"
APPROVED = "approved"
REJECTED = "rejected"
TIMED_OUT = "timed_out"
class HumanInTheLoop:
"""人在回路审批管理器"""
def __init__(self, notification_service, default_timeout=3600):
self.notifier = notification_service
self.default_timeout = default_timeout
self._pending = {} # task_id -> {"future": Future, "status": ApprovalStatus}
async def request_approval(self, task: SubTask, plan: ExecutionPlan,
reason: str) -> bool:
"""请求人类审批，返回是否通过"""
approval_id = f"approval_{task.id}"
# 创建异步 Future
loop = asyncio.get_event_loop()
future = loop.create_future()
self._pending[approval_id] = {
"future": future,
"status": ApprovalStatus.PENDING,
"task_id": task.id,
"requested_at": time.time()
}
# 发送通知（邮件/钉钉/飞书/企业微信）
await self.notifier.send(
title=f"需要审批: {task.description}",
body=f"计划: {plan.goal}\n任务: {task.description}\n"
f"原因: {reason}\n工具: {task.action}({task.parameters})",
approval_id=approval_id,
actions=["approve", "reject", "modify"]
)
try:
result = await asyncio.wait_for(
future, timeout=self.default_timeout
)
self._pending.pop(approval_id, None)
return result
except asyncio.TimeoutError:
self._pending[approval_id]["status"] = ApprovalStatus.TIMED_OUT
# 超时后的默认策略：根据任务配置决定跳过或终止
task.error = "审批超时"
self._pending.pop(approval_id, None)
return False
def resolve_approval(self, approval_id: str, decision: str,
modified_params: dict = None):
"""外部回调：用户做出审批决定"""
if approval_id in self._pending:
pending = self._pending[approval_id]
if decision == "approve":
pending["status"] = ApprovalStatus.APPROVED
if not pending["future"].done():
pending["future"].set_result(True)
elif decision == "reject":
pending["status"] = ApprovalStatus.REJECTED
if not pending["future"].done():
pending["future"].set_result(False)6. 生产级可观测性
长任务的可观测性是工程落地的关键。必须做到：实时进度可见、异常及时告警、事后可追溯分析。
from contextlib import contextmanager
from dataclasses import dataclass
from typing import Callable
import uuid
@dataclass
class TaskSpan:
"""类似 OpenTelemetry Span 的任务执行追踪单元"""
trace_id: str
span_id: str
task_id: str
action: str
status: str = "started"
events: list = None
start_time: float = 0
end_time: float = 0
attributes: dict = None
def __post_init__(self):
if self.events is None:
self.events = []
if self.attributes is None:
self.attributes = {}
self.start_time = time.time()
def add_event(self, name: str, attributes: dict = None):
self.events.append({
"name": name,
"timestamp": time.time(),
"attributes": attributes or {}
})
def end(self, status: str = "ok", error: str = None):
self.end_time = time.time()
self.status = status
if error:
self.add_event("error", {"message": error})
@property
def duration_ms(self) -> float:
if self.end_time and self.start_time:
return (self.end_time - self.start_time) * 1000
return 0
class TracingOrchestrationEngine(OrchestrationEngine):
"""带内建追踪的编排引擎"""
def __init__(self, *args, trace_sink: Callable = None, **kwargs):
super().__init__(*args, **kwargs)
self.trace_sink = trace_sink or self._default_trace_sink
self._active_traces = {}
def execute(self, plan: ExecutionPlan) -> ExecutionPlan:
trace_id = str(uuid.uuid4())[:16]
self._active_traces[plan.plan_id] = {
"trace_id": trace_id,
"spans": [],
"start_time": time.time()
}
try:
result = super().execute(plan)
return result
finally:
trace_data = self._active_traces.pop(plan.plan_id)
trace_data["end_time"] = time.time()
trace_data["duration_ms"] = (
trace_data["end_time"] - trace_data["start_time"]
) * 1000
trace_data["plan_id"] = plan.plan_id
trace_data["goal"] = plan.goal
trace_data["final_status"] = plan.status.value
self.trace_sink(trace_data)
def _run_task_with_timeout(self, task: SubTask):
span = TaskSpan(
trace_id=self._active_traces.get(
task.id.split("_")[0] if "_" in task.id else "default",
{"trace_id": "unknown"}
).get("trace_id", "unknown"),
span_id=str(uuid.uuid4())[:12],
task_id=task.id,
action=task.action,
attributes={
"parameters": task.parameters,
"retry_count": task.retry_count
}
)
try:
result = super()._run_task_with_timeout(task)
span.end(status="ok", error=None)
span.attributes["result_type"] = type(result).__name__
return result
except Exception as e:
span.end(status="error", error=str(e))
raise
finally:
# 将 span 关联到当前 plan 的 trace
for plan_id, trace in self._active_traces.items:
trace["spans"].append(span.__dict__)
def _default_trace_sink(self, trace_data: dict):
"""默认追踪输出：可替换为 OTel exporter / 日志 / InfluxDB"""
logger.info(
f"[Trace] plan={trace_data['plan_id']} "
f"status={trace_data['final_status']} "
f"duration={trace_data['duration_ms']:.0f}ms "
f"spans={len(trace_data['spans'])}"
)7. 实战案例：自动化部署流水线 Agent
以下是一个完整的实战案例——部署 Agent 使用长任务编排引擎协调代码构建、测试、人工审批和发布的全流程：
import asyncio
class DeployAgent:
"""部署 Agent：使用长任务编排引擎管理部署流水线"""
def __init__(self):
self.store = PlanStore()
self.tools = self._register_tools()
self.engine = TracingOrchestrationEngine(
tool_registry=self.tools,
store=self.store,
trace_sink=self._export_to_clickhouse
)
self.retry_policy = RetryPolicy(max_retries=3, base_delay=2.0)
self.human_loop = HumanInTheLoop(notification_service=DingTalkNotifier())
def _register_tools(self) -> dict:
return {
"git_clone": git_clone_tool,
"run_tests": run_tests_tool,
"build_container": build_container_tool,
"deploy_staging": deploy_staging_tool,
"run_smoke_tests": run_smoke_tests_tool,
"deploy_production": deploy_production_tool,
"notify_slack": notify_slack_tool,
}
async def deploy(self, repo: str, branch: str, target: str = "staging"):
"""执行完整部署流程"""
plan = await self._create_deployment_plan(repo, branch, target)
# 提交审批
approved = await self.human_loop.request_approval(
task=plan.subtasks[0],
plan=plan,
reason=f"发起部署: {repo}@{branch} → {target}"
)
if not approved:
plan.status = TaskStatus.FAILED
self.store.save_plan(plan)
return plan
# 执行编排
result = self.engine.execute(plan)
return result
async def _create_deployment_plan(self, repo, branch, target) -> ExecutionPlan:
generator = PlanGenerator()
tools_spec = [
{"name": "git_clone", "description": "克隆代码仓库"},
{"name": "run_tests", "description": "运行单元测试和集成测试"},
{"name": "build_container", "description": "构建容器镜像"},
{"name": "deploy_staging", "description": "部署到预发环境"},
{"name": "run_smoke_tests", "description": "运行冒烟测试"},
{"name": "deploy_production", "description": "部署到生产环境"},
{"name": "notify_slack", "description": "通知团队"},
]
return generator.generate(
goal=f"将 {repo} 的 {branch} 分支部署到 {target} 环境",
tools=tools_spec
)
def _export_to_clickhouse(self, trace_data: dict):
"""导出追踪数据到 ClickHouse 进行分析"""
pass
# 工具函数示例
def git_clone_tool(repo: str, branch: str) -> dict:
import subprocess
result = subprocess.run(
["git", "clone", "--branch", branch, "--depth", "1", repo, "/tmp/build"],
capture_output=True, text=True, timeout=120
)
if result.returncode != 0:
raise RuntimeError(f"Git clone failed: {result.stderr}")
return {"status": "success", "path": "/tmp/build"}
def run_tests_tool() -> dict:
import subprocess
result = subprocess.run(
["pytest", "--tb=short", "-q"], capture_output=True,
text=True, timeout=300
)
if result.returncode != 0:
raise RuntimeError(f"Tests failed: {result.stderr}")
return {"status": "all_passed", "output": result.stdout[:500]}8. 常见陷阱与最佳实践
8.1 避免的核心反模式
- 过度分解：将可以原子执行的操作拆分为多个子任务，增加调度开销和网络往返
- 乐观执行：假设每个子任务都会成功，没有设计补偿和回滚路径
- 状态不透明：用户无法了解任务执行到第一步、卡在哪里
- 无限重试：没有熔断机制导致死循环
8.2 最佳实践清单
| 维度 | 实践 |
|---|---|
| 可靠性 | 所有子任务设计为幂等操作，支持安全重试 |
| 持久化 | 每次子任务完成后立即持久化状态，进程崩溃可从断点恢复 |
| 超时控制 | 每个子任务硬超时 + 整个任务全局超时 |
| 可观测 | 内建分布式追踪，每个 step 的输入输出和耗时都要记录 |
| 成本控制 | 限制单次计划中 LLM 调用次数（预算 cap） |
| 安全防护 | 沙箱执行用户提供的参数，防止 prompt injection 蔓延 |
发表评论 取消回复