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

---

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

---

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

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

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

但当你需要：

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

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

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

---

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

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

# P01 的基础 Agent 循环（能跑，但有问题）
def agent_loop(messages):
    while True:
        response = call_llm(messages)
        if response.content:
            return response.content
        if response.tool_calls:
            messages.extend(execute_tools(response.tool_calls))

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

问题	现象	框架能解决什么
状态不持久	服务重启后对话丢失	检查点保存/恢复
无法回退	某步出错只能从头开始	时间旅行调试
人工审核缺失	危险操作无确认就执行	人审拦截点
调试困难	不知道哪一步出错	全链路追踪
多 Agent 协作	复杂任务无法拆分子流程	工作流编排

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

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

---

第二部分：三大框架对比

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

维度	LangGraph	CrewAI	AutoGen
核心模型	状态机/图	角色扮演	对话流
适合场景	复杂控制流 + 生产级	多 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 的决策循环）

import json
from typing import TypedDict, Annotatedd
import operator

# ═══════════════════════════════════════════
# 状态定义：所有节点共享一个状态（P03 上下文管理）
# ═══════════════════════════════════════════
class AgentState(TypedDict):
    messages: list                    # 对话历史（P03 上下文）
    context: dict                     # 任务上下文
    research_results: dict            # 研究结果（P05 记忆）
    final_report: str                 # 最终报告
    errors: list                      # 错误记录（P04 重试）
    human_approved: bool              # 人工审核标志（人审拦截）

步骤 2：定义节点函数（每个节点 = 一个子 Agent）

import httpx
import openai

client = openai.OpenAI(base_url="YOUR_BASE_URL")

# ── 节点 1：搜索 ──
def search_agents(state: AgentState) -> AgentState:
    """收集竞品基本信息（P02 工具调用 + P04 重试）"""
    search_results = {}
    
    for name in state["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()
        except Exception as e:
            state["errors"].append(f"搜索 {name} 失败: {e}")
            search_results[name] = {"error": "search_failed"}
    
    return {"research_results": search_results}


# ── 节点 2：深度分析 ──
def analyze(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：报告生成 ──
def generate_report(state: AgentState) -> AgentState:
    """生成结构化报告（P05 记忆注入 + P07 评测验收条件）"""
    system_prompt = """
    你是一位技术顾问。根据分析报告生成对比报告。
    
    ## 格式要求
    - Markdown 格式
    - 包含对比表格
    - 包含选型建议
    
    ## 验收条件
    - 包含至少 3 个维度的对比
    - 包含明确的选型建议
    - 包含风险提示
    """
    
    # P05 记忆注入
    context = state["context"]
    if "user_profile" in context:
        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：人工审核 ──
def human_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：定义路由逻辑（条件边）

from langgraph.graph import StateGraph, END

# ── 条件路由函数 ──
def should_review(state: AgentState) -> str:
    """报告长度超过 2000 字需要人审"""
    if len(state["final_report"]) > 2000:
        return "human_review"
    return "done"


def is_approved(state: AgentState) -> str:
    """人审不通过则修改"""
    if not state.get("human_approved", False):
        return "modify_report"
    return "done"


def has_errors(state: AgentState) -> str:
    """有错误则重试"""
    if state.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：状态持久化和时间旅行调试

# ── 持久化：保存检查点（服务重启不丢状态） ──
from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
app_with_persistence = workflow.compile(checkpointer=checkpointer)

# 运行并保存状态
config = {"configurable": {"thread_id": "session_001"}}
result = app_with_persistence.invoke(input_state, config=config)

# ── 时间旅行：回退到上一步重新执行 ──
# 获取之前保存的 checkpoint
checkpoints = list(checkpointer.list(config))
for cp in checkpoints:
    print(f"Checkpoint: {cp.thread_id} @ {cp.metadata['ts']}")

# 回退到某个 checkpoint
checkpoint = 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 的知识也能搭：

# 最小可运行版本：不用框架的完整工作流
class SimpleWorkflow:
    """基于 P01-P07 知识的简单工作流"""
    
    def __init__(self):
        self.state = {}         # 共享状态（替代 ContextManager）
        self.messages = []      # 对话历史（P03）
        self.memory = MemoryManager()  # P05 记忆
        self.recovery = RecoveryManager()  # P04 重试
    
    def run(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)
        if not approved:
            # 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}"}]
        )
        
        return report
    
    def _safe_search(self, task):
        """P04 重试：搜索失败自动重试"""
        try:
            return self.recovery.retry_with_backoff(
                lambda: httpx.get("https://search.example.com/api", params={"q": task}).json()
            )
        except Exception as e:
            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 工程的核心。

记住三个原则：

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

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

---

思维导图

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