阅读时间约 15 分钟前置知识理解 Agent 决策循环P01-P07前七篇文章你学完了决策循环、工具调用、上下文管理、重试策略、记忆系统、评测……但每篇文章只讲一个点。现在把它们串起来一个真实的、端到端的 Agent 工作流长什么样前言你不需要框架但你最终需要工作流先说一个反直觉的话大部分 Agent 项目不需要框架。你完全可以用 P01 的 while 循环、P03 的 ContextManager、P04 的 RecoveryManager、P05 的 MemoryManager从零搭一个能跑的 Agent。但当你需要状态持久化服务重启不丢上下文时间旅行调试回退到上一步重新执行多人协作多个 Agent 各司其职人工审核关键步骤暂停等人确认框架才开始有价值。它不解决你的 Agent 智能问题这些你前七篇都学过了它解决的是工程化问题可调试、可维护、可扩展。本章核心框架不是必需品但复杂项目离不开工作流引擎。先理解原理再选框架。P01 决策循环框架需要处理的事P02 工具调用P03 上下文管理P04 重试策略P05 记忆系统P06 评测体系状态机 / 持久化 / 人工审核 / 观测追踪第一部分为什么你需要工作流框架从零搭 Agent 的天花板在哪里以 P01 的基础 while 循环为例# P01 的基础 Agent 循环能跑但有问题defagent_loop(messages):whileTrue:responsecall_llm(messages)ifresponse.content:returnresponse.contentifresponse.tool_calls:messages.extend(execute_tools(response.tool_calls))这个循环能跑但有 5 个生产环境问题问题现象框架能解决什么状态不持久服务重启后对话丢失检查点保存/恢复无法回退某步出错只能从头开始时间旅行调试人工审核缺失危险操作无确认就执行人审拦截点调试困难不知道哪一步出错全链路追踪多 Agent 协作复杂任务无法拆分子流程工作流编排框架解决的不是Agent 智能不够而是Agent 跑不好。本章要点工作流框架不增加智能它解决 5 个工程问题状态持久化、时间旅行调试、人工审核、全链路追踪、多 Agent 协作。第二部分三大框架对比2026 年 Agent 框架三巨头LangGraph、CrewAI、AutoGen。选哪个维度LangGraphCrewAIAutoGen核心模型状态机/图角色扮演对话流适合场景复杂控制流 生产级多 Agent 协作多 Agent 对话持久化✅ 内置❌ 无✅ 有时间旅行✅ 支持❌ 不支持⚠️ 有限人审拦截✅ 节点级别⚠️ 有限⚠️ 有限学习曲线陡峭低中生产成熟度高中中一句话选型要复杂控制和生产级能力 →LangGraph要快速搭多 Agent 协作 →CrewAI要做多 Agent 对话实验 →AutoGen纯用 OpenAI 生态 →OpenAI Agents SDK纯用 Claude 生态 →Anthropic Agent SDK本章要点框架不决定智能上限只决定工程能力。LangGraph 最强状态机持久化时间旅行CrewAI 最简单角色扮演AutoGen 适合对话实验。第三部分实战从零到生产用一个真实的 Agent 项目串起前七篇所有知识点技术竞品分析 Agent。场景用户输入 3 个竞品的名称Agent 搜索信息、分析对比、输出结构化报告。步骤 1定义 Agent 节点基于 P01 的决策循环importjsonfromtypingimportTypedDict,Annotateddimportoperator# ═══════════════════════════════════════════# 状态定义所有节点共享一个状态P03 上下文管理# ═══════════════════════════════════════════classAgentState(TypedDict):messages:list# 对话历史P03 上下文context:dict# 任务上下文research_results:dict# 研究结果P05 记忆final_report:str# 最终报告errors:list# 错误记录P04 重试human_approved:bool# 人工审核标志人审拦截步骤 2定义节点函数每个节点 一个子 Agentimporthttpximportopenai clientopenai.OpenAI(base_urlYOUR_BASE_URL)# ── 节点 1搜索 ──defsearch_agents(state:AgentState)-AgentState:收集竞品基本信息P02 工具调用 P04 重试search_results{}fornameinstate[context].get(competitors,[]):# P04 重试搜索失败自动重试try:resphttpx.get(https://search.example.com/api,params{q:f{name}技术分析},timeout5.0)search_results[name]resp.json()exceptExceptionase:state[errors].append(f搜索{name}失败:{e})search_results[name]{error:search_failed}return{research_results:search_results}# ── 节点 2深度分析 ──defanalyze(state:AgentState)-AgentState:对搜索结果进行深度分析P01 决策循环 P06 工具选择researchstate[research_results]# P02 工具选择正确的分析工具system_prompt 你是技术分析师。根据收集的信息从以下维度对比 - 架构设计 - 性能指标 - 适用场景 - 优缺点 messages[{role:system,content:system_prompt},{role:user,content:json.dumps(research,ensure_asciiFalse,indent2)}]# P02 工具调用分析引擎tools[{type:function,function:{name:compare_technologies,description:对比两种技术架构的优缺点,parameters:{type:object,properties:{tech1:{type:string},tech2:{type:string},dimensions:{type:array,items:{type:string}}}}}}]responseclient.chat.completions.create(modeldeepseek-v4-flash,messagesmessages,toolstools)return{research_results:{**state[research_results],analysis:response.choices[0].message.content}}# ── 节点 3报告生成 ──defgenerate_report(state:AgentState)-AgentState:生成结构化报告P05 记忆注入 P07 评测验收条件system_prompt 你是一位技术顾问。根据分析报告生成对比报告。 ## 格式要求 - Markdown 格式 - 包含对比表格 - 包含选型建议 ## 验收条件 - 包含至少 3 个维度的对比 - 包含明确的选型建议 - 包含风险提示 # P05 记忆注入contextstate[context]ifuser_profileincontext:system_promptf\n\n## 用户背景\n{context[user_profile]}messages[{role:system,content:system_prompt},{role:user,content:f根据以下分析生成报告{state[research_results].get(analysis,)}竞品列表{context.get(competitors,[])}}]responseclient.chat.completions.create(modeldeepseek-v4-flash,messagesmessages)return{final_report:response.choices[0].message.content}# ── 节点 4人工审核 ──defhuman_review(state:AgentState)-AgentState: 关键步骤人工审核人审拦截点 在 LangGraph 中这个节点会暂停执行等待用户回复。 这里是生产环境的关键安全机制。 reportstate[final_report]print(*50)print( 报告审核)print(report[:300])print(...*10)print(*50)print(回复 确认 以继续或提出修改意见。)# 实际生产环境这里等待用户输入# 不是调用 LLM而是等待真人回复approvalinput(是否确认(确认/修改): ).strip()return{human_approved:approval确认}步骤 3定义路由逻辑条件边fromlanggraph.graphimportStateGraph,END# ── 条件路由函数 ──defshould_review(state:AgentState)-str:报告长度超过 2000 字需要人审iflen(state[final_report])2000:returnhuman_reviewreturndonedefis_approved(state:AgentState)-str:人审不通过则修改ifnotstate.get(human_approved,False):returnmodify_reportreturndonedefhas_errors(state:AgentState)-str:有错误则重试ifstate.get(errors):returnretryreturnanalyze# ── 构建图 ──workflowStateGraph(AgentState)# 添加节点workflow.add_node(search,search_agents)workflow.add_node(analyze,analyze)workflow.add_node(generate_report,generate_report)workflow.add_node(human_review,human_review)workflow.add_node(modify_report,generate_report)# 修改报告也调用报告生成# 设置入口workflow.set_entry_point(search)# 添加条件边搜索 → 分析 → 报告 → 人审 → 结束workflow.add_edge(search,analyze)workflow.add_edge(analyze,generate_report)# 报告生成后根据条件路由workflow.add_conditional_edges(generate_report,should_review,{human_review:human_review,done:END})# 人审后根据审批结果路由workflow.add_conditional_edges(human_review,is_approved,{modify_report:modify_report,done:END})# 编译appworkflow.compile(interrupt_before[human_review]# 人审节点前自动暂停)步骤 4状态持久化和时间旅行调试# ── 持久化保存检查点服务重启不丢状态 ──fromlanggraph.checkpoint.memoryimportMemorySaver checkpointerMemorySaver()app_with_persistenceworkflow.compile(checkpointercheckpointer)# 运行并保存状态config{configurable:{thread_id:session_001}}resultapp_with_persistence.invoke(input_state,configconfig)# ── 时间旅行回退到上一步重新执行 ──# 获取之前保存的 checkpointcheckpointslist(checkpointer.list(config))forcpincheckpoints:print(fCheckpoint:{cp.thread_id}{cp.metadata[ts]})# 回退到某个 checkpointcheckpointcheckpoints[-2]# 倒数第二个state_at_checkpointcheckpointer.get_tuple(config).state# 可以在此基础上重新执行本章要点工作流 节点子 Agent 条件边路由 状态共享上下文。人审拦截和状态持久化是生产环境最关键的两个特性。第四部分完整项目架构一个完整的项目不只是代码还包括my_agent_project/ ├── workflow/ │ ├── nodes/ # 节点函数P01 决策循环 │ │ ├── search.py │ │ ├── analyze.py │ │ ├── report.py │ │ └── review.py │ ├── graph.py # 图定义和路由P03 上下文流 │ └── app.py # 入口 ├── tools/ # 工具定义P02 工具调用 P04 重试 │ ├── search.py │ ├── analyze.py │ └── report.py ├── memory/ # 记忆系统P05 记忆 │ ├── episodic.py # 情景记忆 │ └── semantic.py # 语义记忆 ├── eval/ # 评测P07 评测体系 │ ├── test_suite.py # 测试用例 │ └── evaluator.py # 评估逻辑 ├── config/ # 配置 │ ├── prompts.py # System Prompt │ └── model_config.py # 模型配置 └── tests/ # 测试 └── test_workflow.py # 单元测试最小可运行版本如果不想用框架用 P01-P07 的知识也能搭# 最小可运行版本不用框架的完整工作流classSimpleWorkflow:基于 P01-P07 知识的简单工作流def__init__(self):self.state{}# 共享状态替代 ContextManagerself.messages[]# 对话历史P03self.memoryMemoryManager()# P05 记忆self.recoveryRecoveryManager()# P04 重试defrun(self,task:str):执行完整工作流# Step 1: 搜索search_resultsself._safe_search(task)self.state[search_results]search_results# Step 2: 分析analysisself._analyze(search_results)self.state[analysis]analysis# Step 3: 生成报告reportself._generate_report(analysis)self.state[report]report# Step 4: 人审approvedself._human_review(report)ifnotapproved:# P04 重试reportself._modify_report(report)self.state[report]report# Step 5: 持久化记忆self.memory.end_session(f完成了竞品分析任务:{task},key_events[{type:decision,detail:f分析完成:{task}}])returnreportdef_safe_search(self,task):P04 重试搜索失败自动重试try:returnself.recovery.retry_with_backoff(lambda:httpx.get(https://search.example.com/api,params{q:task}).json())exceptExceptionase:self.memory.record(error,str(e))return{error:str(e)}本章要点生产环境的最小架构节点函数 状态管理 工具调用 记忆系统 重试 人审。框架是加速器不是必需品。第五部分框架还是从零搭情况建议个人项目1-2 个工具用 P01 的 while 循环不需要框架团队项目3-5 个 Agent用 LangGraph 或 CrewAI需要时间旅行调试LangGraph快速出原型CrewAI需要人审拦截LangGraph需要持续评测集成任何框架 自建评测循环核心原则先理解原理P01-P07再选框架框架不增加智能只加速工程化能用简单循环解决的问题别上框架复杂度超过你当前团队的维护能力时考虑框架总结系列回顾从 P01 到 P08我们完成了 8 篇文章的旅程。回顾一下篇序核心知识点代码量P01决策循环Agent 的核心~20 行P02工具调用让 Agent 能做事~30 行P03上下文管理让 Agent 看清路~50 行P04重试策略让 Agent 不崩溃~60 行P05记忆系统让 Agent 记住事~80 行P06评测体系让 Agent 可度量~50 行P07持续评测让 Agent 越用越好~40 行P08工作流框架让 Agent 规模化~100 行总代码量约 430 行覆盖一个完整 Agent 工程的核心。记住三个原则原理 框架。框架是加速器不是替代品。简单 复杂。能用 while 循环解决的别上框架。原理 框架。理解了 P01-P07任何框架对你都是透明的。最后的问题你的项目里哪些组件已经落地了哪些还是以后再说思维导图Agent 工作流从原理到生产为什么需要框架5 个工程问题状态/回退/人审/调试/多 Agent框架不解决智能只解决工程化三大框架对比LangGraph状态机 持久化 时间旅行CrewAI角色扮演 快速上手AutoGen对话实验 多 Agent 协作实战项目竞品分析 Agent4 个节点搜索 → 分析 → 报告 → 人审条件路由状态机时间旅行调试架构规范节点函数 / 工具 / 记忆 / 评测 / 配置选型建议个人项目P01 循环团队项目LangGraph / CrewAI核心 Takeaways原理 框架简单 复杂理解 P01-P07任何框架都透明