Week 24 复习——Multi-Agent 系统整合 v1

日期: 2026-10-10 方向: AI系统工程 / Agent 阶段: Phase 3 - Agent架构与多Agent (Day 149-162) 标签: #Review #Integration #MultiAgentV1 #Phase3Midway

---

今日目标

---

一、Phase 3 中段总结（Day 149-162 学了什么）

核心收获

1. Workflow 优先——大多数生产场景不需要 agent
2. Tool description 是高 ROI 优化点
3. 裸 SDK 先理解，框架后选型——LangGraph 适合生产复杂场景
4. 多 agent 不是 free lunch，是 3-10x 成本
5. Memory 分层——working / episodic / semantic / procedural
6. Onchain agent 必须 session key + simulate + budget cap
7. Eval 不是 nice-to-have，是生产必需

---

二、整合 v1 — 系统设计

2.1 用例

"用户：分析一下我的链上 portfolio，建议一个再平衡方案，并在 ≤ $5k 内自动执行调整。"

2.2 架构

┌──────────────────────────────────────────────────────────────────────┐
│                Multi-Agent V1 — Portfolio Rebalance                 │
│                                                                    │
│   user (cold wallet)                                               │
│      │                                                             │
│      │ creates session key (allowed: Uniswap router, $5k cap)       │
│      ▼                                                             │
│   Smart Account                                                     │
│      │                                                             │
│      │                                                             │
│   ┌──────────────────────────────────────────────────────────┐     │
│   │              Orchestrator (LangGraph StateGraph)          │     │
│   │                                                            │     │
│   │   START → load_memory → planner → research_team           │     │
│   │                              │     (Hierarchical)         │     │
│   │                              │      ├── fundamentals      │     │
│   │                              │      ├── onchain_data      │     │
│   │                              │      └── risk              │     │
│   │                              ▼                            │     │
│   │                          debate (Network)                 │     │
│   │                              │      bull <-> bear         │     │
│   │                              │      moderator             │     │
│   │                              ▼                            │     │
│   │                           proposer                        │     │
│   │                              │                            │     │
│   │                              ▼                            │     │
│   │                          [interrupt]                      │     │
│   │                          user_confirm                     │     │
│   │                              │                            │     │
│   │                              ▼                            │     │
│   │                          executor (with session key)      │     │
│   │                              │                            │     │
│   │                              ▼                            │     │
│   │                          verifier → save_to_memory → END  │     │
│   └──────────────────────────────────────────────────────────┘     │
│                                                                    │
│   Tools (via MCP servers):                                          │
│     - fin-mcp:    SEC, peers, ratios                                │
│     - onchain-mcp: balances, prices, simulate, execute              │
│   Memory:                                                           │
│     - 4-layer (Day 158)                                             │
│   Eval:                                                             │
│     - golden suite runs nightly                                     │
└──────────────────────────────────────────────────────────────────────┘

2.3 模式映射

---

三、代码骨架——multi_agent_v1/

multi_agent_v1/
├── README.md
├── pyproject.toml
├── config.yaml
├── .env.example
├── src/
│   ├── orchestrator.py     # LangGraph StateGraph
│   ├── memory.py           # 引用 Day 158
│   ├── research_team.py    # Hierarchical sub-graph
│   ├── debate.py           # Network sub-graph
│   ├── proposer.py         # generates rebalance plan
│   ├── executor.py         # session key + onchain
│   ├── tools/
│   │   ├── fin_tools.py    # 引用 Day 152 tools.py
│   │   └── onchain_tools.py
│   └── eval/               # 引用 Day 160 agent_eval.py
└── mcp_servers/
    ├── fin-mcp/            # 引用 Day 153 mcp_server.py
    └── onchain-mcp/

3.1 orchestrator.py（核心 200 行）

# orchestrator.py
"""
Multi-agent v1 orchestrator.
"""
from typing import Annotated, Literal, TypedDict
from operator import add
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.types import interrupt, Command

from .memory import MemoryManager
from .research_team import build_research_subgraph
from .debate import build_debate_subgraph
from .proposer import propose_rebalance
from .executor import execute_with_session_key

class S(TypedDict):
    user_id: str
    portfolio: dict
    research_findings: dict
    debate_verdict: str
    proposal: dict
    user_approval: str
    execution_result: dict
    rounds: Annotated[int, add]

def load_memory_node(state: S):
    mm = MemoryManager()
    bundle = mm.retrieve_for_query(state["user_id"], "portfolio rebalance preferences")
    return {"research_findings": {"memory_context": bundle}}

def planner_node(state: S):
    # Decide: do we need fundamentals research? onchain data?
    # For brevity, always run research
    return {}

def research_node(state: S):
    sub = build_research_subgraph()
    out = sub.invoke({"portfolio": state["portfolio"]})
    findings = dict(state.get("research_findings", {}))
    findings.update(out)
    return {"research_findings": findings}

def debate_node(state: S):
    sub = build_debate_subgraph()
    out = sub.invoke({"findings": state["research_findings"]})
    return {"debate_verdict": out["verdict"]}

def proposer_node(state: S):
    plan = propose_rebalance(state["portfolio"], state["debate_verdict"])
    return {"proposal": plan}

def confirm_node(state: S):
    answer = interrupt({
        "type": "rebalance_proposal",
        "proposal": state["proposal"],
        "estimated_cost_usd": state["proposal"].get("estimated_cost", 0),
    })
    return {"user_approval": answer}

def execute_node(state: S):
    if state["user_approval"] != "approve":
        return {"execution_result": {"status": "user_declined"}}
    res = execute_with_session_key(state["proposal"])
    return {"execution_result": res}

def verify_node(state: S):
    # post-trade balance check, save to memory
    mm = MemoryManager()
    mm.on_message(state["user_id"], "assistant",
                  f"Rebalance executed: {state['execution_result']}")
    mm.flush_facts(state["user_id"])
    return {}

def build_orchestrator(checkpointer):
    g = StateGraph(S)
    g.add_node("load_memory", load_memory_node)
    g.add_node("planner", planner_node)
    g.add_node("research", research_node)
    g.add_node("debate", debate_node)
    g.add_node("proposer", proposer_node)
    g.add_node("confirm", confirm_node)
    g.add_node("execute", execute_node)
    g.add_node("verify", verify_node)

    g.add_edge(START, "load_memory")
    g.add_edge("load_memory", "planner")
    g.add_edge("planner", "research")
    g.add_edge("research", "debate")
    g.add_edge("debate", "proposer")
    g.add_edge("proposer", "confirm")
    g.add_edge("confirm", "execute")
    g.add_edge("execute", "verify")
    g.add_edge("verify", END)
    return g.compile(checkpointer=checkpointer)

3.2 research_team.py（Hierarchical 子图）

# research_team.py
"""
Sub-graph: hierarchical research team.
"""
from typing import TypedDict, Annotated, Literal
from operator import add
from langgraph.graph import StateGraph, START, END
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage, SystemMessage

class RS(TypedDict):
    portfolio: dict
    findings: dict
    next_worker: str
    rounds: Annotated[int, add]

WORKERS = ["fundamentals", "onchain_data", "risk"]
opus = ChatAnthropic(model="claude-opus-4-7", max_tokens=1024)
sonnet = ChatAnthropic(model="claude-sonnet-4-6", max_tokens=800)

def manager(state: RS):
    done = list(state.get("findings", {}).keys())
    if state["rounds"] >= 5 or set(done) == set(WORKERS):
        return {"next_worker": "DONE", "rounds": 1}
    pending = [w for w in WORKERS if w not in done]
    return {"next_worker": pending[0], "rounds": 1}

def fund_w(state: RS):
    r = sonnet.invoke([
        SystemMessage(content="Fundamentals analyst. Score each portfolio holding A/B/C/D."),
        HumanMessage(content=str(state["portfolio"])),
    ])
    f = dict(state.get("findings", {}))
    f["fundamentals"] = r.content
    return {"findings": f, "next_worker": "manager"}

def onchain_w(state: RS):
    r = sonnet.invoke([
        SystemMessage(content="Onchain analyst. Identify which holdings have onchain risk (low liquidity, suspicious activity)."),
        HumanMessage(content=str(state["portfolio"])),
    ])
    f = dict(state.get("findings", {}))
    f["onchain_data"] = r.content
    return {"findings": f, "next_worker": "manager"}

def risk_w(state: RS):
    r = sonnet.invoke([
        SystemMessage(content="Risk analyst. Calculate concentration & correlation risks of the portfolio."),
        HumanMessage(content=str(state["portfolio"])),
    ])
    f = dict(state.get("findings", {}))
    f["risk"] = r.content
    return {"findings": f, "next_worker": "manager"}

def route(state: RS) -> Literal["fundamentals","onchain_data","risk","end"]:
    nx = state.get("next_worker")
    if nx in WORKERS: return nx
    return "end"

def build_research_subgraph():
    g = StateGraph(RS)
    g.add_node("manager", manager)
    g.add_node("fundamentals", fund_w)
    g.add_node("onchain_data", onchain_w)
    g.add_node("risk", risk_w)
    g.add_edge(START, "manager")
    g.add_conditional_edges("manager", route, {
        "fundamentals":"fundamentals","onchain_data":"onchain_data",
        "risk":"risk","end": END,
    })
    g.add_edge("fundamentals", "manager")
    g.add_edge("onchain_data", "manager")
    g.add_edge("risk", "manager")
    return g.compile()

同样模式，debate.py 实现 Network 拓扑（bull/bear/moderator），executor.py 用 Day 161 的 OnchainAgent + SessionKey。代码量约 800 行总和。

---

四、关键设计决策

4.1 为什么这种组合

• LangGraph orchestrator：需要 checkpoint + interrupt（user_confirm 节点）。
• Research 用 Hierarchical：需要多角色但有清晰主管。
• Debate 用 Network：bull/bear 对抗能挖出更全面 risk。
• MCP tools：跨客户端复用（同 fin-mcp 也给 Claude Desktop 用）。
• Memory 4 层：用户长期偏好 → semantic；本次研究流程 → procedural skill 模板。
• Session key 执行：限权 + 单笔 cap。
• Interrupt + user 确认：destructive 必须人工签。

4.2 不要做的事

• 不要给 debate 配 onchain 写权限——它只产 verdict
• 不要让 research workers 直接执行——只产 findings
• executor 是唯一能动钱的节点——单点权限收敛便于审计
• 不要把 memory 完整 dump 给所有节点——按需 retrieve

---

五、整合后端到端 Walkthrough

User: "Rebalance my portfolio. I'm in moderate-risk mode."

[orchestrator] start
[load_memory] retrieved facts: user.risk=moderate, user.tax_residence=US
[planner] decides: full research needed
[research → manager] picks fundamentals
[research → fundamentals_worker] each holding scored A-D
[research → manager] picks onchain_data
[research → onchain_data_worker] flags 1 low-liq token
[research → manager] picks risk
[research → risk_worker] portfolio HHI = 0.43 (concentrated in tech)
[research → manager] DONE

[debate → bull] "Tech concentration is justified by AI cycle, expect +20% next 12mo"
[debate → bear] "HHI 0.43 is dangerous if Fed pivots, suggest trim to 0.30"
[debate → quant] "12-month vol scenario: -15% to +35%"
[debate → bull] (round 2) acknowledge
[debate → bear] (round 2) push trim
[debate → moderator] verdict: "MODERATE TRIM: reduce tech 50% → 35%, add 10% gold proxy, 5% USDC"

[proposer] generates plan:
  - Sell 0.5 ETH
  - Sell 100 NVDA-stk
  - Buy 200 PAXG
  - Buy USDC delta
  Estimated cost: $1850 USD value moved, expected slippage 0.2%

[interrupt → user_confirm]
  UI shows: "I propose [plan]. Estimated $1850 of trades. Approve?"

User: "approve"

[execute → session_key.authorize ok]
  - swap 1: ETH -> USDC ($800)
  - swap 2: USDC -> PAXG ($600)
  - swap 3: NVDA-stk -> USDC ($450)
  All onchain executed via bundler
[verify] balances confirmed, deltas match plan ±0.1%
[memory.flush] saves: last_rebalance=today, holdings_after=...
END

Total: 6 LLM iter (research) + 4 LLM iter (debate) + 1 (proposer) + 1 (verify) = 12 calls
Cost: ~$0.35 LLM + ~$3 gas+slippage
Latency: ~90 seconds

---

六、生产经验与陷阱（整合层面）

1. 复杂度爆炸 12 LLM call 才完成一次 rebalance。简单场景可能不需要 debate。按价值分级：< $1k 跳过 debate；> $10k 强制 human review。

2. 状态膨胀 research findings + debate full transcript + proposal + memory 全塞 state，token 极高。措施：每个子图返回 summary，详细写文件。

3. Checkpoint 数据敏感 钱包地址、余额都进 checkpoint DB。需要加密 + 访问控制。

4. Interrupt 超时管理 用户离线 3 天，pending state 占资源。设 TTL（如 1h）+ 提醒，超时自动 cancel。

5. Memory 跨用户串扰 research findings 的 cache key 必须含 user_id。

6. Eval coverage 漏洞 有 unit eval（每个子图）+ integration eval（端到端）。后者贵，每天跑 1-3 个 representative case。

7. 回滚困难 onchain 不可逆，verify 失败只能"反向交易"补偿（仍然花 gas）。proposer 的 plan 要预留 unwind plan。

8. 依赖故障 MCP server / RPC / bundler 任何一个挂掉，整链路断。每个外部依赖加 timeout + circuit breaker。

---

七、监控指标

---

八、Web3 全集成清单

✅ Session keys (ERC-7715-style) ✅ Smart account (ERC-4337) ✅ Onchain simulate before execute ✅ Per-call + daily USD caps ✅ User interrupt for destructive ✅ Audit trail of every onchain action ✅ x402 ready for paid APIs（虽本 demo 没用） ✅ Bundler + paymaster 支持

---

九、剩余 Phase 3 内容预告（Day 163+）

后续会展开：

• Day 163-176：高级 RAG / hybrid search / GraphRAG / 长上下文工程 / context window 利用
• Day 177-190：可观测性 / LangSmith / Helicone / 自建 trace 平台
• Day 191-204：训练 / fine-tuning / RLHF / DPO / agent specialization
• Day 205+：生产化（multi-tenancy / cost FinOps / SLO / on-call）

---

十、面试题（整合 / Phase 3 中段总检）

Q1: 给一个 14 天前没接触过 agent 的工程师，让他设计一个金融 multi-agent 系统，推荐顺序？

A: ① 先 workflow 跑通（不是 agent）；② 升级为单 ReAct agent；③ 加 tools.py + 严格 schema；④ 加 memory；⑤ 加 eval；⑥ 引入 plan-execute（如果任务长）；⑦ 引入第二 agent（critic / reviewer）；⑧ 引入 LangGraph + checkpoint；⑨ 加 onchain（先 testnet + read only）；⑩ 加 session key + 主网（小金额）。每步加监控、回滚预案。永远不要一次性堆 5 种 pattern。

Q2: 一个 multi-agent 系统跑 3 个月后 cost 翻倍但 quality 没动，怎么诊断？

A: ① 先看 trace 分布——是不是某些 agent rounds 失控？② 看 model routing——是不是没用便宜模型该用的地方？③ 看 cache hit rate——降了还是没开？④ 看 tool call 重复率；⑤ 看 prompt 是否变长；⑥ 模型 provider 涨价？⑦ 用户分布变了（新用例更复杂）？⑧ 框架升级引入了额外 LLM call？⑨ Memory bloated triggers more retrieval。诊断后做 prioritized cost optimization plan。

Q3: 设计 onchain agent 的安全护栏，列出至少 8 项。

A: ① Session key 范围（合约白名单 + 方法白名单 + 金额上限 + 时间窗）；② 主钱包私钥在硬件钱包，不接触 runtime；③ Simulate before execute；④ 单笔 + 日累 + 任务 USD cap；⑤ 大额需用户二次确认（interrupt）；⑥ Audit log 全链路；⑦ 异常 revert 自动暂停；⑧ MEV 保护（Flashbots Protect 或 CoW）；⑨ Slippage min_out 强约束；⑩ Cross-chain 操作避免，单链 atomic 优先；⑪ Tool input schema 强校验；⑫ x402 amount cap；⑬ 紧急 kill switch（撤销 session key）。

Q4: 如何说服 CTO 上 multi-agent 系统而不是单 agent？

A: 不应该上来就推。流程：① 拿生产数据，证明单 agent 在某类任务的失败模式（confirmation bias / 单视角盲区）；② 算 ROI：multi-agent 多花 X，能避免 Y 类高价值错误；③ 跑 A/B 6 周数据；④ 显示 user satisfaction 提升 + cost 在容忍区；⑤ 提出降级 plan（如果数据不支持就回退）。多 agent 不是技术炫技，是数据驱动的产品决策。

Q5: Phase 3 中段你最大的 takeaway？

A: 三点：① "From simple to complex"——Anthropic 的建议是真理，工程师本能想上复杂方案，要克制；② Tools/memory/eval/observability 比 framework 选择重要；③ Onchain agent 的工程难度比想象高 —— security/compliance/UX 等额外成本（permissioning、simulation、interrupt、audit）让链下经验不能直接迁移。Phase 3 后半段会继续深化 RAG / 训练 / observability，但 Day 149-162 已经把"agent engineering"的基本骨架立起来了。

---

十一、Phase 3 中段（Day 121-162）能力盘点

复盘前一阶段 + 本周后，到 Day 162 你应该具备的能力：

LLM API 层（Day 121-148 收获）

• Anthropic SDK（Messages API、tool_use、streaming、batch）
• Prompt caching（5 分钟 TTL，省 90% 费用）
• Extended thinking + reasoning tokens
• Structured outputs（response_format）
• PDF / vision / files 输入
• 长上下文（1M context）
• Multi-provider routing

Agent 工程（Day 149-162 本段）

• BaseAgent 接口设计 + AgentTrace 一等公民
• ReAct 裸 SDK 实现
• Plan-Execute + Replanner
• Tool design 三原则（一事一 tool / strict schema / 错误分类）
• MCP server 部署（Claude Desktop / Code 联调）
• 双 agent 通信 + 终止条件
• LangGraph StateGraph + checkpoint + interrupt
• CrewAI / AutoGen / LangGraph 框架对比
• 4 层 memory（working/episodic/semantic/procedural）
• 三种多 agent 拓扑（sequential/hierarchical/network）
• Agent eval 套件（trajectory/step/end-to-end）
• Onchain agent（session key/x402/simulate）

综合

• 能在 1 小时内为新场景判断 workflow vs agent
• 能写完整 trace 并 reason about cost/latency
• 能 review 别人的 agent 代码 + 指出反模式
• 能给 PM 解释清楚每个抽象层的 trade-off

---

十二、本周给职业的"硬产出"清单

求职时这 14 天能直接放简历/作品集的内容：

简历加分：能展示"既写过 LangGraph 又写过裸 SDK 又用过 CrewAI/AutoGen"——这是框架无关的能力，比"只会某一个"值钱得多。

---

十三、Phase 3 后半段路线图（Day 163-220）

到 Day 220 应该能拿出"完整的 production-ready agent platform"作品集，足以申请 staff/principal 级 AI engineer / AI architect。

---

十四、自我反思 prompts（写日志用）

每周用这些问题反问自己：

1. 本周写的代码我能在白板 30 分钟内默写其中一个吗？
2. 如果让我代表团队 review 一个外部 multi-agent 项目，我能指出至少 3 个问题吗？
3. 我能用 5 句话说服 CTO："我们应该上 agent / 不应该上 agent"吗？
4. 我能解释 prompt caching / interrupt / session key / MCP 给非技术 PM 吗？
5. 如果生产突然 cost 翻倍，我能在 1 小时内列出 5 个排查方向吗？
6. 我对哪个概念还模糊？（找出来下周加深）

---

十五、本系统的扩展路线

v1 完成后能想到的下一步：

v1.1（短期，1-2 周）

• 接入真实 MCP server（Day 153 的 fin-mcp + 自建 onchain-mcp）
• LangSmith tracing
• 简单 web UI（Streamlit）展示 plan + interrupt confirm

v1.5（1 个月）

• Multi-tenant（多用户隔离）
• 接 Postgres checkpointer（替代 SQLite）
• 加 cost dashboard（Helicone 或自建）
• Eval 自动跑（CI 集成）
• 接 EIP-7702 让 EOA 也能用 session key

v2（季度级）

• 加 RAG（监管文件 + 内部研报）
• Voice interface（用户语音说"重平衡"）
• 多链支持（Base + Solana）
• Agent 之间的 x402 互付（research agent 调付费数据 API 自动结算）
• Onchain audit log（关键决策 hash 上链）

v3（半年）

• 可微调（用 user 的反馈 SFT/DPO 一个 specialized agent）
• Plugin 市场（业务团队可贡献 sub-graph）
• 监管报送自动化（与监管 API 直接集成）
• ZK proof of reserves（证明 agent 没违规调用合约）

---

十六、本系统的反例——什么时候不该用

这套 multi-agent v1 不是万金油。明确不应用的场景：

---

十七、面试 take-home 模拟题

假设面试要求 3 天完成一个 take-home：

题目："设计一个 onchain investment research + execution agent，给 retail 用户用。给一个架构设计 + 关键代码 + 风险分析。"

答题骨架：

1. 范围与 trade-off（300 字）：选 ReAct 还是 multi-agent？为什么这个 user 量级用 LangGraph？
2. 架构图（C4 风格）：包含 client、agent、MCP、链上三层
3. 关键代码（500 行）：选 1-2 个核心节点写完整
4. 安全模型：session key scope / interrupt / kill switch / audit log
5. eval 计划：内部 eval set 如何建、CI 流程
6. 运营：cost monitoring、failure handling、on-call
7. 演进路线：MVP → v1 → v2

本周（Day 149-162）的内容正好是这个 take-home 的素材库。每个章节都对应一个答题模块。

---

十八、本系统能解决什么真实痛点

为什么这个 v1 系统是有商业价值的（而非纯练习）：

这是金融科技的真实创业方向之一（已有 Robo-advisor 2.0 / On-chain Brokerage 类项目在做）。

---

十九、Phase 3 中段总检（Self-quiz）

下面 10 题如果都能答出来，Day 149-162 内容你掌握了：

1. 默写 ReAct loop 核心 50 行代码
2. 默写 Plan-Execute 三阶段（planner / executor / synthesizer）的接口
3. 写一个 tool 的完整 JSON schema（含 enum / pattern / required）
4. 解释 MCP 三类原语 + 何时用哪种
5. 给一个金融场景，决定单 agent / 双 agent / hierarchical
6. 写 LangGraph state TypedDict 含 3 种 reducer
7. 设计一个 4 层 memory 的写入/读取流程
8. 列 onchain agent 8 项安全护栏
9. 设计一个 eval 套件（5 cases + 5 judges + 报告结构）
10. 把以上整合成一张 multi-agent v1 架构图（10 分钟内）

每题不会答的，回去补对应 Day。

---

二十、本阶段（Day 149-162）的"硬币两面"

学到了什么（正面）

• 14 天系统化建立了 agent 工程能力
• 9 个核心 Python 文件 ≈ 一个完整 stack
• 能在面试中讲清楚 agent vs workflow vs chatbot
• 能为新场景做设计选型

没学到什么（盲区，下阶段补）

• 高级 RAG（Day 163-176 补）
• 生产级 observability（Day 177-190 补）
• Fine-tune / 训练（Day 191-204 补）
• Multi-tenant / FinOps / on-call（Day 205+ 补）
• LLM 内部机制（attention / KV cache 优化等）
• 安全 / red team（专题）

知道盲区比假装全懂更重要。Phase 3 后半段会按上述顺序补齐。

---

明日预告

Day 163: 高级 RAG 技术——Hybrid search / Reranking / HyDE / GraphRAG

• Phase 3 进入"信息检索 + 长上下文工程"主题
• 经典 RAG 的局限
• 现代 hybrid + rerank 流水线
• GraphRAG (Microsoft 2024) 的 entity graph 思路