PDCA 循环执行
PDCA（Plan-Do-Check-Act） 是源自质量管理的经典循环方法论，在 Agent 系统中用于实现持续改进的执行流程。PDCA 让 Agent 具备"计划-执行-检查-处理"的闭环能力。
一、核心概念
1.1 PDCA 是什么？
PDCA 是四个阶段的循环：
┌─────────────────────────────────────────────────────────────┐
│ PDCA 循环 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────┐ │
│ ┌────│ P │────┐ │
│ │ │ Plan │ │ │
│ │ │ 计划 │ │ │
│ │ └─────────┘ │ │
│ │ │ │
│ ↓ ↓ │
│ ┌─────────┐ ┌─────────┐ │
│ │ A │ │ D │ │
│ │ Act │←────────│ Do │ │
│ │ 处理 │ │ 执行 │ │
│ └─────────┘ └─────────┘ │
│ │ │ │
│ │ ┌─────────┐ │ │
│ └────│ C │────┘ │
│ │ Check │ │
│ │ 检查 │ │
│ └─────────┘ │
│ │
│ 循环往复，持续改进 │
│ │
└─────────────────────────────────────────────────────────────┘1.2 四个阶段详解
| 阶段 | 英文 | 核心任务 | Agent 应用 |
|---|---|---|---|
| Plan | 计划 | 制定执行方案 | 任务分解、步骤规划 |
| Do | 执行 | 按计划执行 | 调用工具、执行步骤 |
| Check | 检查 | 验证执行结果 | 结果验证、质量评估 |
| Act | 处理 | 根据检查结果调整 | 问题修复、经验总结 |
1.3 PDCA 与 Agent 的结合
┌─────────────────────────────────────────────────────────────┐
│ Agent 中的 PDCA │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ Plan（计划） │ │
│ │ │ │
│ │ • 分析任务目标 │ │
│ │ • 分解为子任务 │ │
│ │ • 制定执行步骤 │ │
│ │ • 预估资源和风险 │ │
│ │ │ │
│ │ 输出：执行计划 [step1, step2, step3, ...] │ │
│ └─────────────────────────────────────────────────────┘ │
│ │ │
│ ↓ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ Do（执行） │ │
│ │ │ │
│ │ • 按计划调用工具 │ │
│ │ • 记录执行过程 │ │
│ │ • 处理执行异常 │ │
│ │ • 收集执行数据 │ │
│ │ │ │
│ │ 输出：执行结果、执行日志 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │ │
│ ↓ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ Check（检查） │ │
│ │ │ │
│ │ • 验证结果是否符合预期 │ │
│ │ • 评估执行质量 │ │
│ │ • 分析偏差原因 │ │
│ │ • 检查是否有遗漏 │ │
│ │ │ │
│ │ 输出：检查报告、偏差分析 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │ │
│ ↓ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ Act（处理） │ │
│ │ │ │
│ │ • 根据检查结果决定下一步 │ │
│ │ - 通过：进入下一阶段 │ │
│ │ - 失败：调整计划，重新执行 │ │
│ │ - 部分成功：修复问题，补充执行 │ │
│ │ • 总结经验教训 │ │
│ │ • 更新执行策略 │ │
│ │ │ │
│ │ 输出：下一轮计划或最终结果 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘二、工作流程
2.1 完整 PDCA 工作流程
┌─────────────────────────────────────────────────────────────────────┐
│ PDCA 完整工作流程 │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────┐ │
│ │ 用户输入 │ │
│ │ (Query) │ │
│ └──────┬──────┘ │
│ │ │
│ ↓ │
│ ╔═════════════════════════════════════════════════════════════╗ │
│ ║ PDCA 循环 ║ │
│ ║ ║ │
│ ║ ┌─────────────────────────────────────────────────────┐ ║ │
│ ║ │ Plan: 生成执行计划 │ ║ │
│ ║ │ plan = planner.generate_plan(task, context) │ ║ │
│ ║ └─────────────────────────────────────────────────────┘ ║ │
│ ║ │ ║ │
│ ║ ↓ ║ │
│ ║ ┌─────────────────────────────────────────────────────┐ ║ │
│ ║ │ Do: 执行计划 │ ║ │
│ ║ │ result = executor.execute(plan) │ ║ │
│ ║ └─────────────────────────────────────────────────────┘ ║ │
│ ║ │ ║ │
│ ║ ↓ ║ │
│ ║ ┌─────────────────────────────────────────────────────┐ ║ │
│ ║ │ Check: 验证结果 │ ║ │
│ ║ │ check = validator.check(result, criteria) │ ║ │
│ ║ └─────────────────────────────────────────────────────┘ ║ │
│ ║ │ ║ │
│ ║ ↓ ║ │
│ ║ ┌─────────────────────────────────────────────────────┐ ║ │
│ ║ │ Act: 根据检查结果决定 │ ║ │
│ ║ │ if check.passed: │ ║ │
│ ║ │ return result # 完成 │ ║ │
│ ║ │ else: │ ║ │
│ ║ │ plan = adjuster.adjust(plan, check.feedback) │ ║ │
│ ║ │ continue # 继续循环 │ ║ │
│ ║ └─────────────────────────────────────────────────────┘ ║ │
│ ║ ║ │
│ ╚═════════════════════════════════════════════════════════════╝ │
│ │ │
│ ↓ │
│ ┌─────────────────────┐ │
│ │ 返回最终结果 │ │
│ └─────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────┘2.2 状态转换
┌─────────────────────────────────────────────────────────────┐
│ PDCA 状态转换 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────┐ │
│ ┌─────────│ START │─────────┐ │
│ │ └────┬────┘ │ │
│ │ │ │ │
│ ↓ ↓ ↓ │
│ ┌────────┐ ┌──────────┐ ┌──────────┐ │
│ │ PLAN │ │ DO │ │ CHECK │ │
│ │ 计划 │───→│ 执行 │──→│ 检查 │ │
│ └────────┘ └──────────┘ └────┬─────┘ │
│ ↑ │ │
│ │ │ │
│ │ ┌──────────────┼──────────────┐ │
│ │ ↓ ↓ ↓ │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐│
│ │ │ 通过 │ │ 部分通过 │ │ 失败 ││
│ │ └────┬─────┘ └────┬─────┘ └────┬─────┘│
│ │ │ │ │ │
│ │ ↓ ↓ │ │
│ │ ┌──────────┐ ┌──────────┐ │ │
│ │ │ 完成 │ │ 补充执行 │ │ │
│ │ │ END │ └──────────┘ │ │
│ │ ┌────┘ │
│ │ ↓ │
│ └───────────────────────────────┌──────────┐ │
│ │ ACT │ │
│ │ 调整 │ │
│ └──────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘三、代码实现
3.1 基础 PDCA 实现
"""
PDCA 循环执行器基础实现
"""
from typing import List, Optional, Dict, Any
from dataclasses import dataclass
from enum import Enum
from langchain_openai import ChatOpenAI
class PDCAState(Enum):
"""PDCA 状态"""
PLAN = "plan"
DO = "do"
CHECK = "check"
ACT = "act"
COMPLETED = "completed"
class CheckResult(Enum):
"""检查结果"""
PASSED = "passed" # 通过
PARTIAL = "partial" # 部分通过
FAILED = "failed" # 失败
@dataclass
class ExecutionPlan:
"""执行计划"""
steps: List[str] # 执行步骤
current_step: int = 0 # 当前步骤
context: Dict[str, Any] = None # 上下文
@dataclass
class CheckReport:
"""检查报告"""
result: CheckResult # 检查结果
passed: bool # 是否通过
issues: List[str] # 问题列表
feedback: str # 反馈信息
class Plannner:
"""计划器"""
PLAN_PROMPT = """
请为以下任务制定执行计划。
任务：{task}
请输出：
1. 任务分析
2. 执行步骤（编号列表）
3. 预期结果
格式：
分析：...
步骤：
1. ...
2. ...
预期：...
"""
def __init__(self, model_name: str = "gpt-4"):
self.llm = ChatOpenAI(model=model_name, temperature=0)
def generate_plan(self, task: str) -> ExecutionPlan:
"""生成执行计划"""
prompt = self.PLAN_PROMPT.format(task=task)
response = self.llm.invoke(prompt)
# 解析步骤
steps = []
in_steps = False
for line in response.content.split('\n'):
line = line.strip()
if line.startswith('步骤'):
in_steps = True
continue
if in_steps and line and line[0].isdigit():
step = line.split('.', 1)[1].strip() if '.' in line else line
steps.append(step)
elif in_steps and line.startswith('预期'):
break
return ExecutionPlan(steps=steps, context={})
class Executor:
"""执行器"""
def __init__(self, model_name: str = "gpt-4"):
self.llm = ChatOpenAI(model=model_name, temperature=0)
self.execution_log: List[Dict[str, Any]] = []
def execute(self, plan: ExecutionPlan) -> str:
"""执行计划"""
results = []
for i, step in enumerate(plan.steps):
prompt = f"""
请执行以下步骤：
步骤：{step}
上下文：
{self._format_context(plan.context)}
请输出执行结果。
"""
response = self.llm.invoke(prompt)
# 记录执行日志
self.execution_log.append({
"step": i + 1,
"action": step,
"result": response.content
})
results.append(f"步骤 {i+1}: {step}\n结果: {response.content}\n")
# 更新上下文
plan.context[f"step_{i+1}_result"] = response.content
plan.current_step = len(plan.steps)
return "\n".join(results)
def _format_context(self, context: Dict[str, Any]) -> str:
"""格式化上下文"""
if not context:
return "（开始）"
return "\n".join(f"- {k}: {v}" for k, v in context.items())
class Validator:
"""验证器"""
CHECK_PROMPT = """
请检查以下执行结果是否符合预期。
任务：{task}
执行结果：{result}
检查标准：
1. 是否完成了任务目标？
2. 结果是否符合预期格式？
3. 是否有明显错误？
请输出：
状态：通过/部分通过/失败
问题：（如有）
建议：（如有）
"""
def __init__(self, model_name: str = "gpt-4"):
self.llm = ChatOpenAI(model=model_name, temperature=0)
def check(self, task: str, result: str) -> CheckReport:
"""检查执行结果"""
prompt = self.CHECK_PROMPT.format(task=task, result=result)
response = self.llm.invoke(prompt)
content = response.content
# 解析结果
if "通过" in content and "失败" not in content and "部分" not in content:
check_result = CheckResult.PASSED
passed = True
elif "部分通过" in content:
check_result = CheckResult.PARTIAL
passed = False
else:
check_result = CheckResult.FAILED
passed = False
# 提取问题
issues = []
for line in content.split('\n'):
if '问题' in line and '：' in line:
issue = line.split('：', 1)[1].strip()
if issue:
issues.append(issue)
return CheckReport(
result=check_result,
passed=passed,
issues=issues,
feedback=content
)
class Adjuster:
"""调整器"""
ADJUST_PROMPT = """
根据检查结果调整执行计划。
原计划：
{original_plan}
检查结果：
{check_result}
请分析问题原因，并给出调整后的执行计划。
调整后计划：
1. ...
2. ...
"""
def __init__(self, model_name: str = "gpt-4"):
self.llm = ChatOpenAI(model=model_name, temperature=0.7)
def adjust(self, plan: ExecutionPlan, check: CheckReport) -> ExecutionPlan:
"""调整计划"""
original_plan = "\n".join(f"{i+1}. {s}" for i, s in enumerate(plan.steps))
prompt = self.ADJUST_PROMPT.format(
original_plan=original_plan,
check_result=check.feedback
)
response = self.llm.invoke(prompt)
# 解析调整后的步骤
new_steps = []
for line in response.content.split('\n'):
line = line.strip()
if line and line[0].isdigit() and '.' in line:
step = line.split('.', 1)[1].strip()
new_steps.append(step)
return ExecutionPlan(
steps=new_steps if new_steps else plan.steps,
context=plan.context
)
class PDCAAgent:
"""
PDCA Agent
实现 Plan-Do-Check-Act 循环
"""
def __init__(self, model_name: str = "gpt-4", max_cycles: int = 3):
self.planner = Plannner(model_name)
self.executor = Executor(model_name)
self.validator = Validator(model_name)
self.adjuster = Adjuster(model_name)
self.max_cycles = max_cycles
def run(self, task: str) -> dict:
"""
执行 PDCA 循环
Args:
task: 任务描述
Returns:
执行结果
"""
cycle = 0
plan = None
result = None
check = None
while cycle < self.max_cycles:
cycle += 1
# Plan
if plan is None:
plan = self.planner.generate_plan(task)
# Do
result = self.executor.execute(plan)
# Check
check = self.validator.check(task, result)
# Act
if check.passed:
return {
"task": task,
"success": True,
"result": result,
"cycles": cycle,
"plan": plan.steps,
"check": check.feedback
}
# 调整计划
plan = self.adjuster.adjust(plan, check)
return {
"task": task,
"success": False,
"result": result,
"cycles": cycle,
"plan": plan.steps if plan else [],
"check": check.feedback if check else ""
}
# 使用示例
if __name__ == "__main__":
agent = PDCAAgent(max_cycles=3)
result = agent.run(
"写一个 Python 函数实现快速排序算法"
)
print(f"任务：{result['task']}")
print(f"成功：{result['success']}")
print(f"循环次数：{result['cycles']}")
print(f"\n执行计划：")
for i, step in enumerate(result['plan'], 1):
print(f" {i}. {step}")
print(f"\n最终结果：\n{result['result'][:500]}...")3.2 使用 LangGraph 实现
"""
使用 LangGraph 实现 PDCA 循环
"""
from typing import TypedDict, List, Optional
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
class PDCAState(TypedDict):
"""PDCA 状态"""
task: str
plan: List[str]
current_step: int
execution_result: str
check_passed: bool
check_feedback: str
cycle: int
max_cycles: int
def create_pdca_graph():
"""创建 PDCA 图"""
llm = ChatOpenAI(model="gpt-4", temperature=0)
# Plan 节点
def plan_node(state: PDCAState) -> PDCAState:
"""生成执行计划"""
prompt = f"""
为以下任务制定执行计划（3-5步）：
任务：{state['task']}
请输出编号步骤：
1. ...
2. ...
"""
response = llm.invoke(prompt)
# 解析步骤
steps = []
for line in response.content.split('\n'):
line = line.strip()
if line and line[0].isdigit() and '.' in line:
step = line.split('.', 1)[1].strip()
steps.append(step)
return {
**state,
"plan": steps if steps else ["执行任务"],
"cycle": state["cycle"] + 1
}
# Do 节点
def do_node(state: PDCAState) -> PDCAState:
"""执行计划"""
steps_text = "\n".join(f"{i+1}. {s}" for i, s in enumerate(state["plan"]))
prompt = f"""
请按计划执行任务：
任务：{state['task']}
计划：
{steps_text}
请执行并输出结果。
"""
response = llm.invoke(prompt)
return {
**state,
"execution_result": response.content
}
# Check 节点
def check_node(state: PDCAState) -> PDCAState:
"""检查结果"""
prompt = f"""
检查执行结果：
任务：{state['task']}
结果：{state['execution_result']}
是否通过？请回答：通过/失败
"""
response = llm.invoke(prompt)
passed = "通过" in response.content and "失败" not in response.content
return {
**state,
"check_passed": passed,
"check_feedback": response.content
}
# Act 节点
def act_node(state: PDCAState) -> PDCAState:
"""处理结果"""
if state["check_passed"]:
return state
# 生成调整建议
prompt = f"""
根据检查结果调整计划：
原计划：{state['plan']}
检查反馈：{state['check_feedback']}
请给出调整后的计划步骤。
"""
response = llm.invoke(prompt)
# 解析新步骤
new_steps = []
for line in response.content.split('\n'):
line = line.strip()
if line and line[0].isdigit() and '.' in line:
step = line.split('.', 1)[1].strip()
new_steps.append(step)
return {
**state,
"plan": new_steps if new_steps else state["plan"]
}
# 条件判断
def should_continue(state: PDCAState) -> str:
"""判断是否继续"""
if state["check_passed"]:
return "end"
if state["cycle"] >= state["max_cycles"]:
return "end"
return "continue"
# 创建图
workflow = StateGraph(PDCAState)
# 添加节点
workflow.add_node("plan", plan_node)
workflow.add_node("do", do_node)
workflow.add_node("check", check_node)
workflow.add_node("act", act_node)
# 设置入口
workflow.set_entry_point("plan")
# 添加边
workflow.add_edge("plan", "do")
workflow.add_edge("do", "check")
workflow.add_conditional_edges(
"check",
should_continue,
{
"end": END,
"continue": "act"
}
)
workflow.add_edge("act", "plan")
return workflow.compile()
# 使用示例
if __name__ == "__main__":
graph = create_pdca_graph()
result = graph.invoke({
"task": "实现一个二分查找算法",
"plan": [],
"current_step": 0,
"execution_result": "",
"check_passed": False,
"check_feedback": "",
"cycle": 0,
"max_cycles": 3
})
print(f"任务：{result['task']}")
print(f"成功：{result['check_passed']}")
print(f"循环次数：{result['cycle']}")四、PDCA 的应用场景
4.1 适用场景
| 场景 | PDCA 价值 | 示例 |
|---|---|---|
| 质量敏感任务 | 确保结果符合质量标准 | 代码生成、文档撰写 |
| 迭代改进任务 | 需要多轮优化 | 方案设计、创意写作 |
| 复杂流程任务 | 需要检查验证 | 数据处理管道 |
| 学习型任务 | 需要从失败中学习 | 新领域探索 |
4.2 PDCA 与其他方法的关系
┌─────────────────────────────────────────────────────────────┐
│ PDCA 与其他方法的关系 │
├─────────────────────────────────────────────────────────────┤
│ │
│ PDCA 与 Plan-Execute： │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ Plan-Execute 是 PDCA 的简化版 │ │
│ │ │ │
│ │ Plan-Execute：Plan → Execute → Done │ │
│ │ PDCA：Plan → Do → Check → Act → (循环) │ │
│ │ │ │
│ │ PDCA 多了 Check 和 Act 阶段，支持迭代改进 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ PDCA 与 Reflexion： │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ Reflexion 是 PDCA 的特化实现 │ │
│ │ │ │
│ │ Reflexion：Actor → Evaluator → Reflector │ │
│ │ PDCA：Plan → Do → Check → Act │ │
│ │ │ │
│ │ Reflexion 专注于反思学习，PDCA 更通用 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ PDCA 与 ReAct： │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ ReAct 是 PDCA 的原子化实现 │ │
│ │ │ │
│ │ ReAct：Thought → Action → Observation（每步循环） │ │
│ │ PDCA：整个任务级别的循环 │ │
│ │ │ │
│ │ 可以组合使用：ReAct 做 Do，PDCA 做整体控制 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘五、PDCA 的优化策略
5.1 效率优化
| 策略 | 描述 | 效果 |
|---|---|---|
| 提前终止 | 检查通过后立即结束 | 减少不必要的循环 |
| 增量执行 | 只执行未完成的部分 | 减少重复执行 |
| 并行检查 | 多个检查标准并行验证 | 加速 Check 阶段 |
| 缓存复用 | 复用之前的结果 | 避免重复计算 |
5.2 质量优化
┌─────────────────────────────────────────────────────────────┐
│ PDCA 质量优化策略 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 1. 多维度检查 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ Check 阶段检查多个维度： │ │
│ │ • 功能正确性：是否完成核心功能 │ │
│ │ • 格式规范性：是否符合输出格式 │ │
│ │ • 边界完整性：是否处理边界情况 │ │
│ │ • 性能合理性：是否满足性能要求 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 2. 渐进式改进 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 每轮 Act 改进一个方面： │ │
│ │ 第1轮：修复核心功能错误 │ │
│ │ 第2轮：完善边界处理 │ │
│ │ 第3轮：优化代码质量 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 3. 经验积累 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 将每轮 PDCA 的经验存入记忆： │ │
│ │ • 成功模式：记录有效的计划模式 │ │
│ │ • 失败模式：记录应避免的做法 │ │
│ │ • 调整策略：记录有效的调整方法 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘六、面试高频问题
Q1: PDCA 循环的核心思想是什么？
答案要点：
- Plan：制定计划
- Do：执行计划
- Check：检查结果
- Act：处理改进
- 形成闭环，持续改进
Q2: PDCA 与 Plan-Execute 的区别？
答案要点：
- Plan-Execute：一次性执行，无反馈机制
- PDCA：循环执行，有检查和调整
- PDCA 支持迭代改进，Plan-Execute 不支持
- PDCA 适合质量敏感任务
Q3: 如何避免 PDCA 无限循环？
答案要点：
- 设置最大循环次数
- 检测无进展（多轮结果无变化）
- 设置递减的改进阈值
- 失败后降级到其他策略
Q4: PDCA 适用于哪些场景？
答案要点：
- 质量敏感任务（代码生成、文档撰写）
- 需要迭代改进的任务（方案设计）
- 复杂流程任务（数据处理管道）
- 学习型任务（新领域探索）
Q5: 如何衡量 PDCA 的效果？
答案要点：
- 任务成功率
- 平均循环次数
- 资源消耗（Token 使用量）
- 改进幅度（每轮的质量提升）
七、小结
| 概念 | 一句话总结 | 面试关键词 |
|---|---|---|
| PDCA | 计划-执行-检查-处理的循环 | 持续改进、闭环 |
| Plan | 制定执行方案 | 任务分解、步骤规划 |
| Check | 验证执行结果 | 结果验证、质量评估 |
| Act | 根据检查结果调整 | 问题修复、经验总结 |
一句话总结：PDCA 是源自质量管理的经典方法论，让 Agent 具备"计划-执行-检查-处理"的闭环能力，实现持续改进。
最后更新：2026年3月19日