Agent 编排框架实战

在多 Agent 系统中，选择合适的编排框架至关重要。本章对比主流框架并通过实战演示如何构建可靠的多 Agent 编排系统。

编排框架生态

框架对比

LangGraph 风格编排

LangGraph 使用有向图建模 Agent 流程，每个节点是一个处理步骤，边定义转换条件。

"""
LangGraph 风格的状态图编排实现
"""
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Callable
class NodeStatus(Enum):
PENDING = "pending"
RUNNING = "running"
COMPLETED = "completed"
FAILED = "failed"
@dataclass
class GraphState:
"""图执行状态"""
data: dict = field(default_factory=dict)
current_node: str = ""
history: list[str] = field(default_factory=list)
errors: list[str] = field(default_factory=list)
def set(self, key: str, value: Any) -> None:
self.data[key] = value
def get(self, key: str, default: Any = None) -> Any:
return self.data.get(key, default)
@dataclass
class Edge:
"""图的边"""
source: str
target: str
condition: Callable[[GraphState], bool] | None = None
class AgentGraph:
"""Agent 编排图"""
def __init__(self, name: str):
self.name = name
self.nodes: dict[str, Callable[[GraphState], GraphState]] = {}
self.edges: list[Edge] = []
self.entry_point: str = ""
def add_node(self, name: str, func: Callable[[GraphState], GraphState]) -> "AgentGraph":
"""添加节点"""
self.nodes[name] = func
return self
def add_edge(self, source: str, target: str) -> "AgentGraph":
"""添加无条件边"""
self.edges.append(Edge(source=source, target=target))
return self
def add_conditional_edge(
self,
source: str,
condition: Callable[[GraphState], bool],
true_target: str,
false_target: str,
) -> "AgentGraph":
"""添加条件边"""
self.edges.append(Edge(source, true_target, condition))
self.edges.append(
Edge(source, false_target, lambda s, c=condition: not c(s))
)
return self
def set_entry(self, node: str) -> "AgentGraph":
self.entry_point = node
return self
def run(self, initial_state: GraphState | None = None) -> GraphState:
"""执行图"""
state = initial_state or GraphState()
state.current_node = self.entry_point
max_steps = 50
for step in range(max_steps):
node_name = state.current_node
if node_name not in self.nodes:
break
print(f"  [{step+1}] 执行节点: {node_name}")
state.history.append(node_name)
try:
func = self.nodes[node_name]
state = func(state)
except Exception as e:
state.errors.append(f"{node_name}: {e}")
break
# 查找下一个节点
next_node = self._find_next(node_name, state)
if next_node is None:
break
state.current_node = next_node
return state
def _find_next(self, current: str, state: GraphState) -> str | None:
"""查找下一个节点"""
for edge in self.edges:
if edge.source != current:
continue
if edge.condition is None or edge.condition(state):
return edge.target
return None
# 使用示例：研究 Agent 编排
def research_node(state: GraphState) -> GraphState:
query = state.get("query", "")
state.set("research_result", f"[研究结果] 关于 '{query}' 的资料...")
return state
def analyze_node(state: GraphState) -> GraphState:
research = state.get("research_result", "")
state.set("analysis", f"[分析] 基于 {research[:30]}... 的分析结论")
return state
def review_node(state: GraphState) -> GraphState:
analysis = state.get("analysis", "")
quality_ok = len(analysis) > 20
state.set("quality_passed", quality_ok)
return state
def output_node(state: GraphState) -> GraphState:
state.set("final_report", "最终报告已生成")
return state
# 构建图
graph = (
AgentGraph("research-pipeline")
.add_node("research", research_node)
.add_node("analyze", analyze_node)
.add_node("review", review_node)
.add_node("output", output_node)
.set_entry("research")
.add_edge("research", "analyze")
.add_edge("analyze", "review")
.add_conditional_edge(
"review",
condition=lambda s: s.get("quality_passed", False),
true_target="output",
false_target="research",  # 质量不合格则重新研究
)
)

CrewAI 风格角色编排

CrewAI 以角色（Role）和任务（Task）为核心抽象，更接近真实团队协作。

"""
CrewAI 风格角色编排
"""
from dataclasses import dataclass, field
@dataclass
class Role:
"""Agent 角色"""
name: str
goal: str
backstory: str
tools: list[str] = field(default_factory=list)
@dataclass
class Task:
"""任务定义"""
description: str
assigned_to: str  # Role name
expected_output: str
context_from: list[str] = field(default_factory=list)  # 依赖的前置任务
@dataclass
class TaskResult:
"""任务结果"""
task_description: str
output: str
agent: str
class Crew:
"""团队编排器"""
def __init__(self, name: str):
self.name = name
self.roles: dict[str, Role] = {}
self.tasks: list[Task] = []
def add_role(self, role: Role) -> "Crew":
self.roles[role.name] = role
return self
def add_task(self, task: Task) -> "Crew":
self.tasks.append(task)
return self
def kickoff(self, initial_context: str = "") -> list[TaskResult]:
"""按顺序执行所有任务"""
results: list[TaskResult] = []
context_map: dict[str, str] = {"initial": initial_context}
for task in self.tasks:
role = self.roles.get(task.assigned_to)
if role is None:
continue
# 收集上下文
context_parts = [
context_map[dep]
for dep in task.context_from
if dep in context_map
]
context = "\n".join(context_parts) if context_parts else initial_context
print(f"  [{role.name}] 执行: {task.description[:50]}...")
output = self._execute_task(role, task, context)
result = TaskResult(
task_description=task.description,
output=output,
agent=role.name,
)
results.append(result)
context_map[task.description] = output
return results
def _execute_task(self, role: Role, task: Task, context: str) -> str:
"""执行单个任务（模拟 LLM 调用）"""
return (
f"[{role.name}] 完成任务: {task.description}\n"
f"上下文长度: {len(context)} 字符\n"
f"预期输出: {task.expected_output}"
)

编排模式选择决策树

本章小结

下一章：自我反思与改进