背景为什么需要专门的编排框架早期的AI应用模式很简单用户输入 → LLM处理 → 输出结果。但随着业务复杂度的提升这种单步模式已经无法满足需求。真实世界的任务需要多步骤规划、工具调用、条件分支、并行执行、状态持久化……这就催生了AI Agent编排框架的需求。2026年这个领域已经形成了几个主流方向LangGraph图状态机、LlamaIndex Workflows事件驱动、Microsoft AutoGen多智能体对话、以及新兴的裸写状态机范式。本文从工程实践角度分析各框架的适用场景和核心用法。## 一、LangGraph图状态机的工程标准### 1.1 核心概念LangGraph把Agent工作流建模为有向图节点是处理步骤边是条件路由State是在节点间传递的共享状态。pythonfrom langgraph.graph import StateGraph, ENDfrom typing import TypedDict, List, Optional, Annotatedimport operator# 定义工作流状态class ResearchState(TypedDict): query: str # 用户查询 search_results: List[str] # 搜索结果 analyzed_results: List[dict] # 分析后的结果 draft_report: Optional[str] # 草稿报告 final_report: Optional[str] # 最终报告 iteration_count: int # 迭代次数 messages: Annotated[List, operator.add] # 消息历史自动追加# 定义节点处理函数def search_node(state: ResearchState) - ResearchState: 执行搜索 results perform_search(state[query]) return {search_results: results}def analyze_node(state: ResearchState) - ResearchState: 分析搜索结果 analyzed [] for result in state[search_results]: analysis llm_analyze(result) analyzed.append(analysis) return {analyzed_results: analyzed}def write_report_node(state: ResearchState) - ResearchState: 生成报告 report llm_write_report(state[analyzed_results]) return {draft_report: report}def review_node(state: ResearchState) - ResearchState: 审核报告质量 quality llm_review(state[draft_report]) return { final_report: state[draft_report] if quality[score] 8 else None, iteration_count: state[iteration_count] 1 }# 路由函数def should_revise(state: ResearchState) - str: 决定是完成还是修改 if state[final_report]: return complete if state[iteration_count] 3: return force_complete # 强制结束防止死循环 return revise# 构建图workflow StateGraph(ResearchState)# 添加节点workflow.add_node(search, search_node)workflow.add_node(analyze, analyze_node)workflow.add_node(write, write_report_node)workflow.add_node(review, review_node)# 添加边workflow.set_entry_point(search)workflow.add_edge(search, analyze)workflow.add_edge(analyze, write)workflow.add_edge(write, review)# 条件路由workflow.add_conditional_edges( review, should_revise, { complete: END, force_complete: END, revise: write # 回到写作节点重新生成 })# 编译app workflow.compile()# 执行result app.invoke({ query: 2026年AI编排框架的发展趋势, search_results: [], analyzed_results: [], draft_report: None, final_report: None, iteration_count: 0, messages: []})### 1.2 持久化与断点恢复LangGraph的一大优势是内置的持久化支持pythonfrom langgraph.checkpoint.sqlite import SqliteSaverfrom langgraph.checkpoint.memory import MemorySaver# 使用SQLite持久化with SqliteSaver.from_conn_string(:memory:) as memory: app workflow.compile(checkpointermemory) # 每个thread_id是独立的会话 config {configurable: {thread_id: session_001}} # 第一次运行 result1 app.invoke({query: AI trends 2026}, configconfig) # 中断后恢复从断点继续 current_state app.get_state(config) print(f当前节点: {current_state.next}) # 可以修改状态后继续 app.update_state(config, {iteration_count: 0}) result2 app.invoke(None, configconfig) # 从中断点继续### 1.3 Human-in-the-Looppython# 在关键节点等待人工确认from langgraph.graph import interruptdef critical_decision_node(state: ResearchState) - ResearchState: 需要人工确认的关键决策 # interrupt会暂停执行等待人工输入 human_feedback interrupt({ action: 需要确认, draft: state[draft_report], question: 这份报告是否符合要求如不满意请提供修改意见。 }) if human_feedback[approved]: return {final_report: state[draft_report]} else: # 根据反馈修改 revised llm_revise(state[draft_report], human_feedback[comments]) return {draft_report: revised}## 二、LlamaIndex Workflows事件驱动编排LlamaIndex Workflows采用事件驱动模型更适合复杂的异步工作流pythonfrom llama_index.core.workflow import ( Workflow, StartEvent, StopEvent, step, Event)from typing import Union# 定义事件类型class QueryEvent(Event): query: strclass SearchResultEvent(Event): results: list query: strclass AnalysisEvent(Event): analysis: str query: strclass ResearchWorkflow(Workflow): step async def start(self, ev: StartEvent) - QueryEvent: 接收用户输入发出查询事件 return QueryEvent(queryev.input) step async def search(self, ev: QueryEvent) - SearchResultEvent: 执行搜索 results await async_search(ev.query) return SearchResultEvent(resultsresults, queryev.query) step async def analyze(self, ev: SearchResultEvent) - AnalysisEvent: 分析搜索结果 analysis await async_analyze(ev.results) return AnalysisEvent(analysisanalysis, queryev.query) step async def generate_report(self, ev: AnalysisEvent) - StopEvent: 生成最终报告 report await async_generate_report(ev.analysis, ev.query) return StopEvent(resultreport)# 运行工作流workflow ResearchWorkflow(timeout120)result await workflow.run(inputAI Agent编排框架对比)### 2.1 并行步骤事件驱动模型天然支持并行pythonclass ParallelResearchWorkflow(Workflow): step async def start(self, ev: StartEvent) - QueryEvent: return QueryEvent(queryev.input) step async def parallel_search(self, ev: QueryEvent) - Union[ WebSearchEvent, DatabaseSearchEvent, NewsSearchEvent ]: 分发到多个并行搜索 # 同时触发三个搜索事件 self.send_event(WebSearchEvent(queryev.query)) self.send_event(DatabaseSearchEvent(queryev.query)) self.send_event(NewsSearchEvent(queryev.query)) step async def collect_results( self, web: WebSearchEvent, db: DatabaseSearchEvent, news: NewsSearchEvent ) - AnalysisEvent: 等待所有搜索完成后汇总 all_results web.results db.results news.results return AnalysisEvent(combined_resultsall_results)## 三、AutoGen多智能体对话框架AutoGen的核心理念是多个Agent通过对话协作完成任务pythonimport autogenfrom autogen import AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager# 配置LLMllm_config { config_list: [{model: gpt-4o, api_key: YOUR_API_KEY}], temperature: 0.1}# 定义专业Agentplanner AssistantAgent( namePlanner, system_message你是项目规划专家。 负责分析任务、制定计划、协调其他Agent。 当收到任务时先制定清晰的执行计划。, llm_configllm_config)researcher AssistantAgent( nameResearcher, system_message你是技术研究专家。 负责收集资料、深度分析、提供技术洞察。 只做研究工作不做规划和写作。, llm_configllm_config)writer AssistantAgent( nameWriter, system_message你是技术写作专家。 负责基于研究结果编写清晰、结构化的技术文档。 写作要准确、易读、有实用价值。, llm_configllm_config)reviewer AssistantAgent( nameReviewer, system_message你是质量审核专家。 负责审核文档的准确性、完整性和可读性。 发现问题时给出具体的修改建议。, llm_configllm_config)# 创建代理用户触发对话user_proxy UserProxyAgent( nameUser, human_input_modeNEVER, # 自动化模式 code_execution_config{work_dir: output, use_docker: False}, is_termination_msglambda x: FINAL_REPORT in x.get(content, ))# 群组对话group_chat GroupChat( agents[user_proxy, planner, researcher, writer, reviewer], messages[], max_round15, # 最大对话轮数 speaker_selection_methodauto)manager GroupChatManager(groupchatgroup_chat, llm_configllm_config)# 启动任务user_proxy.initiate_chat( manager, message请创作一篇关于LLM应用安全最佳实践的技术文章)## 四、何时用框架何时裸写### 4.1 框架的真实成本框架带来的问题-调试困难执行链路长出错时堆栈很深-学习成本LangGraph的图概念、LlamaIndex的事件系统各有学习曲线-版本依赖框架更新快API变化频繁-过度封装简单任务用了复杂框架反而增加复杂度### 4.2 判断标准任务是否需要以下特性├── 需要持久化状态/断点恢复 → LangGraph├── 需要多Agent协作对话 → AutoGen ├── 有复杂的异步并行流程 → LlamaIndex Workflows└── 以上都不需要逻辑相对简单 → 裸写状态机### 4.3 裸写状态机示例对于中等复杂度的任务简单的状态机往往更清晰pythonfrom enum import Enumfrom dataclasses import dataclassfrom typing import Optionalclass TaskState(Enum): INIT init SEARCHING searching ANALYZING analyzing WRITING writing REVIEWING reviewing DONE done FAILED faileddataclassclass WorkflowContext: query: str state: TaskState TaskState.INIT search_results: list None analysis: str None report: str None error: Optional[str] None retry_count: int 0def run_workflow(query: str) - WorkflowContext: ctx WorkflowContext(queryquery) while ctx.state not in [TaskState.DONE, TaskState.FAILED]: try: if ctx.state TaskState.INIT: ctx.state TaskState.SEARCHING elif ctx.state TaskState.SEARCHING: ctx.search_results do_search(ctx.query) ctx.state TaskState.ANALYZING elif ctx.state TaskState.ANALYZING: ctx.analysis do_analysis(ctx.search_results) ctx.state TaskState.WRITING elif ctx.state TaskState.WRITING: ctx.report do_write(ctx.analysis) ctx.state TaskState.REVIEWING elif ctx.state TaskState.REVIEWING: if review_quality(ctx.report) 7: ctx.state TaskState.DONE elif ctx.retry_count 2: ctx.retry_count 1 ctx.state TaskState.WRITING # 重写 else: ctx.state TaskState.DONE # 强制结束 except Exception as e: ctx.error str(e) ctx.state TaskState.FAILED return ctx简洁、可测试、易调试——有时候朴素才是最好的工程选择。## 结语2026年Agent编排框架的选择核心原则是匹配复杂度-LangGraph生产级、有状态的复杂工作流需要持久化和人工干预点-LlamaIndex Workflows异步并行、事件驱动的工作流-AutoGen需要多Agent协作讨论的任务-裸写逻辑清晰、复杂度适中的工作流框架是工具不是目的。能用简单方案解决的问题不要引入复杂框架。真正重要的是系统能稳定运行出了问题能快速定位——这才是工程化的本质。