阅读时间：约 15 分钟
前置知识：理解 Agent 决策循环（P01-P07）

前七篇文章，你学完了：决策循环、工具调用、上下文管理、重试策略、记忆系统、评测……但每篇文章只讲一个点。现在把它们串起来：一个真实的、端到端的 Agent 工作流长什么样？

前言：你不需要框架，但你最终需要工作流

先说一个反直觉的话：大部分 Agent 项目不需要框架。

你完全可以用 P01 的 while 循环、P03 的 ContextManager、P04 的 RecoveryManager、P05 的 MemoryManager，从零搭一个能跑的 Agent。

但当你需要：

• 状态持久化（服务重启不丢上下文）
• 时间旅行调试（回退到上一步重新执行）
• 多人协作（多个 Agent 各司其职）
• 人工审核（关键步骤暂停等人确认）

框架才开始有价值。它不解决你的 Agent 智能问题：这些你前七篇都学过了：它解决的是工程化问题：可调试、可维护、可扩展。

📌本章核心：框架不是必需品，但复杂项目离不开工作流引擎。先理解原理，再选框架。

第一部分：为什么你需要工作流框架

从零搭 Agent 的天花板在哪里？以 P01 的基础 while 循环为例：

# P01 的基础 Agent 循环（能跑，但有问题）defagent_loop(messages):whileTrue:response=call_llm(messages)ifresponse.content:returnresponse.contentifresponse.tool_calls:messages.extend(execute_tools(response.tool_calls))

这个循环能跑，但有 5 个生产环境问题：

框架解决的不是"Agent 智能不够"，而是"Agent 跑不好"。

📌本章要点：工作流框架不增加智能，它解决 5 个工程问题：状态持久化、时间旅行调试、人工审核、全链路追踪、多 Agent 协作。

第二部分：三大框架对比

2026 年 Agent 框架三巨头：LangGraph、CrewAI、AutoGen。选哪个？

一句话选型：

• 要复杂控制和生产级能力 →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：定义节点函数（每个节点 = 一个子 Agent）

importhttpximportopenai client=openai.OpenAI(base_url="YOUR_BASE_URL")# ── 节点 1：搜索 ──defsearch_agents(state:AgentState)->AgentState:"""收集竞品基本信息（P02 工具调用 + P04 重试）"""search_results={}fornameinstate["context"].get("competitors",[]):# P04 重试：搜索失败自动重试try:resp=httpx.get("https://search.example.com/api",params={"q":f"{name}技术分析"},timeout=5.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 工具选择）"""research=state["research_results"]# P02 工具：选择正确的分析工具system_prompt=""" 你是技术分析师。根据收集的信息，从以下维度对比： - 架构设计 - 性能指标 - 适用场景 - 优缺点 """messages=[{"role":"system","content":system_prompt},{"role":"user","content":json.dumps(research,ensure_ascii=False,indent=2)}]# 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"}}}}}}]response=client.chat.completions.create(model="deepseek-v4-flash",messages=messages,tools=tools)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 记忆注入context=state["context"]if"user_profile"incontext:system_prompt+=f"\n\n## 用户背景\n{context['user_profile']}"messages=[{"role":"system","content":system_prompt},{"role":"user","content":f"""根据以下分析生成报告：{state['research_results'].get('analysis','')}竞品列表：{context.get('competitors',[])}"""}]response=client.chat.completions.create(model="deepseek-v4-flash",messages=messages)return{"final_report":response.choices[0].message.content}# ── 节点 4：人工审核 ──defhuman_review(state:AgentState)->AgentState:""" 关键步骤人工审核（人审拦截点） 在 LangGraph 中，这个节点会暂停执行，等待用户回复。 这里是生产环境的关键安全机制。 """report=state["final_report"]print("="*50)print("📋 报告审核")print(report[:300])print("..."*10)print("="*50)print("回复 '确认' 以继续，或提出修改意见。")# 实际生产环境这里等待用户输入# 不是调用 LLM，而是等待真人回复approval=input("是否确认？(确认/修改): ").strip()return{"human_approved":approval=="确认"}

步骤 3：定义路由逻辑（条件边）

fromlanggraph.graphimportStateGraph,END# ── 条件路由函数 ──defshould_review(state:AgentState)->str:"""报告长度超过 2000 字需要人审"""iflen(state["final_report"])>2000:return"human_review"return"done"defis_approved(state:AgentState)->str:"""人审不通过则修改"""ifnotstate.get("human_approved",False):return"modify_report"return"done"defhas_errors(state:AgentState)->str:"""有错误则重试"""ifstate.get("errors"):return"retry"return"analyze"# ── 构建图 ──workflow=StateGraph(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})# 编译app=workflow.compile(interrupt_before=["human_review"]# 人审节点前自动暂停)

步骤 4：状态持久化和时间旅行调试

# ── 持久化：保存检查点（服务重启不丢状态） ──fromlanggraph.checkpoint.memoryimportMemorySaver checkpointer=MemorySaver()app_with_persistence=workflow.compile(checkpointer=checkpointer)# 运行并保存状态config={"configurable":{"thread_id":"session_001"}}result=app_with_persistence.invoke(input_state,config=config)# ── 时间旅行：回退到上一步重新执行 ──# 获取之前保存的 checkpointcheckpoints=list(checkpointer.list(config))forcpincheckpoints:print(f"Checkpoint:{cp.thread_id}@{cp.metadata['ts']}")# 回退到某个 checkpointcheckpoint=checkpoints[-2]# 倒数第二个state_at_checkpoint=checkpointer.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={}# 共享状态（替代 ContextManager）self.messages=[]# 对话历史（P03）self.memory=MemoryManager()# P05 记忆self.recovery=RecoveryManager()# P04 重试defrun(self,task:str):"""执行完整工作流"""# Step 1: 搜索search_results=self._safe_search(task)self.state["search_results"]=search_results# Step 2: 分析analysis=self._analyze(search_results)self.state["analysis"]=analysis# Step 3: 生成报告report=self._generate_report(analysis)self.state["report"]=report# Step 4: 人审approved=self._human_review(report)ifnotapproved:# P04 重试report=self._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)}

📌本章要点：生产环境的最小架构：节点函数 + 状态管理 + 工具调用 + 记忆系统 + 重试 + 人审。框架是加速器，不是必需品。

第五部分：框架还是从零搭？

核心原则：

• 先理解原理（P01-P07），再选框架
• 框架不增加智能，只加速工程化
• 能用简单循环解决的问题，别上框架
• 复杂度超过你当前团队的维护能力时，考虑框架

总结：系列回顾

从 P01 到 P08，我们完成了 8 篇文章的旅程。回顾一下：

总代码量约 430 行，覆盖一个完整 Agent 工程的核心。

记住三个原则：

1. 原理 > 框架。框架是加速器，不是替代品。
2. 简单 > 复杂。能用 while 循环解决的，别上框架。
3. 原理 > 框架。理解了 P01-P07，任何框架对你都是透明的。

🤔最后的问题：你的项目里，哪些组件已经落地了？哪些还是"以后再说"？

思维导图

• Agent 工作流：从原理到生产
  • 为什么需要框架
    • 5 个工程问题（状态/回退/人审/调试/多 Agent）
    • 框架不解决智能，只解决工程化
  • 三大框架对比
    • LangGraph：状态机 + 持久化 + 时间旅行
    • CrewAI：角色扮演 + 快速上手
    • AutoGen：对话实验 + 多 Agent 协作
  • 实战项目
    • 竞品分析 Agent
    • 4 个节点（搜索 → 分析 → 报告 → 人审）
    • 条件路由（状态机）
    • 时间旅行调试
  • 架构规范
    • 节点函数 / 工具 / 记忆 / 评测 / 配置
  • 选型建议
    • 个人项目：P01 循环
    • 团队项目：LangGraph / CrewAI
  • 核心 Takeaways
    • 原理 > 框架
    • 简单 > 复杂
    • 理解 P01-P07，任何框架都透明