上一章我们实现了多 Agent 协作的基本工作流：研究员查事实，推理员做判断，审核员把关，执行员输出回复，编排器把它们按链式或条件式流程串起来。但那个版本还比较"脆弱"——如果某个 Agent 调用失败，整个工作流就断了；我们也不知道每一步花了多长时间、中间状态是什么；更没法在任务中断后恢复执行。
本章要解决的就是这些工程化问题。我们把编排器升级成一个真正的调度器，加入状态管理、超时重试、执行追踪和可视化调试能力。
7.1 从工作流编排到任务调度
先厘清两个概念的区别：
- 工作流编排（Orchestration）：关注"谁来执行、按什么顺序执行"。比如先研究员后推理员，这是流程设计。
- 任务调度（Scheduling）：关注"执行过程中怎么管理状态、怎么处理异常、怎么追踪进度"。比如某个 Agent 超时了要不要重试、重试几次、失败后怎么降级。
上一章的 WorkflowOrchestrator 做的是编排，本章的 AgentScheduler 要做的是调度。调度器在编排的基础上增加了四层能力：
- 状态管理：记录任务当前执行到哪一步、每一步的输入输出是什么；
- 容错处理：超时中断、失败重试、降级策略；
- 执行追踪：记录每次调用的耗时、Token 用量、错误信息；
- 状态恢复：任务中断后可以从断点继续执行。
7.2 任务状态机设计
多 Agent 任务的执行过程本质上是一个状态机。每个任务有若干个状态，状态之间按规则流转：
1创建 → 排队 → 执行中 → 完成
2 ↓
3 失败 → 重试中 → 执行中
4 ↓
5 已取消
我们需要为每个任务维护一个状态对象：
1{
2 "task_id": "uuid", # 任务唯一标识
3 "status": "running", # 当前状态
4 "plan": [...], # 执行计划（Agent 序列）
5 "current_step": 0, # 当前执行到第几步
6 "context": {...}, # 中间结果累积
7 "traces": [...], # 每一步的执行追踪记录
8 "created_at": "...", # 创建时间
9 "updated_at": "...", # 最后更新时间
10}
状态机的核心思想是：每一步执行前后都更新状态，任何异常都不会导致状态丢失。这样即使程序崩溃，重启后也能从 workflow_logs.json 中读取未完成的任务继续执行。
7.3 调度器核心实现
下面是一个完整的、可直接运行的调度器实现。它复用了第六章的四个 Agent，但加入了状态管理、超时重试和追踪能力。
"""
agent_scheduler.py - 多 Agent 调度器
相比第六章的 WorkflowOrchestrator，本调度器增加了：
1. 任务状态机：每个任务有独立状态，支持断点恢复
2. 超时控制：单个 Agent 调用超时自动中断
3. 失败重试：可配置最大重试次数
4. 降级策略：重试耗尽后使用降级回复
5. 执行追踪：记录每步耗时、输入输出、错误信息
6. 持久化：任务状态自动保存到 JSON 文件
运行前请确保已安装：
pip install langchain langchain-openai python-dotenv
并创建 .env 文件，写入 OPENAI_API_KEY。
"""
import json
import os
import time
import uuid
from datetime import datetime
from typing import Any
from dotenv import load_dotenv
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
# 加载环境变量
load_dotenv()
# ========================
# 配置区
# ========================
MODEL_NAME = os.getenv("MODEL_NAME", "gpt-4o")
MODEL_BASE_URL = os.getenv("MODEL_BASE_URL", "")
STATE_FILE = "task_states.json" # 任务状态持久化文件
DEFAULT_TIMEOUT = 30 # 单步超时（秒）
DEFAULT_MAX_RETRIES = 2 # 最大重试次数
# ========================
# 工具定义（复用第六章）
# ========================
@tool
def get_weather(city: str) -> str:
"""查询指定城市的天气情况。
参数：
city: 城市名称，如"郑州"、"北京"
返回：
天气描述字符串
"""
# 实际项目中可替换为真实天气 API
weather_db = {
"郑州": "晴天，气温 25°C，风力 2 级，空气质量良",
"北京": "多云，气温 22°C，风力 3 级，空气质量轻度污染",
"上海": "小雨，气温 20°C，风力 1 级，空气质量优",
}
return weather_db.get(city, f"未找到 {city} 的天气信息，请确认城市名称是否正确。")
@tool
def get_stock_price(symbol: str) -> str:
"""查询股票当前价格。
参数：
symbol: 股票代码，如"AAPL"、"600519"
返回：
股价描述字符串
"""
price_db = {
"AAPL": "Apple 当前股价：$178.50，今日涨幅 +1.2%",
"GOOGL": "Google 当前股价：$141.80，今日涨幅 -0.5%",
"600519": "贵州茅台 当前股价：¥1,680.00，今日涨幅 +0.8%",
}
return price_db.get(symbol, f"未找到 {symbol} 的股价信息。")
# ========================
# Agent 工厂函数
# ========================
def _create_llm() -> ChatOpenAI:
"""创建统一的 LLM 实例，所有 Agent 共享。"""
kwargs = {"model": MODEL_NAME, "temperature": 0.3}
if MODEL_BASE_URL:
kwargs["base_url"] = MODEL_BASE_URL
return ChatOpenAI(**kwargs)
def create_researcher_agent() -> AgentExecutor:
"""研究员：负责查询外部事实信息。"""
llm = _create_llm()
tools = [get_weather, get_stock_price]
prompt = ChatPromptTemplate.from_messages([
("system", "你是研究员，负责调用工具查询事实信息。只返回工具输出的原始结果，不要添加分析。"),
MessagesPlaceholder(variable_name="chat_history"),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad"),
])
agent = create_tool_calling_agent(llm, tools, prompt)
return AgentExecutor(agent=agent, tools=tools, verbose=False)
def create_reasoner_agent() -> AgentExecutor:
"""推理员：负责分析和推理，不调用工具。"""
llm = _create_llm()
prompt = ChatPromptTemplate.from_messages([
("system", "你是推理员，负责根据已有信息进行逻辑分析和推理。给出清晰的推理过程和结论。"),
MessagesPlaceholder(variable_name="chat_history"),
("human", "{input}"),
])
# 推理员不使用工具，直接构造 AgentExecutor
agent = create_tool_calling_agent(llm, [], prompt)
return AgentExecutor(agent=agent, tools=[], verbose=False)
def create_reviewer_agent() -> AgentExecutor:
"""审核员：负责检查回复的准确性、安全性和完整性。"""
llm = _create_llm()
prompt = ChatPromptTemplate.from_messages([
("system", "你是审核员，负责检查回复是否存在事实错误、逻辑漏洞或不安全内容。"
"如果发现问题，指出问题并给出修改建议；如果没有问题，回复'审核通过'。"),
MessagesPlaceholder(variable_name="chat_history"),
("human", "{input}"),
])
agent = create_tool_calling_agent(llm, [], prompt)
return AgentExecutor(agent=agent, tools=[], verbose=False)
def create_executor_agent() -> AgentExecutor:
"""执行员：负责整合所有信息，生成最终回复给用户。"""
llm = _create_llm()
prompt = ChatPromptTemplate.from_messages([
("system", "你是执行员，负责整合研究员、推理员、审核员的输出，"
"生成对用户友好、清晰、完整的最终回复。"),
MessagesPlaceholder(variable_name="chat_history"),
("human", "{input}"),
])
agent = create_tool_calling_agent(llm, [], prompt)
return AgentExecutor(agent=agent, tools=[], verbose=False)
# ========================
# 任务状态管理
# ========================
class TaskState:
"""单个任务的状态对象。
属性说明：
task_id: 任务的唯一 UUID，用于标识和恢复
task: 用户的原始任务描述
status: 当前状态，取值为 pending/running/completed/failed/cancelled
plan: 执行计划，即 Agent 名称列表，如 ["researcher", "reasoner", "executor"]
current_step: 当前执行到第几步（从 0 开始）
context: 累积的中间结果，键为 Agent 名称，值为该步的输出
traces: 执行追踪记录列表，每步一条
created_at / updated_at: 时间戳
"""
def __init__(self, task: str, plan: list[str]):
self.task_id = str(uuid.uuid4())
self.task = task
self.status = "pending"
self.plan = plan
self.current_step = 0
self.context: dict[str, Any] = {}
self.traces: list[dict] = []
self.created_at = datetime.now().isoformat()
self.updated_at = self.created_at
def to_dict(self) -> dict:
"""序列化为字典，用于 JSON 存储。"""
return {
"task_id": self.task_id,
"task": self.task,
"status": self.status,
"plan": self.plan,
"current_step": self.current_step,
"context": self.context,
"traces": self.traces,
"created_at": self.created_at,
"updated_at": self.updated_at,
}
@classmethod
def from_dict(cls, data: dict) -> "TaskState":
"""从字典反序列化，用于从文件恢复状态。"""
state = cls(data["task"], data["plan"])
state.task_id = data["task_id"]
state.status = data["status"]
state.current_step = data["current_step"]
state.context = data.get("context", {})
state.traces = data.get("traces", [])
state.created_at = data.get("created_at", "")
state.updated_at = data.get("updated_at", "")
return state
def update(self):
"""更新最后修改时间。"""
self.updated_at = datetime.now().isoformat()
# ========================
# 调度器核心
# ========================
class AgentScheduler:
"""多 Agent 调度器。
职责：
1. 管理多个任务的状态（创建、执行、恢复、查询）
2. 为每一步执行提供超时控制和失败重试
3. 记录详细的执行追踪日志
4. 支持从持久化文件恢复未完成的任务
参数：
agents: Agent 名称到 AgentExecutor 实例的映射
config: 可选配置字典，支持以下键：
- max_retries: 最大重试次数，默认 2
- timeout: 单步超时秒数，默认 30
- state_file: 状态持久化文件路径，默认 task_states.json
"""
def __init__(self, agents: dict[str, AgentExecutor], config: dict | None = None):
self.agents = agents
self.config = config or {}
self.max_retries = self.config.get("max_retries", DEFAULT_MAX_RETRIES)
self.timeout = self.config.get("timeout", DEFAULT_TIMEOUT)
self.state_file = self.config.get("state_file", STATE_FILE)
self.states: dict[str, TaskState] = {} # task_id -> TaskState
self._load_states() # 启动时加载已有状态
# ---------- 持久化 ----------
def _load_states(self):
"""从 JSON 文件加载未完成的任务状态。"""
if not os.path.exists(self.state_file):
return
try:
with open(self.state_file, "r", encoding="utf-8") as f:
data = json.load(f)
for item in data:
state = TaskState.from_dict(item)
# 只恢复未完成的任务
if state.status in ("pending", "running"):
self.states[state.task_id] = state
print(f"[调度器] 从 {self.state_file} 恢复了 {len(self.states)} 个未完成任务。")
except (json.JSONDecodeError, KeyError) as e:
print(f"[调度器] 加载状态文件失败：{e}")
def _save_states(self):
"""将所有任务状态保存到 JSON 文件。"""
data = [s.to_dict() for s in self.states.values()]
with open(self.state_file, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2)
# ---------- 单步执行（含超时和重试）----------
def _invoke_with_retry(self, agent: AgentExecutor, input_text: str, step_name: str) -> dict:
"""带超时和重试的单步执行。
参数：
agent: 要执行的 AgentExecutor 实例
input_text: 传给 Agent 的输入文本
step_name: 步骤名称，用于日志记录
返回：
包含 output、retries、error 等信息的字典
"""
last_error = None
for attempt in range(1, self.max_retries + 1):
start_time = time.time()
try:
# 注意：LangChain 的 AgentExecutor 本身不支持 timeout 参数，
# 这里用简单的计时 + 异常捕获模拟超时控制。
# 在生产环境中，建议使用 asyncio.wait_for 或进程级超时。
result = agent.invoke({"input": input_text})
elapsed = time.time() - start_time
# 检查是否超时
if elapsed > self.timeout:
raise TimeoutError(f"步骤 '{step_name}' 执行超时（{elapsed:.1f}s > {self.timeout}s）")
return {
"output": result.get("output", ""),
"retries": attempt,
"error": None,
"elapsed": round(elapsed, 3),
}
except Exception as e:
last_error = str(e)
elapsed = time.time() - start_time
print(f" [重试] 步骤 '{step_name}' 第 {attempt} 次尝试失败：{e}")
# 如果还有重试机会，短暂等待后继续
if attempt < self.max_retries:
time.sleep(1)
# 所有重试都失败，返回降级结果
return {
"output": f"[降级] 步骤 '{step_name}' 执行失败，已重试 {self.max_retries} 次。错误：{last_error}",
"retries": self.max_retries,
"error": last_error,
"elapsed": 0,
}
# ---------- 任务执行 ----------
def submit(self, task: str, plan: list[str]) -> str:
"""提交一个新任务。
参数：
task: 用户的任务描述
plan: Agent 执行序列，如 ["researcher", "reasoner", "executor"]
返回：
任务 ID（task_id）
"""
state = TaskState(task, plan)
self.states[state.task_id] = state
self._save_states()
print(f"[调度器] 任务已提交，task_id = {state.task_id}")
return state.task_id
def run(self, task_id: str) -> dict:
"""执行指定任务。
参数：
task_id: 要执行的任务 ID
返回：
任务最终状态字典
"""
if task_id not in self.states:
raise ValueError(f"任务 {task_id} 不存在")
state = self.states[task_id]
state.status = "running"
state.update()
print(f"\n{'=' * 60}")
print(f"开始执行任务：{state.task}")
print(f"执行计划：{' -> '.join(state.plan)}")
print(f"{'=' * 60}\n")
# 从 current_step 开始执行，支持断点续跑
for i in range(state.current_step, len(state.plan)):
agent_name = state.plan[i]
if agent_name not in self.agents:
raise ValueError(f"未知 Agent：{agent_name}")
agent = self.agents[agent_name]
state.current_step = i
state.update()
self._save_states()
# 构造输入：如果是第一步，用原始任务；否则加上前序结果
if i == 0:
input_text = state.task
else:
# 把前序所有结果拼接到输入中
context_parts = [f"[{k}] {v}" for k, v in state.context.items()]
input_text = state.task + "\n\n前序结果：\n" + "\n".join(context_parts)
print(f"[步骤 {i + 1}/{len(state.plan)}] 调用 Agent：{agent_name}")
trace = self._invoke_with_retry(agent, input_text, agent_name)
print(f" 耗时：{trace['elapsed']}s，重试：{trace['retries']}次")
# 记录追踪信息
trace["step"] = i
trace["agent"] = agent_name
trace["timestamp"] = datetime.now().isoformat()
state.traces.append(trace)
# 保存结果到上下文
state.context[agent_name] = trace["output"]
# 如果该步失败且无有效输出，可以选择中断或继续
if trace["error"] and not trace["output"].startswith("[降级]"):
state.status = "failed"
state.update()
self._save_states()
print(f"\n[调度器] 任务失败，已保存状态。")
return state.to_dict()
state.update()
self._save_states()
# 所有步骤完成
state.status = "completed"
state.current_step = len(state.plan)
state.update()
self._save_states()
final_output = state.context.get(state.plan[-1], "")
print(f"\n{'=' * 60}")
print(f"任务完成。最终回复：\n{final_output}")
print(f"{'=' * 60}\n")
return state.to_dict()
def get_state(self, task_id: str) -> dict | None:
"""查询任务状态。"""
state = self.states.get(task_id)
return state.to_dict() if state else None
def list_tasks(self) -> list[dict]:
"""列出所有任务摘要。"""
return [
{
"task_id": s.task_id,
"task": s.task,
"status": s.status,
"steps": f"{s.current_step}/{len(s.plan)}",
}
for s in self.states.values()
]
def cancel(self, task_id: str) -> bool:
"""取消一个任务。"""
state = self.states.get(task_id)
if not state:
return False
state.status = "cancelled"
state.update()
self._save_states()
return True
# ========================
# 交互式入口
# ========================
def main():
"""交互式运行入口。"""
# 创建四个 Agent
agents = {
"researcher": create_researcher_agent(),
"reasoner": create_reasoner_agent(),
"reviewer": create_reviewer_agent(),
"executor": create_executor_agent(),
}
# 创建调度器
scheduler = AgentScheduler(agents, config={
"max_retries": 2,
"timeout": 30,
"state_file": STATE_FILE,
})
print("\n多 Agent 调度器已就绪。")
print("可用命令：")
print(" run <任务描述> - 提交并执行任务（默认流程：researcher -> reasoner -> executor）")
print(" plan <任务描述> - 仅提交任务，不执行（可稍后用 run_id 执行）")
print(" run_id <task_id> - 执行已提交的任务")
print(" status <task_id> - 查询任务状态")
print(" list - 列出所有任务")
print(" cancel <task_id> - 取消任务")
print(" q - 退出\n")
while True:
try:
line = input(">>> ").strip()
except (EOFError, KeyboardInterrupt):
break
if not line:
continue
if line.lower() == "q":
break
parts = line.split(" ", 1)
cmd = parts[0].lower()
arg = parts[1] if len(parts) > 1 else ""
if cmd == "run":
if not arg:
print("用法：run <任务描述>")
continue
# 默认执行流程：研究员 -> 推理员 -> 执行员
default_plan = ["researcher", "reasoner", "executor"]
task_id = scheduler.submit(arg, default_plan)
scheduler.run(task_id)
elif cmd == "plan":
if not arg:
print("用法：plan <任务描述>")
continue
# 仅提交任务，不立即执行，稍后可用 run_id 手动触发
plan = ["researcher", "reasoner", "reviewer", "executor"]
task_id = scheduler.submit(arg, plan)
print(f"任务已提交，task_id = {task_id}")
print("稍后可用 run_id 执行。")
elif cmd == "run_id":
if not arg:
print("用法：run_id <task_id>")
continue
try:
scheduler.run(arg)
except ValueError as e:
print(f"错误：{e}")
elif cmd == "status":
if not arg:
print("用法：status <task_id>")
continue
state = scheduler.get_state(arg)
if state is None:
print("未找到该任务。")
else:
print(json.dumps(state, ensure_ascii=False, indent=2))
elif cmd == "list":
tasks = scheduler.list_tasks()
if not tasks:
print("暂无任务。")
else:
for t in tasks:
print(f" [{t['status']}] {t['task_id'][:8]}... | {t['task'][:40]} | {t['steps']}")
elif cmd == "cancel":
if not arg:
print("用法：cancel <task_id>")
continue
ok = scheduler.cancel(arg)
print("已取消。" if ok else "未找到该任务。")
else:
print(f"未知命令：{cmd}，输入 q 退出。")
if __name__ == "__main__":
main()
7.4 验证脚本
调度器写完后，需要一个验证脚本来测试各项能力。下面是一个可直接运行的 verify_scheduler.py：
1"""
2verify_scheduler.py - 调度器功能验证脚本
3
4测试覆盖：
51. 任务提交与执行
62. 状态持久化与恢复
73. 超时与重试
84. 执行追踪查询
95. 任务取消
10"""
11
12import json
13import time
14from agent_scheduler import AgentScheduler, create_researcher_agent, create_reasoner_agent, create_executor_agent
15
16def test_basic_execution():
17 """测试 1：基本执行流程。"""
18 print("\n" + "=" * 60)
19 print("测试 1：基本执行流程")
20 print("=" * 60)
21
22 agents = {
23 "researcher": create_researcher_agent(),
24 "reasoner": create_reasoner_agent(),
25 "executor": create_executor_agent(),
26 }
27 scheduler = AgentScheduler(agents, config={"state_file": "test_states.json"})
28
29 task_id = scheduler.submit("郑州今天天气怎么样，适合出门吗？", ["researcher", "executor"])
30 result = scheduler.run(task_id)
31
32 assert result["status"] == "completed", f"期望 completed，实际 {result['status']}"
33 assert "researcher" in result["context"], "研究员结果未保存"
34 assert "executor" in result["context"], "执行员结果未保存"
35 print("✓ 测试 1 通过")
36 return task_id
37
38
39def test_state_persistence():
40 """测试 2：状态持久化与恢复。"""
41 print("\n" + "=" * 60)
42 print("测试 2：状态持久化与恢复")
43 print("=" * 60)
44
45 # 新建一个调度器，从文件加载状态
46 agents = {
47 "researcher": create_researcher_agent(),
48 "reasoner": create_reasoner_agent(),
49 "executor": create_executor_agent(),
50 }
51 scheduler = AgentScheduler(agents, config={"state_file": "test_states.json"})
52
53 tasks = scheduler.list_tasks()
54 assert len(tasks) > 0, "未能从文件恢复任务"
55 print(f"✓ 恢复了 {len(tasks)} 个任务")
56
57 # 查询某个任务的状态
58 state = scheduler.get_state(tasks[0]["task_id"])
59 assert state is not None, "状态查询返回 None"
60 print(f"✓ 任务状态：{state['status']}")
61 print("✓ 测试 2 通过")
62
63
64def test_trace_logging():
65 """测试 3：执行追踪记录。"""
66 print("\n" + "=" * 60)
67 print("测试 3：执行追踪记录")
68 print("=" * 60)
69
70 agents = {
71 "researcher": create_researcher_agent(),
72 "reasoner": create_reasoner_agent(),
73 "executor": create_executor_agent(),
74 }
75 scheduler = AgentScheduler(agents, config={"state_file": "test_trace.json"})
76
77 task_id = scheduler.submit("北京天气如何？", ["researcher", "executor"])
78 result = scheduler.run(task_id)
79
80 traces = result.get("traces", [])
81 assert len(traces) >= 2, f"追踪记录不足，实际 {len(traces)} 条"
82
83 for trace in traces:
84 assert "agent" in trace, "追踪缺少 agent 字段"
85 assert "elapsed" in trace, "追踪缺少 elapsed 字段"
86 assert "timestamp" in trace, "追踪缺少 timestamp 字段"
87 print(f" [{trace['agent']}] 耗时 {trace['elapsed']}s")
88
89 print("✓ 测试 3 通过")
90
91
92def test_cancel():
93 """测试 4：任务取消。"""
94 print("\n" + "=" * 60)
95 print("测试 4：任务取消")
96 print("=" * 60)
97
98 agents = {
99 "researcher": create_researcher_agent(),
100 "executor": create_executor_agent(),
101 }
102 scheduler = AgentScheduler(agents, config={"state_file": "test_cancel.json"})
103
104 task_id = scheduler.submit("测试取消", ["researcher", "executor"])
105 ok = scheduler.cancel(task_id)
106 assert ok, "取消失败"
107
108 state = scheduler.get_state(task_id)
109 assert state["status"] == "cancelled", f"期望 cancelled，实际 {state['status']}"
110 print("✓ 测试 4 通过")
111
112
113def test_unknown_agent():
114 """测试 5：未知 Agent 报错。"""
115 print("\n" + "=" * 60)
116 print("测试 5：未知 Agent 报错")
117 print("=" * 60)
118
119 agents = {"researcher": create_researcher_agent()}
120 scheduler = AgentScheduler(agents, config={"state_file": "test_error.json"})
121
122 task_id = scheduler.submit("测试", ["nonexistent"])
123 try:
124 scheduler.run(task_id)
125 assert False, "应该抛出 ValueError"
126 except ValueError as e:
127 print(f" 捕获异常：{e}")
128 print("✓ 测试 5 通过")
129
130
131if __name__ == "__main__":
132 print("\n开始调度器验证测试...\n")
133
134 test_basic_execution()
135 test_state_persistence()
136 test_trace_logging()
137 test_cancel()
138 test_unknown_agent()
139
140 print("\n" + "=" * 60)
141 print("所有测试通过 ✓")
142 print("=" * 60)
运行验证脚本：
1python verify_scheduler.py
7.5 可观测性：从追踪日志到调试
调度器每执行一步都会记录一条追踪（trace），包含以下字段：
| 字段 | 含义 | 用途 |
|---|---|---|
agent | 执行的 Agent 名称 | 定位哪一步出了问题 |
step | 步骤序号 | 还原执行顺序 |
output | Agent 输出 | 查看中间结果 |
elapsed | 耗时（秒） | 发现慢查询或超时 |
retries | 重试次数 | 判断稳定性 |
error | 错误信息 | 快速定位异常 |
timestamp | 时间戳 | 分析时间分布 |
这些追踪数据保存在 task_states.json 中，可以用以下方式查看：
1import json
2
3with open("task_states.json", "r", encoding="utf-8") as f:
4 tasks = json.load(f)
5
6# 查看某个任务的执行链路
7for task in tasks:
8 print(f"\n任务：{task['task'][:30]}...")
9 for trace in task["traces"]:
10 status = "✓" if not trace["error"] else "✗"
11 print(f" {status} [{trace['agent']}] {trace['elapsed']}s (重试{trace['retries']}次)")
如果想做更深入的调试，可以把追踪导出为 CSV 或用 Pandas 分析：
1import csv
2
3traces = []
4for task in tasks:
5 for trace in task["traces"]:
6 traces.append({
7 "task_id": task["task_id"],
8 "agent": trace["agent"],
9 "elapsed": trace["elapsed"],
10 "retries": trace["retries"],
11 "has_error": bool(trace["error"]),
12 })
13
14with open("traces.csv", "w", newline="", encoding="utf-8") as f:
15 writer = csv.DictWriter(f, fieldnames=traces[0].keys())
16 writer.writeheader()
17 writer.writerows(traces)
18
19print(f"已导出 {len(traces)} 条追踪记录到 traces.csv")
7.6 本章小结
本章把第六章的编排器升级成了真正的调度器，核心变化如下：
| 能力 | 第六章编排器 | 第七章调度器 |
|---|---|---|
| 流程控制 | 串行/条件式执行 | 同左 + 断点续跑 |
| 状态管理 | 内存中临时变量 | 持久化 JSON 文件 |
| 容错 | 无 | 超时 + 重试 + 降级 |
| 可观测性 | 无 | 每步追踪日志 |
| 任务管理 | 单次执行 | 提交/查询/取消/恢复 |
调度器为第八章的评估和调试打下了数据基础——有了完整的追踪日志，我们才能做回放测试、性能分析和效果对比。
7.7 常见问题
-
状态文件越来越大怎么办？
可以定期清理已完成的任务，或把旧任务归档到单独的日志文件。 -
重试会不会导致重试风暴？
建议设置合理的max_retries（通常 2-3 次），并在重试之间加入退避延迟（如time.sleep(1 * attempt)）。 -
JSON 文件不适合生产环境怎么办？
生产环境可以用 SQLite、PostgreSQL 或 Redis 替代 JSON 文件，接口保持一致即可。 -
超时控制不够精确怎么办？
LangChain 的AgentExecutor本身不支持超时参数。如果需要精确超时，可以用asyncio.wait_for或在子进程中执行。 -
多进程同时写入状态文件会冲突吗？
会。多进程场景下应使用数据库或加文件锁（如filelock库）。
转载自 CSDN-专业IT技术社区
原文链接：https://blog.csdn.net/wodewojueding/article/details/167493914
