第 11 章 StrandsMulti-AgentGraph
第 11 章:多 Agent 模式:Graph 工作流
第 11 章:多 Agent 模式:Graph 工作流
关于本课
本课程的视频是某个时间点的快照。Strands 一直在活跃开发中,所以本页展示的代码反映的是最新的模式,但视频里讲的概念依然适用。拿不准的时候,以代码为准。
本课代码:samples/11-graphs
对执行过程的确定性控制
Graph 和 workflow 让你能确定性地控制多 Agent 的执行过程。agents-as-tools 是由编排器动态决定流程,而这些模式不一样:执行顺序、依赖关系、并行分支都在事前定好。只要你能在白板上把工作流画出来,就该用它们。
Graph
每个节点都是一个完整的 Agent,边表示依赖关系。当一个节点的所有入边都满足时它才执行,哪些并行、哪些等待由 Graph 自己解析。
from strands import Agent
from strands.multiagent import GraphBuilder
from strands.vended_tools.web_fetch import web_fetch
researcher = Agent(
name="researcher",
system_prompt="Gather comprehensive information from the web.",
tools=[web_fetch],
)
analyst = Agent(
name="analyst",
system_prompt="Identify patterns, trends, and key insights from research.",
)
summarizer = Agent(
name="summarizer",
system_prompt="Condense raw research into concise key points.",
)
report_writer = Agent(
name="report_writer",
system_prompt="Synthesize analysis and summaries into a final report.",
)
builder = GraphBuilder()
builder.add_node(researcher, "research")
builder.add_node(analyst, "analysis")
builder.add_node(summarizer, "summarize")
builder.add_node(report_writer, "report")
builder.add_edge("research", "analysis")
builder.add_edge("research", "summarize") # analyst + summarizer run in parallel
builder.add_edge("analysis", "report")
builder.add_edge("summarize", "report") # report waits for both
builder.set_execution_timeout(600)
graph = builder.build()
result = graph("Research the impact of AI on healthcare")
数据流转:入口节点收到原始任务。下游节点收到原始任务,再加上来自各自依赖的带标签输出。用 invocation_state 传元数据(用户 ID、feature flag),这些信息不该暴露给模型。
常见形态:顺序流水线、并行扇出(如上例)、条件分支、循环反馈。如果你构造了环,要设置 set_max_node_executions,否则它会一直转下去。
核心概念
- 节点(Nodes) 是 Agent(也可以是任意可调用对象)。所有入边满足时它们才执行。
- 边(Edges) 定义依赖。从 A 到 B 的边意味着 B 要等 A。
- 并行扇出(Parallel fan-out):彼此没有依赖的节点并发执行。
- 上下文传递:每一步都会收到前驱节点的输出。
- 适用场景:结构已知的流水线、fan-out/fan-in、需要保证执行顺序的工作流,以及需要审计轨迹的流程。
对比
| 模式 | 流程控制 | 最适合 |
|---|---|---|
| Agents as Tools | 编排器决定 | 可分离的领域、综合汇总 |
| Graphs | 由边定义 | 已知流水线、并行 |
| Swarms | Agent 决定 | 未知序列、探索 |