# 多 Agent 协作的策略评测平台：回测 / 过拟合检测 / Walk-Forward 全链路

最近半年在做一个稍微复杂一点的 Agent 系统：让多个 Agent 协作完成"策略开发 → 回测 → 过拟合检测 → Walk-Forward 验证 → 实盘小资金验证"的端到端量化策略评测流程。这篇文章把整个多 Agent 协作系统的架构设计、协作机制、状态同步、容错处理全部整理出来，重点依然是工程实现，不是金融业务。

如果你正在设计多 Agent 系统，或者在思考"多个 LLM 怎么协作做金融研究"，这篇文章大概率能给你一些参考。

## 一、为什么需要多 Agent 而不是单 Agent

在动手之前，我需要先回答一个基础问题：为什么不直接做一个"超级单 Agent"，让它自己完成策略评测的所有事情？

我尝试过这条路，做了一个"全能 Agent"，给它注册 15 个 Tool，让它从回测到过拟合检测全部自己做。结果发现三个致命问题。

**问题 1：上下文窗口爆炸**

单 Agent 完成一个完整的策略评测需要调用 15 个 Tool，每个 Tool 的输出都进入消息历史。到任务后半段，单次 LLM 输入超过 80K Token，推理延迟从 3 秒飙升到 25 秒，成本从 $0.03 飙升到 $0.5 。

**问题 2：角色混乱**

单 Agent 同时要做"策略代码生成""回测执行""过拟合检测""Walk-Forward 验证""报告生成"，但每种角色的最佳 prompt、最佳模型、最佳工具集都不一样。比如"回测执行"适合 Claude Haiku 这种快模型 结构化 Tool；"过拟合检测"适合 Claude Sonnet 这种推理强的模型；"报告生成"适合 GPT-4 这种长上下文强的模型。强行用一个 Agent 一个模型，会顾此失彼。

**问题 3：难以迭代**

单 Agent 是"一锅炖"，改任何一个 Tool 或 prompt 都要全量回归测试。多 Agent 是"模块化"的，改其中一个 Agent 只需要回归测试这个 Agent 上游下游对接，不会波及其他模块。

基于这三个原因，多 Agent 协作是必然选择。

## 二、整体架构：5 个 Agent 1 个 Coordinator

我设计的系统包含 5 个专业 Agent 1 个协调 Agent，每个 Agent 都有自己的职责、工具集、模型偏好。

```text

Coordinator Agent（协调者）

↓ 任务分发

├─→ CodeGenerator Agent（策略代码生成）

├─→ BacktestExecutor Agent（回测执行）

├─→ OverfitDetector Agent（过拟合检测）

├─→ WalkForwardValidator Agent（Walk-Forward 验证）

└─→ ReportGenerator Agent（报告生成）

↓ 状态同步

Shared Memory Pool（共享记忆池）

```

每个 Agent 的职责分工：

- **CodeGenerator Agent**：根据策略描述生成 Python 策略代码。负责"策略代码开发"。

- **BacktestExecutor Agent**：基于本地行情数据执行回测。负责"回测执行"。

- **OverfitDetector Agent**：检测策略是否过拟合。负责"过拟合识别"。

- **WalkForwardValidator Agent**：做滚动 Walk-Forward 测试。负责"稳健性验证"。

- **ReportGenerator Agent**：把所有 Agent 的输出汇总成评测报告。负责"最终输出"。

- **Coordinator Agent**：负责任务分发、状态跟踪、异常处理、最终决策。负责"协调"。

三、Agent 间通信：消息中间件选型

多 Agent 协作最关键的工程问题是"它们怎么通信"。我尝试了三种方案。

### 3.1 方案一：直接函数调用（最初版本）

最初我的实现是 Agent A 直接调用 Agent B 的 Python 函数：

```python

def backtest_executor_agent(strategy_code):

result = backtest(strategy_code)

overfit_result = overfit_detector_agent(result)

return overfit_result

```

这种实现的问题是"同步阻塞"——一个 Agent 卡住会阻塞所有后续 Agent。改成异步版本后问题变成"状态难追踪"——一个 Agent 出错后，没法知道其他 Agent 处于什么状态。

放弃这个方案。

### 3.2 方案二：消息队列（Redis Streams）

第二个版本用 Redis Streams 做消息队列。每个 Agent 都是独立的进程，从 Redis Streams 读任务、把结果写回 Redis Streams：

```python

class AgentWorker:

def __init__(self, agent_name, redis_url):

self.agent_name = agent_name

self.redis = redis.from_url(redis_url)

self.stream_key = f"agent:stream:{agent_name}"

async def run(self):

last_id = "0"

while True:

messages = self.redis.xread(

{self.stream_key: last_id},

block=1000,

count=1

)

for stream, msgs in messages:

for msg_id, data in msgs:

task = json.loads(data[b"task"])

try:

result = await self._execute_task(task)

self.redis.xadd(

f"agent:results:{task['parent_task_id']}",

{"result": json.dumps(result)}

)

except Exception as e:

self.redis.xadd(

f"agent:errors:{task['parent_task_id']}",

{"error": str(e)}

)

last_id = msg_id

```

Redis Streams 方案的好处是"异步 持久化 状态可追踪"。每个任务的消息 ID 都有时间戳，可以追溯每个 Agent 在什么时间做了什么。

### 3.3 方案三：消息中间件 共享状态（最终版本）

最终版本结合了 Redis Streams（异步消息） Shared Memory Pool（共享状态）。每个 Agent 把自己的中间结果写入 Shared Memory Pool，下游 Agent 从 Shared Memory Pool 读取：

```python

class SharedMemoryPool:

def __init__(self, redis_client):

self.redis = redis_client

def put(self, task_id, key, value, ttl=3600):

"""写入共享状态"""

self.redis.hset(f"task:{task_id}:state", key, json.dumps(value))

self.redis.expire(f"task:{task_id}:state", ttl)

def get(self, task_id, key):

"""读取共享状态"""

raw = self.redis.hget(f"task:{task_id}:state", key)

if raw is None:

return None

return json.loads(raw)

def wait_for_keys(self, task_id, keys, timeout=300):

"""等待多个上游 key 都就绪"""

start = time.time()

while time.time() - start < timeout:

all_keys = self.redis.hkeys(f"task:{task_id}:state")

if all(k in all_keys for k in keys):

return True

time.sleep(0.5)

return False

```

这套架构的好处是**状态可追溯 任务可重放 跨 Agent 状态同步直接**。每个任务的中间状态都在 Shared Memory Pool 里，任何一个 Agent 出错都可以重放而不需要从头开始。

四、任务编排：Coordinator Agent 的核心逻辑

Coordinator Agent 是整个系统的"大脑"，负责把策略评测任务拆解成 DAG 任务图、分发给各个 Agent、跟踪每个 Agent 的状态。

```python

class CoordinatorAgent:

def __init__(self, memory_pool, agent_registry):

self.memory = memory_pool

self.agents = agent_registry

async def evaluate_strategy(self, strategy_description: str):

# Step 1: 任务拆解，把策略评测任务拆成 DAG

task_graph = await self._decompose_evaluation_task(strategy_description)

# Step 2: 创建任务 ID

task_id = str(uuid.uuid4())

# Step 3: 按 DAG 顺序执行

completed = set()

while len(completed) < len(task_graph):

ready_nodes = [

node for node in task_graph

if node.node_id not in completed and

all(dep in completed for dep in node.depends_on)

]

dispatch_tasks = [

self._dispatch_node(task_id, node) for node in ready_nodes

]

await asyncio.gather(*dispatch_tasks, return_exceptions=True)

for node in ready_nodes:

await self._wait_node_complete(task_id, node.node_id)

completed.add(node.node_id)

# Step 4: 汇总结果，输出最终评测报告

return await self._generate_final_report(task_id)

```

策略评测的典型 DAG 任务图：

```text

[CodeGenerator] → [BacktestExecutor] → [OverfitDetector] → [ReportGenerator]

↓

[WalkForwardValidator] → [ReportGenerator]

```

注意 WalkForwardValidator 和 OverfitDetector 可以并行——它们都依赖 BacktestExecutor 的输出，但彼此不依赖。这套并发调度把策略评测的总耗时从 30 分钟压缩到 10 分钟。

## 五、状态一致性与容错处理

多 Agent 系统的另一个核心难题是"状态一致性"。一个 Agent 出错后，如何保证其他 Agent 不被污染？

### 5.1 三层状态模型

我把任务状态分成三层：

```text

Layer 1: 任务级状态（task:{task_id}:state）

- 所有 Agent 共享的中间结果

- TTL 1 小时，超时自动清理

Layer 2: Agent 级状态（agent:{agent_name}:state）

- Agent 自身的运行时状态（当前在处理哪个任务、推理进度）

- 持久化到 Redis

Layer 3: 全局级状态（global:state）

- 跨任务的全局信息（如模型配置、Tool 注册表更新）

- 用 Redis Pub/Sub 广播变更

```

### 5.2 失败重试与降级

每个 Agent 都包装了一层"重试 降级"机制：

```python

class ResilientAgent:

def __init__(self, agent, max_retries=3):

self.agent = agent

self.max_retries = max_retries

async def execute_with_retry(self, task):

last_error = None

for attempt in range(self.max_retries):

try:

return await self.agent.run(task)

except Exception as e:

last_error = e

await asyncio.sleep(2 ** attempt)

return await self._fallback_to_rules(task)

async def _fallback_to_rules(self, task):

"""降级：用预定义规则替代 LLM 推理"""

return {"fallback": True, "result": "use local rules"}

```

降级机制是"多 Agent 系统能稳定运行"的关键。当 LLM API 故障、Tool 故障、网络故障时，系统可以降级到"本地规则引擎"，不至于完全崩溃。

### 5.3 死锁检测

多 Agent 协作的另一个坑是"死锁"——Agent A 等 Agent B、Agent B 等 Agent C、Agent C 等 Agent A。我的解决方法是 Coordinator Agent 每 30 秒扫描一次所有任务，检测"等待时间超过阈值"的任务，自动报警 强制中断。

## 六、性能优化经验

半年实战下来，积累了一些性能优化经验。

**优化 1：Agent 复用**

最初的实现是"每个任务启动一个新 Agent 进程"，导致任务切换开销很大。改成"Agent 长连接 任务队列"后，吞吐量提升了 8 倍。Agent 进程启动一次后一直运行，任务从 Redis 队列里读取。

**优化 2：批量 Tool 调用**

Agent 在做"全市场策略回测"这种任务时，需要调用 5000 只股票的 Tool。每次 Tool 调用 1-3 秒，5000 次就是 1.5 万秒。改成"批量 Tool 调用"（一次调用返回 100 只股票的结果）后，耗时降到 50 秒，提升 300 倍。

**优化 3：模型分层使用**

不是每个 Agent 都需要 GPT-4 这么强的模型。CodeGenerator Agent 用 Claude Haiku（快、便宜）就够了，OverfitDetector Agent 才用 Claude Sonnet（推理强）。模型分层使用后，整个系统的 LLM 成本降了 70%。

**优化 4：共享 LLM 客户端**

最初每个 Agent 都自己 new 一个 LLM 客户端（带连接池），浪费资源。改成"全局共享 LLM 客户端池"后，连接数从 100 降到 20，内存占用降了一半。

## 七、可观测性：让 Agent 系统"可调试"

多 Agent 系统的"可调试性"比单 Agent 差得多——一个问题可能涉及 5 个 Agent、15 个 Tool 调用、3 个 LLM 推理。

```python

class AgentTracer:

def __init__(self):

self.langsmith = LangSmithClient()

def trace_evaluation_task(self, task_id):

"""追踪整个评测任务的执行链路"""

all_states = self.memory.get_all(task_id)

trace_tree = []

for agent_name, agent_state in all_states.items():

trace_tree.append({

"agent": agent_name,

"events": agent_state.get("events", []),

"tool_calls": agent_state.get("tool_calls", []),

"llm_calls": agent_state.get("llm_calls", []),

"start_time": agent_state.get("start_time"),

"end_time": agent_state.get("end_time"),

"tokens_used": agent_state.get("tokens_used", 0),

"cost_usd": agent_state.get("cost_usd", 0)

})

return trace_tree

```

可观测性的核心在于“把每个 Agent 的执行细节完整记录下来”——比如何时开始、何时结束、调用了哪些 Tool、使用了几次 LLM、消耗了多少 Token、花费了多少钱。当这些数据接入 LangSmith 后，就能够通过时间轴视图清晰呈现整个任务的执行链路，从而让问题排查与定位更加高效，效率通常可提升 10 倍以上。

八、未来想做的事

最后聊聊接下来想做的几个方向。

**方向 1：Agent 自适应协作**

现在的多 Agent 系统是"静态"的任务图，每个任务都按预定义的 DAG 执行。下一步想做"动态任务图"——让 Coordinator Agent 根据任务难度自动调整 Agent 数量（简单策略评测用 3 个 Agent，复杂策略用 6 个 Agent）。

**方向 2：Agent 间的"知识蒸馏"**

现在的 Agent 各自有独立的 prompt 和 few-shot example，没有共享。下一步想做"知识蒸馏"——让表现好的 Agent 把经验蒸馏成可复用的"操作手册"，分发给其他 Agent。

**方向 3：可视化编排工具**

现在的任务图是写在 Python 代码里的，非工程师改不动。下一步想做一个"拖拽式"的可视化编排工具——业务专家可以像画流程图一样设计 Agent 协作流程，工具自动生成对应的 Python 代码。

九、写在最后

多 Agent 系统不是"几个 Agent 加起来"那么简单。它涉及到通信协议、状态同步、容错处理、可观测性、性能优化等多个工程难题，每一个都值得深入研究。

这套系统我从 0 搭到稳定运行花了大约 4 个月时间，中间踩了无数坑。最深的感悟是：**Agent 系统的工程化难度被低估了**——大部分人关注的是 LLM 的能力提升，但工程化才是让 Agent 系统真正稳定运行的关键。

如果你也在做多 Agent 系统，希望这篇文章能给你一些参考。

资料参考：ig50","createTime":1786590090,"ext":{"closeTextLink":0,"comment_ban":1,"description":"","focusRead":0},"fa vNum":0,"html":"","isOriginal":0,"likeNum":0,