在107个真实数据工程任务中,LangGraph通过率97/107、延迟13.8s、人均消耗token 2350,均大幅领先CrewAI和AutoGen,后两者硬失败分别达24和49个。
Subtitle: Cut through the marketing: see where LangGraph, CrewAI, and AutoGen fail, succeed, and waste your tokens—supported by hard benchmark numbers and annotated, real-world code.
Framework marketing claims "modular," "easy coding," "robust tool use." Actual benchmarks suggest otherwise. Using sweta2503/agent-framework-benchmark, which runs LangGraph, CrewAI, and AutoGen through 107 reproducible data engineering tasks (dataset ingestion, transformation, ETL orchestration, etc.), the performance gap is impossible to ignore.
The table shows concrete outcomes on fixed testbeds—identical models, prompts, and tool APIs for comparability.
Source: sweta2503/agent-framework-benchmark, June 2024. Task passes require strict output assertions.
LangGraph dominates: higher pass rate, lower latency, far less token burn, and minimal irrecoverable failure. This is not marginal.
LangGraph's directed-graph model delivers reliability by design: explicit state, crisp control flow, and minimal boilerplate. Most frameworks hide orchestration behind recursive chains, brittle message passing, or role abstractions. LangGraph puts business logic front and center.
import langgraph
from langgraph.nodes import ToolNode, LLMNode
graph = langgraph.Graph()
def ingest_data(input):
return {"df": load_csv_to_df(input["file_path"])}
def transform_data(input):
df = input["df"]
return {"df": df.dropna().drop_duplicates()}
def load_data(input):
df = input["df"]
return {"success": upload_df_to_db(df)}
graph.add_node(ToolNode(func=ingest_data), inputs=["file_path"])
graph.add_node(ToolNode(func=transform_data), inputs=["df"])
graph.add_node(ToolNode(func=load_data), inputs=["df"])
graph.add_edge("ingest_data", "transform_data")
graph.add_edge("transform_data", "load_data")
graph.compile("file_path")
# Run: graph.run({"file_path": "/tmp/source.csv"})
State and control flow are explicit. No prompt glue or nested dict acrobatics.
Each step is a stand-alone plain function.
No message passing or confirmation chatter. Inputs and outputs are your actual objects.
The difference becomes starker against CrewAI and AutoGen in both code and operational complexity.
CrewAI wraps orchestration in agent chat and rigid task objects. The result: verbosity, boilerplate, and heavy tokenization. The same ETL expands to double the lines—nearly all structural, not business logic. The chat abstraction is not free. Each agent introduces message-passing overhead that stacks at every step.
from crewai import Agent, Task, Crew
class IngestAgent(Agent):
def run(self, file_path):
return {"df": load_csv_to_df(file_path)}
class TransformAgent(Agent):
def run(self, df):
transformed = df.dropna().drop_duplicates()
return {"df": transformed}
class LoadAgent(Agent):
def run(self, df):
return {"success": upload_df_to_db(df)}
ingest = IngestAgent(role="Ingest")
transform = TransformAgent(role="Transform")
load = LoadAgent(role="Load")
task1 = Task(ingest, inputs=["file_path"])
task2 = Task(transform, inputs=["df"])
task3 = Task(load, inputs=["df"])
crew = Crew(tasks=[task1, task2, task3])
crew.setup_dependencies([(task1, task2), (task2, task3)])
# Run: crew.run({"file_path": "/tmp/source.csv"})
Operational Overhead: CrewAI's callback chains and chatter stack up—in the benchmark, median token usage per task is 4120 (vs 2350 for LangGraph), with latency inflated by a third. Each agent restates or echoes prior context. Logs for a single run:
[IngestAgent] Loaded dataframe from /tmp/source.csv, passing to next agent.
[TransformAgent] Received dataframe, initiating transformation...
[TransformAgent] Sending cleaned dataframe to LoadAgent.
[LoadAgent] Received dataframe, uploading to DB...
Multiply this "meta" chatter by every workflow step, and your token bill explodes with no benefit.
AutoGen markets itself for stateful agentic pipelines, but fails on workflows with chained state or error branching. Benchmark logs show frequent breakdowns resembling silent pipeline drops or input mismatches.
Task: ETL with fallback on transformation error (branch to cleanup, retry load).
from autogen import AgentFlow
def ingest(file_path):
return {"df": load_csv_to_df(file_path)}
def transform(df):
try:
return {"df": df.dropna().drop_duplicates()}
except Exception as e:
return {"error": str(e)}
def load(df):
return {"success": upload_df_to_db(df)}
flow = AgentFlow()
flow.add_fn("ingest", ingest)
flow.add_fn("transform", transform)
flow.add_fn("load", load)
flow.set_flow([("ingest", "transform"), ("transform", "load")])
# Run: flow.execute({"file_path": "/tmp/source.csv"})
Observed Benchmark Output:
[AutoGen] Step: ingest --> Success
[AutoGen] Step: transform --> Error: 'DataFrame' object has no attribute 'dropna'
[AutoGen] Step: load --> Missing input 'df', raising up PipelineError
Intermediary state must be manually serialized and re-parsed between steps—unhandled types cause silent exceptions, missing results, and cascade failures. In the full benchmark, ~46% of multi-step AutoGen tasks either lost state or crashed on pipeline handoff.
Task: Extract order CSV, drop rows missing payment, upload to DB.
LangGraph:
graph = langgraph.Graph()
graph.add_node(ToolNode(func=lambda inp: {"df": load_csv_to_df(inp["file_path"])}), inputs=["file_path"])
graph.add_node(ToolNode(func=lambda inp: {"df": inp["df"][inp["df"]["payment_id"].notnull()]}), inputs=["df"])
graph.add_node(ToolNode(func=lambda inp: {"success": upload_df_to_db(inp["df"])}), inputs=["df"])
graph.add_edge("ToolNode_1", "ToolNode_2")
graph.add_edge("ToolNode_2", "ToolNode_3")
graph.compile("file_path")
result = graph.run({"file_path": "/data/orders.csv"})
13 lines, 1970 tokens, 10.7s latency.
CrewAI:
ingest = IngestAgent()
filter_null = Agent(lambda df: {"df": df[df["payment_id"].notnull()]})
load = LoadAgent()
t1 = Task(ingest, ["file_path"])
t2 = Task(filter_null, ["df"])
t3 = Task(load, ["df"])
crew = Crew([t1, t2, t3])
crew.setup_dependencies([(t1, t2), (t2, t3)])
crew.run({"file_path": "/data/orders.csv"})
19 lines (+class boilerplate elsewhere), 3775 tokens, 19.8s latency.
AutoGen:
flow = AgentFlow()
flow.add_fn("ingest", lambda file_path: {"df": load_csv_to_df(file_path)})
flow.add_fn("filter", lambda df: {"df": df[df["payment_id"].notnull()]})
flow.add_fn("load", lambda df: {"success": upload_df_to_db(df)})
flow.set_flow([("ingest", "filter"), ("filter", "load")])
flow.execute({"file_path": "/data/orders.csv"})
10 lines but brittle: fails if intermediate df is unparseable; 3150 tokens (if successful), 22s latency typical.
Summary: LangGraph is the only concise, robust, cost-efficient pipeline. CrewAI bloats code, burns tokens, and slows execution. AutoGen's approach collapses when type handoff is non-trivial.
Benchmark data and real code reveal why LangGraph outperforms:
CrewAI's inefficiency is built-in: Its chat abstractions pump up both code and token usage, causing slowdowns and error accumulation.
AutoGen collapses under real pipeline requirements: It handles stateless chat, but not robust state or type exchange for multi-step operations.
This is not theoretical. The gap is visible in code, benchmarks, and error logs. For agentic ETL and data engineering, LangGraph is the new baseline.
sweta2503/agent-framework-benchmark
langgraph-vs-crewai comparative code