你的 CrewAI 流水线里有三个 Agent:CodeReviewer、SecurityAuditor 和 DocWriter。结果一个「检查 SQL 注入漏洞」的任务被路由给了 DocWriter,输出了一段关于如何写 SQL 注入文档的内容,完全不是你想要的。或者在 AutoGen 里,编排器 Agent 连续把任务转发了 5 次,每次被接收的 Agent 都说「这不在我的能力范围内」,任务最终超时。路由失效在基于 LLM 决策的动态编排系统里尤其常见——模型对任务描述的理解和对 Agent 能力描述的匹配有时会出人意料地错误。
常见原因
1. Agent 的角色描述过于模糊或重叠
多个 Agent 的 role 或 description 在语义上高度相似,编排器的 LLM 路由器无法区分。例如「代码助手」和「编程专家」都可能被理解为适合处理代码任务,路由结果取决于 token 采样的随机性。
怎么判断:把所有 Agent 的描述并排列出,问一个中立的 LLM「哪个 Agent 适合处理 X 任务」,如果答案不一致或有歧义,描述就需要收窄。
2. 路由器没有使用结构化的能力声明
编排器 Agent 自己是一个 LLM,它根据系统提示里的自由文本描述来决定路由。这种「LLM 路由 LLM」的链路引入了语义歧义,任务描述里的微小措辞变化就可能导致完全不同的路由结果。
怎么判断:用相同的任务,改变描述里的一两个词,多次运行编排器,检查路由结果是否稳定。如果同一个任务每次路由到不同的 Agent,就是路由器不稳定。
3. 新增 Agent 后没有更新路由规则
流水线最初有 2 个 Agent,后来加了第 3 个 Agent 负责安全审计。但编排器的 system prompt 里只描述了前 2 个 Agent,新 Agent 对路由器来说是「不可见的」,安全相关任务继续被路由到原有的 Agent。
怎么判断:检查路由器的 system prompt 或路由规则配置,列出它「知道」的 Agent 列表,与实际运行的 Agent 数量对比。
4. 任务描述缺少明确的分类标签
「帮我看看这段代码」这种任务描述没有指定是性能分析、安全审计还是格式检查,路由器只能猜测。猜错的概率随 Agent 数量增加而增大。
怎么判断:统计路由失败的任务描述,检查它们是否都缺少明确的操作类型或领域标签。
5. 路由使用了语义相似度而不是能力匹配
有些框架用 embedding 相似度来路由:计算任务描述与各 Agent 描述的余弦相似度,选最高的。但语义相似度高不等于能力匹配——「写一份 SQL 注入报告」的 embedding 可能更接近「文档编写」Agent 而不是「安全审计」Agent。
怎么判断:打印路由时的相似度得分,检查最终选择的 Agent 是否真的是能力最匹配的。
6. 工具可用性没有纳入路由决策
任务需要调用特定工具(如代码执行沙箱、漏洞扫描器),但路由器只根据语义匹配,没有检查目标 Agent 是否拥有所需工具。结果任务被路由给了「看起来合适」但缺少必要工具的 Agent。
怎么判断:列出任务所需的工具列表,与被选中 Agent 的 tools 列表对比,检查是否有缺失。
最短修复路径
Step 1:用结构化的能力 schema 替代自由文本描述
from pydantic import BaseModel
from enum import Enum
class TaskDomain(str, Enum):
CODE_REVIEW = "code_review"
SECURITY_AUDIT = "security_audit"
DOCUMENTATION = "documentation"
DATA_ANALYSIS = "data_analysis"
class AgentCapability(BaseModel):
agent_id: str
domains: list[TaskDomain] # 严格的领域标签
required_tools: list[str] # 必须拥有的工具
max_task_complexity: int # 1-5 复杂度上限
description: str # 仅供人类阅读
AGENT_REGISTRY = [
AgentCapability(
agent_id="security_auditor",
domains=[TaskDomain.SECURITY_AUDIT],
required_tools=["code_scanner", "cve_lookup"],
max_task_complexity=5,
description="专注于安全漏洞分析"
),
AgentCapability(
agent_id="doc_writer",
domains=[TaskDomain.DOCUMENTATION],
required_tools=["web_search"],
max_task_complexity=3,
description="技术文档撰写"
),
]
Step 2:实现基于规则的确定性路由
def route_task(task: dict, registry: list[AgentCapability]) -> str:
"""基于规则的确定性路由,不依赖 LLM 决策。"""
required_domain = TaskDomain(task["domain"]) # 任务必须声明领域
required_tools = set(task.get("required_tools", []))
candidates = [
a for a in registry
if required_domain in a.domains
and required_tools.issubset(set(a.required_tools))
and task.get("complexity", 1) <= a.max_task_complexity
]
if not candidates:
raise RoutingError(f"没有 Agent 匹配任务领域 '{required_domain}' + 工具 {required_tools}")
# 按复杂度上限升序选最精准的 Agent(避免过度分配)
return sorted(candidates, key=lambda a: a.max_task_complexity)[0].agent_id
Step 3:在任务创建时强制声明领域标签
# 不允许创建没有 domain 标签的任务
class Task(BaseModel):
description: str
domain: TaskDomain # 必填
required_tools: list[str] = []
complexity: int = 1 # 1-5
# 禁止空 description
@field_validator("description")
def desc_not_empty(cls, v: str) -> str:
assert len(v.strip()) >= 10, "任务描述至少 10 个字符"
return v
Step 4:为路由逻辑写单元测试
# tests/test_routing.py
def test_sql_injection_routes_to_security():
task = Task(
description="检查这段代码是否存在 SQL 注入漏洞",
domain=TaskDomain.SECURITY_AUDIT,
required_tools=["code_scanner"]
)
assert route_task(task.dict(), AGENT_REGISTRY) == "security_auditor"
def test_routing_fails_on_missing_tool():
task = Task(
description="扫描 CVE 漏洞",
domain=TaskDomain.SECURITY_AUDIT,
required_tools=["cve_lookup", "nonexistent_tool"]
)
with pytest.raises(RoutingError):
route_task(task.dict(), AGENT_REGISTRY)
Step 5:添加路由失败告警和回退
def route_with_fallback(task: dict) -> str:
try:
return route_task(task, AGENT_REGISTRY)
except RoutingError as e:
# 路由失败时发告警,不要静默失败
alert_ops(f"路由失败:{e},任务 ID:{task['id']}")
# 回退到通用 Agent(能力最广但质量一般)
return "general_agent"
预防建议
- 用枚举类型定义任务领域,所有任务创建时必须声明
domain,不允许自由文本路由 - 为每个新 Agent 写路由测试用例,PR 合并前必须通过
- 定期统计实际路由结果与期望路由的偏差率,超过 5% 时触发路由规则审查
- 在 Agent 描述里明确写「不处理哪类任务」,帮助路由器做排除决策
- 当 Agent 收到超出能力范围的任务时,应该拒绝(返回错误)而不是尝试完成,防止输出低质量结果
- 避免用 LLM 做路由决策,改用规则引擎或基于工具可用性的确定性匹配
- 新增 Agent 时,必须同步更新路由规则和 registry,并在 CI 里验证所有现有路由测试仍然通过
常见问答 (FAQ)
Q: CrewAI 默认的路由机制是什么,如何覆盖它?
A: CrewAI 的 Crew 在 Process.hierarchical 模式下,由 manager LLM 根据各 Agent 的 role 和 goal 描述来分配任务,本质是 LLM 决策。可以改用 Process.sequential 按顺序执行(完全确定性),或者自定义 Task.agent 直接指定处理 Agent,绕过动态路由。
Q: 如果不得不用 LLM 路由,如何提高准确率? A: 有三个方法:1)在路由提示里提供 few-shot 示例,覆盖容易出错的边界案例;2)让路由 LLM 输出结构化 JSON 而不是自由文本,强制它从 Agent ID 枚举里选择;3)用温度 0 运行路由 LLM,减少随机性。
Q: 任务已经路由到错误 Agent 并执行了一半,如何中止并重路由?
A: 在编排器里拦截 Agent 的 on_error 回调(大多数框架支持),检查错误类型是否是「能力不匹配」(如 ToolNotFoundError、OutOfScopeError),如果是则取消当前 Agent 的任务,重新路由给正确的 Agent。注意这时任务的执行状态需要从头开始,之前的部分输出通常不可复用。
Q: 在生产环境如何快速发现路由错误? A: 在每次路由决策时记录「任务 ID + 选中 Agent + 路由依据」的结构化日志,然后在任务完成时记录人工或自动的质量评分。通过 Grafana 或 Datadog 把「低质量任务的路由结果」可视化,就能发现系统性的路由错误模式。