根据工作负载特征选择代理框架:六个维度与决策路径
根据工作负载特点选择代理框架:LangGraph、CrewAI、AutoGen、Temporal或无需使用,其选择标准为确定性、生命周期、幂等性、可审计性以及成本。
来源: Your AI Agent Framework Is Probably the Wrong One Here’s How to Actually Choose
询问该使用哪种代理框架是个错误的开场问题。更好的问题是工作负载的具体形态,因为框架的选择应基于这个答案,而非仅凭基准测试文章来决定。那些在第一周就根据社交媒体上的讨论来选择框架的团队,往往要在第三到第六个月间才发现该框架的核心假设与他们实际要构建的系统并不匹配。问题很少出在框架本身,因为在选择旨在满足某些需求的工具之前,根本没人明确说明工作负载究竟需要哪些特性。
这种方法不是比较各种功能,而是确定决定适配性的六个维度,展示每个维度在代码中的表现形式,进而构建决策路径。有一个实际运行的示例:基础设施警报分类代理。它很朴实、贴近现实,涉及所有这些维度。该代理需要判断警报是否为真正的问题,通过通知相关人员或自动修复来采取行动,有时需要等待数小时才能得到人工回复,并且绝不会因为在凌晨4点重复尝试就两次通知值班工程师。
第一个维度:分支决策的确定性要求有多高?
这个维度暂时跳过,稍后再详细讨论。真正的测试标准是:如果相同输入有2%的概率走不同的代码路径,是否会造成任何问题。有些分支能够容忍这种差异,比如选择总结邮件的语气风格。
# A branch where non-determinism is FINE — picking a tone for a summary email.
# If the agent occasionally phrases things slightly differently, nobody's paged.
def draft_summary_tone(context: dict) -> str:
return llm_call(
prompt=f"Summarize this incident in a {context['audience']}-appropriate tone.",
temperature=0.7, # variability here is a feature, not a bug
)
其他情况则不然,比如决定是否在凌晨4点唤醒人类或自动进行修复。这类决策必须由代码来决定,而非通过提示语。
# A branch where non-determinism is NOT fine — deciding whether to page a human
# at 4am versus auto-remediating. This must be code, not a prompt.
def route_alert(alert: dict) -> str:
if alert["severity"] == "critical" and alert["service"] in PAGE_ALWAYS_SERVICES:
return "page_oncall"
if alert["auto_remediation_available"] and alert["confidence"] > 0.9:
return "auto_remediate"
if alert["severity"] == "critical":
return "page_oncall"
return "log_and_monitor"
当工作流中的关键决策表现为第二种函数形式——具有确定性、可审计性,且符合合规审查人员希望以纯代码形式查看的要求时,框架必须将此类逻辑与大型语言模型隔离开来。LangGraph的条件边功能非常适合这一需求,因为路由功能是普通的Python代码,模型仅会处理那些被刻意传递给它的部分,比如分类或草拟工作。
CrewAI和AutoGen的构建方式则相反,由智能体自身的推理决定下一步该做什么。当目标是实现创意灵活性时,这是优势;但若灵活性偶尔意味着跳过合规检查,那就成了弊端。如果大多数关键分支都必须是确定性的,那么仅这一要求就足以让该领域的大部分方案被淘汰。
第二个维度:一个工作单元的存续时间有多长?
一个短生命周期的单元在单个HTTP请求中启动并结束。在这个阶段无需任何框架,否则只会增加不必要的开销。
# Short-lived: starts and finishes inside one HTTP request.
# This is the "no framework needed" zone — a framework here is pure overhead.
async def handle_summarize_request(request: SummarizeRequest) -> SummarizeResponse:
text = await fetch_document(request.doc_id)
summary = await llm_summarize(text, max_tokens=300)
return SummarizeResponse(summary=summary)
长期运行的单元则有所不同。某个警报可能会在值班工程师睡觉时等待六小时才得到人工确认,而在后续的部署替换了其底层的容器后,它会通过另一个流程继续处理。
# Long-lived: this alert might sit in "awaiting human ack" for six hours
# while the on-call engineer is asleep, then resume on a completely
# different process after a deploy rotated the pods underneath it.
class AlertTriageWorkflow:
async def run(self, alert: dict) -> dict:
decision = await self.classify_and_route(alert)
if decision == "page_oncall":
await self.page(alert)
await self.wait_for_ack(timeout_hours=1) # this line is the whole ballgame
...
如果某项任务确实能在几秒内完成且完全存在于单个进程的内存中,那么考虑使用框架或许有些过度了。一个带有重试装饰器的普通异步函数在可维护性方面就优于任何框架,因为无需进行任何协调工作;而学习 LangGraph 的状态模型或为无状态摘要接口运行 Temporal 集群,都是成本高昂却毫无回报的做法。
一旦某项任务需要持续运行超过一个进程的生命周期,因为它要等待人工处理、批量作业完成或另一团队的服务等级协议要求,那么以小时计时的持久化执行就变得必不可少。目前流行的三种大语言模型编排框架都不是为充当持久化层而设计的。这种情况下,更适合选择类似 Temporal 的工具,或者至少构建一个包含持久化状态管理以及可恢复步骤的任务队列的手工系统。
第三个维度:如果某个步骤被执行了两次怎么办?
团队通常在事故回顾中发现这个问题。下方的第一个版本在演示中看起来没有问题,但实际上隐患已存在,因为重新尝试该操作会再次启动服务,从而引发事故。
# BEFORE — looks fine in a demo, is a live incident waiting to happen
async def auto_remediate(alert: dict):
await restart_service(alert["service"]) # what if this activity gets retried?
第二个版本按设计就是幂等的。
# AFTER — idempotent by construction
async def auto_remediate(alert: dict, idempotency_key: str):
if await remediation_ledger.already_applied(idempotency_key):
logger.info("remediation already applied, skipping", key=idempotency_key)
return await remediation_ledger.get_result(idempotency_key)
result = await restart_service(alert["service"])
await remediation_ledger.record(idempotency_key, result)
return result
幂等键必须来自业务数据中稳定的元素,比如警报ID,绝不能来自每次尝试时生成的UUID,否则重试将不再安全。这属于工具功能而非任何框架的功能,但它会影响哪些框架是可用的,因为有些框架容易让人忘记已经进行了重试。AutoGen的基于聊天的执行方式无法提供结构化信号来表明工具调用是重试还是新的决策。从对话记录来看,智能体只是简单地决定再次调用工具,而这是重试还是重复操作则无法被追踪。
当副作用成本高昂、具有危险性,或涉及资金或基础设施变更时,就需要一个框架,或是在框架之下自行构建的层,将副作用是否已经发生视为一个核心问题,而非从聊天记录中推断出来的内容。
第四个维度:谁会在之后阅读决策内容,以及如何阅读?
无论是由什么机制推动了相关步骤,事后分析中重要的是与具体框架无关的审计记录。
# A framework-agnostic audit record — this is what actually matters
# in a postmortem, regardless of what orchestrated the steps.
@dataclass
class DecisionRecord:
alert_id: str
timestamp: float
step: str
reasoning: str # what the LLM said, verbatim
decision: str # the structured outcome, not prose
confidence: float | None
human_override: bool
async def log_decision(record: DecisionRecord):
await audit_store.insert(record)
# Also emit as a structured log line — cheap insurance for when
# the audit store itself is the thing that's down during an incident.
logger.info("agent_decision", **asdict(record))
每个框架都支持记录某些内容。问题在于,其结构是否能自然生成清晰、可查询的决策轨迹,还是实现这一目标需要克服抽象层的限制。LangGraph的状态对象通过命名节点传递,因此要查询路由决策之前的状态,只需直接查询从未被转换为自由文本的TypedDict即可。而AutoGen的对话内容则必须经过解析才能重建相同的事实,因为它的原生存储格式是散文式文本,而非结构化状态。在那些有合规要求、设有审计团队或习惯进行彻底事后分析的组织中,这一因素的重要性应高于营销材料所体现的程度。
第五个维度:什么限制了团队的工作效率?
这种因素很少出现在技术对比图表中,却常常能决定最终结果。一家仅有四名成员的初创公司,在尚未编写任何业务逻辑之前,就不应花费三周时间去研究Temporal的确定性约束条件。
# Week-one prototype: prove the concept fast, accept the debt knowingly.
from crewai import Agent, Task, Crew
triage_agent = Agent(role="Alert Triage", goal="Decide how to handle infra alerts")
crew = Crew(agents=[triage_agent], tasks=[Task(description="Triage: {alert}", agent=triage_agent)])
crew.kickoff(inputs={"alert": alert_payload})
那完全是第一周就可能出现的问题。错误并不在于没有为原型选择CrewAI,而在于六个月后,当真正的值班任务和基础设施变更真正出现时,团队却再也没有重新考虑过这个选择,因为原型已经能够正常运行了。框架的选择存在有效期,这个有效期取决于系统中的使用频率,而在选择时将这个日期记录下来就能避免其被默认设置。
第六个维度:每次决策的延迟和成本预算是多少?
每一个需要通过LLM来处理的路由决策,都会在多智能体之间的多轮对话中被反复使用;再加上每天数百次的警报处理,以及在事故发生时数千次的突发请求,费用会迅速上升。
# Expensive pattern: every routing decision is its own LLM call,
# multiplied across a multi-agent conversation with several turns.
# At alert volumes (hundreds/day, sometimes bursts of thousands during
# a real incident), this is a real line item, not a rounding error.
async def route_via_llm(alert: dict) -> str:
return await llm_call(f"How should we handle this alert? {alert}")
# Cheaper, faster, and more auditable: cheap deterministic pre-filtering
# in code, LLM reserved for genuinely ambiguous cases.
async def route_alert_efficiently(alert: dict) -> str:
if alert["service"] in KNOWN_NOISY_SERVICES and alert["severity"] == "low":
return "log_and_monitor" # zero LLM calls for the common case
if alert["signature"] in KNOWN_REMEDIATION_PLAYBOOK:
return "auto_remediate" # deterministic lookup, zero LLM calls
return await llm_call(f"Novel alert, needs judgment: {alert}") # LLM only when genuinely needed
这与其说是一种框架轴,不如说是一门能辅助各类系统的架构学科。那些能让低成本、确定性强的代码与高成本模型调用轻松结合的框架,能够在大规模应用中节省大量成本。LangGraph具备此功能,将LLM调用封装为单一任务的流程引擎如Temporal也是如此。另一种方式则是通过另一个模型中介的对话轮次来处理所有内容,这是当speaker_selection_method设置为auto时AutoGen的默认做法。
决策路径,而非决策树
若用流程图来表示只产生一个答案的情况,那是不准确的,因为不同区域存在重叠。下图更贴近实际决策的生成过程。
Is this unit of work stateless and finishes in seconds?
└─ YES → skip the framework entirely. Plain functions + retries. Ship it.
└─ NO, continue.
Does it need to survive process restarts / wait on humans for hours-to-days?
└─ YES → you need durable execution (Temporal or equivalent) as the backbone,
regardless of what else you pick for the reasoning layer.
└─ NO, continue.
Are the important branches safety- or compliance-critical
(money, infra changes, irreversible external actions)?
└─ YES → LangGraph-style explicit graphs, keep LLM scoped to narrow nodes.
└─ NO, mostly exploratory/creative → CrewAI or AutoGen are legitimate defaults.
Is this still a prototype whose findings might get thrown away?
└─ YES → optimize for speed of iteration over long-term correctness,
but write down when you'll revisit that tradeoff.
将该方案应用于警报分类代理后,便形成了分层设计结构。由于该系统需在重启后仍能正常运行并等待人工处理,因此需要一个稳定的执行层。其中最关键的部分涉及安全保障,即选择分页通知还是自动修复,因此在每个处理步骤中都需要类似LangGraph的明确路由机制。一旦具备了分页功能,它就已远超出原型阶段。此时时间维度成为核心框架,而范围限定的大语言模型协调步骤则作为其内部的一个处理活动存在。
@activity.defn
async def classify_alert_activity(alert: dict) -> dict:
# LangGraph-style graph runs here — bounded reasoning, deterministic routing —
# inside an activity Temporal will retry and time-box like any other side effect.
result = alert_triage_graph.invoke({"alert": alert, "audit_log": []})
return {"decision": result["decision"], "confidence": result["confidence"]}
@workflow.defn
class AlertTriageWorkflow:
def __init__(self):
self._acked = False
@workflow.signal
async def acknowledge(self):
self._acked = True
@workflow.run
async def run(self, alert: dict) -> dict:
classification = await workflow.execute_activity(
classify_alert_activity, alert,
start_to_close_timeout=timedelta(seconds=20),
retry_policy=workflow.RetryPolicy(maximum_attempts=3),
)
if classification["decision"] == "page_oncall":
await workflow.execute_activity(page_oncall, alert, start_to_close_timeout=timedelta(seconds=10))
await workflow.wait_condition(lambda: self._acked, timeout=timedelta(hours=1))
if not self._acked:
await workflow.execute_activity(escalate_to_secondary, alert, start_to_close_timeout=timedelta(seconds=10))
elif classification["decision"] == "auto_remediate":
await workflow.execute_activity(
auto_remediate, alert, f"remediate-{alert['id']}",
start_to_close_timeout=timedelta(minutes=2),
retry_policy=workflow.RetryPolicy(maximum_attempts=2),
)
return {"alert_id": alert["id"], "decision": classification["decision"]}
实际选择框架的过程并非如此简单,也不存在令人满意的一句答案。这是个组合决策问题,而具备这种组合能力才是关键技能。
屡屡出现的错误
- 仅凭 GitHub 星标来选择。 流行度只能反映社区支持程度以及网上现有解决方案的数量,却无法说明其核心执行模型是否符合工作负载对稳定性与确定性的要求。
- 让原型框架直接用于生产环境。 那个暂时能正常运行的原型会直接变成正式的生产系统,因为人们觉得重写调度逻辑并不会增加新功能,但实际上它提升了可靠性,而这种可靠性在出现问题之前往往是看不见的。
真正的答案
并不存在所谓的最佳代理框架,正是人们误以为有最佳框架,才导致这么多系统在18个月后就需要重新设计。对于具有特定六项属性的工作负载,才存在最适合的框架。正确的做法是在团队开始使用任何框架之前先明确这些属性,一旦完成这一步,通常就能自行选出合适的框架。如果跳过这一步,无论选择了哪种框架,调度层都将在一年内被重写,因为问题从来不在框架本身,而在于在决定构建方式之前没有先明确要构建什么。
相关阅读
- 利用DSPy优化Terraform Agent:先构建验证器再优化 — Terraform Agent会自动生成架构。应先构建一个确定性的验证器作为基准,再使用DSPy和GEPA来优化其技能文件,从而获得准确的结果。
- LangChain深度智能体的流式处理:实时计划、工具调用与令牌 — 用流式处理替代阻塞式的调用方式,让用户能够看到深度智能体的工作过程:包括三种流式模式、子智能体的命名空间以及令牌的过滤与汇总功能。