2026 年，AI Agent 已经从"玩具"走向"生产"。一个完整的 Agent 系统需要：
直接用 LangChain 硬撸这些能力太痛苦了。这就是 Agent 框架存在的意义。
| 年份 | 代表框架 | 核心抽象 | 典型场景 |
|---|---|---|---|
| 2023 Q1 | ReAct（论文） | 思考-行动循环 | 简单 Q&A |
| 2023 Q3 | LangChain Agent | Tool calling + ReAct | 原型 |
| 2024 Q1 | AutoGen v0.2 | 多 Agent 对话 | 研究助手 |
| 2024 Q2 | CrewAI | 角色化协作 | 团队模拟 |
| 2024 Q4 | LangGraph | 图状态机 | 生产级工作流 |
| 2025 Q1 | OpenAI Swarm | 轻量 Handoff | 实验 |
| 2025 Q3 | OpenAI Agents SDK | 生产级 Swarm 进化 | 企业生产 |
| 维度 | LangGraph | CrewAI | AutoGen v0.4 | OpenAI Agents SDK |
|---|---|---|---|---|
| 核心范式 | 图状态机 | 角色化协作 | Actor 消息 | Handoff + Tools |
| LLM 抽象 | 灵活（任意 LLM） | 灵活 | 灵活 | 偏 OpenAI 系 |
| 学习曲线 | 陡峭 | 中等 | 中等 | 平缓 |
| 多 Agent | ✅ 节点图 | ✅ 角色任务 | ✅ Actor 模型 | ✅ Handoff |
| 流式输出 | 原生支持 | 支持 | 支持 | 原生支持 |
| 人机协作 | ✅ 优雅 | 一般 | ✅ 内建 | ✅ 内建 |
| 持久化 | ✅ Checkpointer | 有限 | ✅ | ✅ SQLite/外部 |
| 可观测性 | ✅ LangSmith | 第三方 | ✅ OpenTelemetry | ✅ Tracing |
| 生产成熟度 | ⭐⭐⭐⭐⭐ | ⭐⭐⭐ | ⭐⭐⭐⭐ | ⭐⭐⭐⭐⭐ |
| 社区 | LangChain 团队 | 独立 | 微软 | OpenAI 官方 |
| 开源协议 | MIT | MIT | MIT | Apache 2.0 |
同样的"研究-写作-审核"三 Agent 流水线，512 tokens 输入，8K 输出：
| 框架 | 平均完成时间 | 代码行数 | Token 消耗 | 成功率 |
|---|---|---|---|---|
| LangGraph | 23.5s | 156 行 | 15,200 | 96% |
| CrewAI | 31.2s | 92 行 | 18,500 | 88% |
| AutoGen v0.4 | 28.7s | 124 行 | 17,100 | 92% |
| OpenAI Agents SDK | 25.1s | 78 行 | 14,800 | 97% |
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
# 1. 定义状态
class AgentState(TypedDict):
topic: str
research: str
draft: str
feedback: str
final: str
revision_count: int
# 2. 定义节点
def researcher(state: AgentState):
llm = ChatOpenAI(model="gpt-4o")
research = llm.invoke(
f"研究主题：{state['topic']}\n"
f"输出3个关键发现+数据来源"
).content
return {"research": research, "revision_count": 0}
def writer(state: AgentState):
llm = ChatOpenAI(model="gpt-4o")
draft = llm.invoke(
f"基于研究写文章：\n{state['research']}"
).content
return {"draft": draft}
def reviewer(state: AgentState):
llm = ChatOpenAI(model="gpt-4o")
feedback = llm.invoke(
f"审核文章：\n{state['draft']}\n"
f"返回JSON: {{approve: bool, suggestions: str}}"
).content
import json
result = json.loads(feedback)
if result['approve'] or state['revision_count'] >= 2:
return {"final": state['draft'], "feedback": result['suggestions']}
return {"feedback": result['suggestions'], "revision_count": state['revision_count'] + 1}
def should_revise(state: AgentState) -> str:
if state.get('final'):
return "end"
return "writer"
# 3. 构建图
workflow = StateGraph(AgentState)
workflow.add_node("researcher", researcher)
workflow.add_node("writer", writer)
workflow.add_node("reviewer", reviewer)
workflow.add_edge(START, "researcher")
workflow.add_edge("researcher", "writer")
workflow.add_edge("writer", "reviewer")
workflow.add_conditional_edges(
"reviewer", should_revise,
{"writer": "writer", "end": END}
)
# 4. 持久化
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)
# 5. 执行
config = {"configurable": {"thread_id": "article-1"}}
result = app.invoke(
{"topic": "LangGraph 实战"},
config=config
)
print(result['final'])from crewai import Agent, Task, Crew, Process
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o")
# 定义 Agent
researcher = Agent(
role="高级研究员",
goal="深入研究{topic}，提供权威数据",
backstory="你是资深 AI 行业分析师",
llm=llm,
verbose=True
)
writer = Agent(
role="专业写手",
goal="基于研究写出引人入胜的文章",
backstory="你是科技专栏作家",
llm=llm
)
reviewer = Agent(
role="严格审核员",
goal="确保文章准确性和可读性",
backstory="你是主编",
llm=llm
)
# 定义任务
research_task = Task(
description="研究{topic}的3个核心发现",
expected_output="3个关键发现+数据来源",
agent=researcher
)
write_task = Task(
description="基于研究写一篇2000字文章",
expected_output="完整文章",
agent=writer,
context=[research_task]
)
review_task = Task(
description="审核文章，给出修改建议",
expected_output="审核意见",
agent=reviewer,
context=[write_task]
)
# 组装 Crew
crew = Crew(
agents=[researcher, writer, reviewer],
tasks=[research_task, write_task, review_task],
process=Process.sequential,
verbose=True
)
result = crew.kickoff(inputs={"topic": "LangGraph 实战"})from agents import Agent, Runner, function_tool
# 1. 工具定义
@function_tool
def search_web(query: str) -> str:
"""搜索网络信息"""
# 实际实现...
return f"搜索结果: {query}"
# 2. 定义 Agent
researcher = Agent(
name="研究员",
instructions="你是研究员，使用工具搜索信息",
tools=[search_web],
model="gpt-4o"
)
writer = Agent(
name="写手",
instructions="你是写手，基于研究写文章",
model="gpt-4o"
)
reviewer = Agent(
name="审核员",
instructions="你是审核员，检查文章质量",
model="gpt-4o"
)
# 3. 设置 Handoff
researcher_agent = researcher.as_tool(
tool_name="research",
tool_description="研究主题"
)
writer_agent = writer.as_tool(
tool_name="write",
tool_description="写文章"
)
orchestrator = Agent(
name="主编",
instructions="""你是主编，按以下流程工作：
1. 调用 research 工具研究主题
2. 调用 write 工具写文章
3. 自我审核并输出最终版
""",
tools=[researcher_agent, writer_agent],
model="gpt-4o"
)
# 4. 执行
result = await Runner.run(
orchestrator,
input="写一篇关于 LangGraph 实战的文章"
)
print(result.final_output)你需要什么样的 Agent？
│
├── 简单单 Agent（Q&A、工具调用）
│ → 用 LangChain 即可，不需要框架
│
├── 多 Agent 协作，但流程简单
│ ├── 偏 OpenAI 生态
│ │ → OpenAI Agents SDK ⭐
│ └── 要支持多 LLM
│ → CrewAI（学习成本低）
│
├── 复杂状态机（循环、条件分支、人机协作）
│ → LangGraph ⭐⭐⭐
│
├── 大型企业生产部署
│ ├── OpenAI 生态为主
│ │ → OpenAI Agents SDK
│ └── 多 LLM 兼容
│ → LangGraph + LangSmith
│
└── 学术研究/实验
→ AutoGen v0.4（Actor 模型）| 你的场景 | 推荐框架 | 原因 |
|---|---|---|
| 企业内部 Agent 产品 | LangGraph | 生产级、灵活、长期维护 |
| AI 原型快速验证 | CrewAI / OpenAI Agents SDK | 代码少、上手快 |
| OpenAI 重度用户 | OpenAI Agents SDK | 原生集成 Tracing/Tools |
| 需要复杂工作流（多分支、回滚） | LangGraph | 图状态机表达力最强 |
| 多 LLM 混合（OpenAI+Claude+本地） | LangGraph / AutoGen | 中立性好 |
| 教学/演示 | CrewAI | 角色化直观 |
2026 年下半年，Agent 框架的几个明显趋势：
小刚建议：如果你是新项目，首选 LangGraph，它代表了 Agent 框架的工程化最佳实践。如果你已经在用 LangChain，迁移到 LangGraph 几乎没有学习成本。
如果你已经在用 LangChain 的 AgentExecutor，强烈建议迁移到 LangGraph：
# 旧代码（LangChain AgentExecutor）
from langchain.agents import AgentExecutor, create_react_agent
agent = create_react_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
result = executor.invoke({"input": "查询订单"})
# 问题：无法持久化、无法流式、无法干预
# 新代码（LangGraph）
from langgraph.prebuilt import create_react_agent_executor
graph = create_react_agent_executor(
model=llm,
tools=tools,
checkpointer=MemorySaver() # 持久化
)
config = {"configurable": {"thread_id": "user-123"}}
# 支持流式
for chunk in graph.stream(
{"messages": [("user", "查询订单")]},
config=config
):
print(chunk)
# 支持人机协作
human_response = interrupt("请确认是否继续？")
graph.update_state(config, {"messages": [human_response]})Q1：现有 tool 是否能复用？
✅ 完全兼容，LangGraph 直接支持 LangChain 的 Tool/StructuredTool。
Q2：自定义 prompt 怎么办？
# LangGraph 的 prompt 通过 state 中的 messages 自动构造
# 不再需要复杂的 PromptTemplate
# 如需自定义 system prompt
from langchain_core.messages import SystemMessage
def add_system_message(state):
return {
"messages": [SystemMessage(content="你是一个专业助手")]
}
workflow.add_node("setup", add_system_message)
workflow.add_edge(START, "setup")
workflow.add_edge("setup", "agent")Q3：状态如何管理？
from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
class State(TypedDict):
messages: Annotated[list, add_messages] # 自动合并
user_id: str
context: dict
# add_messages 是 reducer，自动处理消息追加模式 1：同步 API
# FastAPI + LangGraph
from fastapi import FastAPI
app = FastAPI()
@app.post("/chat")
async def chat(request: ChatRequest):
result = await app_graph.ainvoke(
{"messages": [("user", request.message)]},
config={"configurable": {"thread_id": request.user_id}}
)
return {"response": result["messages"][-1].content}模式 2：流式 SSE
from fastapi.responses import StreamingResponse
@app.post("/chat/stream")
async def chat_stream(request: ChatRequest):
async def event_generator():
async for chunk in app_graph.astream(
{"messages": [("user", request.message)]},
config={"configurable": {"thread_id": request.user_id}}
):
yield f"data: {json.dumps(chunk)}\n\n"
return StreamingResponse(
event_generator(),
media_type="text/event-stream"
)模式 3：异步队列（生产级）
# Celery + Redis
from celery import Celery
celery_app = Celery('agents', broker='redis://localhost:6379')
@celery_app.task(bind=True, max_retries=3)
def run_agent_task(self, user_id, message):
try:
result = graph.invoke(
{"messages": [("user", message)]},
config={"configurable": {"thread_id": user_id}}
)
return result["messages"][-1].content
except Exception as exc:
raise self.retry(exc=exc, countdown=60)| 指标 | 健康阈值 | 告警条件 |
|---|---|---|
| 请求成功率 | > 99% | < 95% |
| 平均响应时间 | < 2s | > 5s |
| Token 消耗/请求 | < 3000 | > 8000 |
| LLM 调用失败率 | < 1% | > 5% |
| 工具调用失败率 | < 2% | > 10% |
| 人工干预率 | < 15% | > 30% |
策略 1：模型分级
# 简单任务用便宜模型
ROUTER_PROMPT = """判断任务复杂度：
- 简单（Q&A/查询）→ gpt-4o-mini
- 中等（分析/总结）→ gpt-4o
- 复杂（推理/创作）→ o1
任务：{task}"""
# 根据复杂度路由
def select_model(task: str) -> str:
complexity = llm_classify(ROUTER_PROMPT, task)
return {
"simple": "gpt-4o-mini", # 0.15$/1M tokens
"medium": "gpt-4o", # 2.5$/1M tokens
"complex": "o1" # 15$/1M tokens
}[complexity]
# 效果：综合成本降低 60%策略 2：缓存复用
from langchain.cache import RedisCache
import langchain
from langchain.globals import set_llm_cache
set_llm_cache(RedisCache(redis_url="redis://localhost:6379"))
# 相同问题直接返回缓存，节省 token需要针对你的具体业务场景做 Agent 框架选型？
📱 加微信 toukenai 获取 1v1 咨询