◆ 博主名称： QuZhengRong
AI俘虏，样式苦手
目录
- 一、ReAct 模式：先观察，后行动
- 二、Plan-and-Execute 模式：先拆分任务，再按计划执行
- 三、Reflection 模式：交付前按标准检查
- 四、总结
- 五、LuckReport 项目推荐
通常情况下，Agent 很少能仅靠一轮对话就完成任务，例如排查一个经营问题要查多次数据，做一份调研要分几个方向收集材料。当执行任务的步骤变多时，自然而然就会出现一个问题：任务该怎么推进？
设计模式就是对这个问题的回答。Agent 常用的设计模式有三种：ReAct ，Plan-and-Execute ，Reflection 。下面来分别介绍这3个模式
一、ReAct 模式：先观察，后行动
ReAct 模式和人类解决问题的思考方式高度一致，过程如下：
Thought（思考当前缺什么信息）
↓
Action（调用工具）
↓
Observation（拿到结果）
↓
下一轮 Thought，直到证据足够给出结论
ReAct 模式的推进方式是每观察一次数据再决定下一步该怎么做。它适合排查类任务：门店为什么亏损、程序为什么有 BUG。这类任务事先不知道问题在哪，只能走一步看一步
以勇哥说餐饮的场景为例：深圳南山店这个月亏了，让 Agent 排查原因
# ReAct 模式：手写 Thought -> Action -> Observation 循环，演示 Agent 边想边做、看一步走一步
# 主题：勇哥说餐饮——排查"深圳南山店这个月为什么亏了"，排查路径不确定，正适合 ReAct
from dotenv import load_dotenv
import os
from langchain.tools import tool
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage, ToolMessage
load_dotenv()
# 1. 定义四个排查工具（模拟数据，数据之间留了推理线索，等 Agent 自己串起来）
@tool
def get_month_revenue(store_name: str) -> str:
"""查询指定门店本月的营业额及环比变化。"""
return f"{store_name}：本月营业额 21 万，环比下降 18%"
@tool
def get_cost_detail(store_name: str) -> str:
"""查询指定门店本月的各项成本明细。"""
return f"{store_name}：房租 4.5 万、人工 5.2 万、食材 6.8 万、水电杂费 0.8 万"
@tool
def get_month_traffic(store_name: str) -> str:
"""查询指定门店本月的到店客流量及环比变化。"""
return f"{store_name}：本月到店客流 6200 人次，环比下降 25%"
@tool
def get_nearby_change(store_name: str) -> str:
"""查询指定门店周边最近的经营环境变化。"""
return f"{store_name}：300 米内新开了 2 家连锁快餐店，其中一家正在做开业大促"
tools = [get_month_revenue, get_cost_detail, get_month_traffic, get_nearby_change]
# 工具名 -> 函数 的映射表，方便按名字调用
tool_map = {
t.name: t for t in tools}
# 2. System Prompt 是 ReAct 的灵魂
# 强制模型每一步先说思考（Thought）再调工具（Action），拿到观察（Observation）后继续判断
# 并且明确"不要一开始把所有工具都调一遍"，这才是看一步走一步
SYSTEM_PROMPT = """你是餐饮老板勇哥的排查助手，任务是排查门店经营问题。
工作方式（ReAct 循环）：
1. 每次调用工具前，必须先用一两句话说明这一步的思考：为什么查这个、想验证什么
2. 拿到工具结果后，根据结果决定下一步查什么，禁止一开始就把所有工具全调一遍
3. 证据足够时才给结论，结论必须说明依据了哪些数据
成本判断红线：房租不超营业额 15%，人工不超 20%，食材不超 30%，超标即有问题。"""
# 3. 初始化模型并绑定工具
llm = ChatOpenAI(
model="qwen3.7-max-2026-06-08",
api_key=os.getenv("API_KEY"),
base_url=os.getenv("BASE_URL")
)
llm_with_tools = llm.bind_tools(tools)
def run_react_agent(user_input: str, max_steps: int = 8) -> str:
"""运行 ReAct 循环，打印 Thought/Action/Observation 轨迹，返回最终回答。
Args:
user_input: 用户输入的排查任务
max_steps: 最大步数限制，防止 Agent 跑散后无限调工具
Returns:
模型生成的最终回答文本
"""
messages = [
SystemMessage(content=SYSTEM_PROMPT),
HumanMessage(content=user_input),
]
for step in range(1, max_steps + 1):
# 第一步：模型思考当前缺什么信息，决定要不要调工具
response = llm_with_tools.invoke(messages)
messages.append(response)
# 模型这一步的思考就是 Thought（content 为空说明模型没按规矩说思考，兜底提示一下）
thought = response.content.strip() if response.content else "（模型未输出文字思考，直接行动）"
print(f"\n[第 {step} 步 Thought] {thought}")
# 没有 tool_calls 说明模型认为证据够了，给出最终结论，循环结束
if not response.tool_calls:
print(f"\n[最终回答]\n{response.content}")
return response.content
# 第二步：执行模型要求的工具（Action），把结果喂回去（Observation）
for call in response.tool_calls:
print(f"[Action] {call['name']}({call['args']})")
result = tool_map[call["name"]].invoke(call["args"])
print(f"[Observation] {result}")
# ToolMessage 的 tool_call_id 要和模型给的对应上，模型才知道这是哪个工具的返回
messages.append(ToolMessage(
content=str(result),
tool_call_id=call["id"]
))
# 回到循环顶部：带着 Observation，让模型想下一步
print("\n达到最大步数，强制结束。")
return "达到最大步数，强制结束。"
# 4. 跑一个排查任务：亏钱原因事先不知道，只能查一步看一步
if __name__ == "__main__":
run_react_agent("深圳南山店这个月亏了，帮我排查一下亏在哪。")
运行轨迹：
[第 1 步 Thought] 先查本月营业额，确认亏损幅度和环比变化
[Action] get_month_revenue(store_name='深圳南山店')
[Observation] 深圳南山店：本月营业额 21 万，环比下降 18%
[第 2 步 Thought] 营收降了，查成本明细，看是不是成本占比超标
[Action] get_cost_detail(store_name='深圳南山店')
[Observation] 房租 4.5 万、人工 5.2 万、食材 6.8 万、水电杂费 0.8 万
[第 3 步 Thought] 食材 6.8/21≈32% 超了 30% 红线，但营收本身也在降，查客流确认原因
[Action] get_month_traffic(store_name='深圳南山店')
[Observation] 本月到店客流 6200 人次，环比下降 25%
[第 4 步 Thought] 客流掉了四分之一，查周边环境有没有变化
[Action] get_nearby_change(store_name='深圳南山店')
[Observation] 300 米内新开了 2 家连锁快餐店，其中一家正在做开业大促
[最终回答] 亏损主因是客流下降 25% 导致营业额环比降 18%，周边新开竞对分流是直接原因；同时食材成本占比约 32% 超过 30% 红线，成本端也有问题……
任务推进顺序：查了营收才想查成本，算出占比超标才想查客流，看到客流下降才想查周边。每一步都由上一步的结果来决定。
React 模式的一些注意点：
1、必须有 max_steps 最长步数。ReAct 容易跑偏，忘掉原始目标、在失败结果里反复重试、信息足够了还继续查。没有步数上限，一旦跑偏就会持续消耗 token 。
2、System Prompt 里禁止一开始把所有工具全调一遍。不约束的话，模型经常第一轮就调完四个工具，逐步推理就失去意义了。
3、要求结论说明依据了哪些数据。这是防幻觉的措施，强制模型基于 Observation 分析。
二、Plan-and-Execute 模式：先拆分任务，再按 计划执行
Plan-and-Execute 把执行任务的过程分成两个部分：Planner 先把任务拆成步骤列表，Executor 按计划逐步执行，最后汇总结果。过程如下
用户目标
↓
Planner 拆成步骤列表（结构化输出，可解析）
↓
Executor 逐步执行，每步可调工具，记录每步结果
↓
带着全部步骤结果生成最终交付物
ReAct 是一边想一边做，Plan-and-Execute 是先规划后执行。Executor 执行任务时始终看得见完整计划，不容易分散模型注意力。
举一个分步调研的例子：勇哥要在深圳开一家 100 平米的新面馆，让 Agent 做开店前调研并输出简报。
# Plan-and-Execute 模式：先让 Planner 拆任务，再让 Executor 逐步执行，演示"先规划后执行"
# 主题：勇哥说餐饮——勇哥要在深圳开新面馆，让 Agent 先拆调研步骤，再逐步执行输出简报
from dotenv import load_dotenv
import os
from langchain.tools import tool
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage, ToolMessage
from pydantic import BaseModel, Field
load_dotenv()
llm = ChatOpenAI(
model="qwen3.7-max-2026-06-08",
api_key=os.getenv("API_KEY"),
base_url=os.getenv("BASE_URL")
)
# 1. Planner 的输出结构：用结构化输出保证计划可解析、可逐条执行
# 没有结构化约束，模型可能把计划写成一大段散文，程序没法按步骤推进
class Plan(BaseModel):
steps: list[str] = Field(description="按顺序执行的调研步骤，每步一句话，控制在 3 到 5 步")
# 2. 定义四个调研工具（模拟数据），供 Executor 执行步骤时调用
@tool
def search_market(city: str, category: str) -> str:
"""查询指定城市某品类餐饮的整体市场行情。"""
return f"{city}{category}：市场偏饱和，客单价 25-35 元，新店半年存活率约 55%"
@tool
def estimate_startup_cost(city: str, category: str, area: int) -> str:
"""按城市、品类、面积测算开店启动成本。"""
return f"{city} {area} 平米{category}：启动成本约 38 万（转让费 8 万 + 装修 15 万 + 设备 10 万 + 首批物料 5 万）"
@tool
def get_rent(district: str) -> str:
"""查询指定商圈的铺面租金水平。"""
return f"{district}：临街铺面月租约 300 元/平米，100 平米门店月租约 3 万"
@tool
def check_competition(district: str, category: str) -> str:
"""查询指定商圈某品类的竞争情况。"""
return f"{district}：500 米内已有 4 家同类门店，其中 2 家是连锁品牌，竞争激烈"
tools = [search_market, estimate_startup_cost, get_rent, check_competition]
tool_map = {
t.name: t for t in tools}
# Executor 用带工具的模型；Planner 不带工具，纯思考拆任务，这就是两个角色的分工
llm_with_tools = llm.bind_tools(tools)
def make_plan(goal: str) -> list[str]:
"""调用 Planner 把总目标拆成有序步骤列表。
Args:
goal: 用户的总目标描述
Returns:
按顺序执行的步骤列表
"""
planner_prompt = f"""你是餐饮调研团队的规划员（Planner），负责把老板的目标拆成可执行的调研步骤。
要求：步骤不超过 5 步，每步一句话，按逻辑排序，最后一步必须是"汇总以上信息输出简报"。
老板的目标：{goal}"""
# with_structured_output 强制模型按 Plan 结构输出，返回值直接是 Plan 对象
plan = llm.with_structured_output(Plan).invoke(planner_prompt)
return plan.steps
def execute_step(goal: str, steps: list[str], step_index: int, done_results: list[str], max_iter: int = 4) -> str:
"""执行计划中的单个步骤，步骤内允许调用工具，返回该步骤的执行结果。
Args:
goal: 总目标，让 Executor 知道自己在为哪个目标服务
steps: 完整计划步骤列表
step_index: 当前要执行的步骤下标
done_results: 已完成步骤的结果列表
max_iter: 单步内最大工具调用轮数，防跑偏
Returns:
当前步骤的执行结果文本
"""
# 把总目标、完整计划、已完成结果、当前步骤一起塞给 Executor
# 这是 Plan-and-Execute 的关键：执行时始终看得见全局计划，不容易走散
done_summary = "\n".join(f"步骤{i+1}结果：{r}" for i, r in enumerate(done_results)) or "（还没执行任何步骤）"
system_prompt = f"""你是餐饮调研团队的执行员（Executor），只负责执行指定步骤，不要做整份简报。
可以调用工具获取数据，拿到数据后用两三句话总结本步骤结论。
总目标：{goal}
完整计划：
{chr(10).join(f"{i+1}. {s}" for i, s in enumerate(steps))}
已完成步骤结果：
{done_summary}
当前要执行的步骤：{steps[step_index]}"""
messages = [SystemMessage(content=system_prompt)]
for _ in range(max_iter):
response = llm_with_tools.invoke(messages)
messages.append(response)
# 不再调工具，说明本步骤执行完毕
if not response.tool_calls:
return response.content
# 执行工具并把结果喂回去
for call in response.tool_calls:
result = tool_map[call["name"]].invoke(call["args"])
messages.append(ToolMessage(content=str(result), tool_call_id=call["id"]))
return "本步骤执行超时，跳过。"
def run_plan_and_execute(goal: str) -> str:
"""跑完整 Plan-and-Execute 流程：规划 -> 逐步执行 -> 汇总简报。
Args:
goal: 用户的总目标描述
Returns:
最终生成的调研简报文本
"""
# 第一步：Planner 拆任务
steps = make_plan(goal)
print("=== Planner 生成的计划 ===")
for i, s in enumerate(steps):
print(f"{i + 1}. {s}")
# 第二步：Executor 逐步执行，记录每步结果
done_results: list[str] = []
print("\n=== Executor 开始逐步执行 ===")
for i in range(len(steps) - 1):
print(f"\n--- 执行步骤 {i + 1}/{len(steps)}：{steps[i]} ---")
result = execute_step(goal, steps, i, done_results)
done_results.append(result)
print(f"步骤结果：{result}")
# 第三步：带着全部步骤结果生成最终简报（最后一步是汇总）
print(f"\n--- 执行步骤 {len(steps)}/{len(steps)}：{steps[-1]} ---")
evidence = "\n".join(f"步骤{i+1}结果：{r}" for i, r in enumerate(done_results))
final_prompt = f"""你是餐饮老板勇哥的调研助手，基于以下调研结果写一份开店简报。
要求：结论在前，数据在后，最后给出明确的"开/不开"建议和理由。语气直接，不绕弯子。
总目标：{goal}
调研结果：
{evidence}"""
report = llm.invoke(final_prompt).content
print(f"\n=== 最终简报 ===\n{report}")
return report
# 3. 跑一个复杂多步任务：如果用 ReAct 边想边做，这种任务容易走散，先规划就稳很多
if __name__ == "__main__":
run_plan_and_execute("勇哥计划在深圳开一家 100 平米的新面馆，帮我做开店前调研并输出简报。")
运行后先看到 Planner 拆出的计划，再逐步执行：
=== Planner 生成的计划 ===
1. 调研深圳面馆市场的整体行情和存活率
2. 测算 100 平米面馆的启动成本
3. 调研目标商圈的租金水平和竞争情况
4. 汇总以上信息输出简报
=== Executor 开始逐步执行 ===
--- 执行步骤 1/4：调研深圳面馆市场的整体行情和存活率 ---
[调用 search_market(city='深圳', category='面馆')]
步骤结果：深圳面馆市场偏饱和，客单价 25-35 元，新店半年存活率约 55%，进入门槛不低。
--- 执行步骤 2/4：测算 100 平米面馆的启动成本 ---
[调用 estimate_startup_cost(city='深圳', category='面馆', area=100)]
步骤结果：启动成本约 38 万，转让费是大头之一，回本压力不小。
=== 最终简报 ===
结论：不建议现在贸然入场。市场偏饱和、半年存活率仅 55%；启动成本 38 万；目标商圈已有 4 家同类门店……
Plan-and-Execute 模式的一些注意点：
1、计划用结构化输出。Plan 用 Pydantic 约束，返回值直接是步骤列表，程序才能 for 循环逐步推进
2、执行每一步时，把完整计划和已完成结果都放进上下文。Executor 始终知道当前步骤和总目标的关系，这是它比 ReAct 抗迷失的原因。
3、计划应该可调整。这个例子里计划是固定的，真实任务里经常出现查完发现假设不成立的情况。成熟的做法是把计划当成可更新的任务清单：每完成一步评估一次，假设不成立就重新规划后续步骤。
三、Reflection 模式：交付前按标准检查
Reflection 的做法是增加一个明确的检查环节：先产出结果，再对照验收标准检查，发现问题就修正，校验通过才交付。核心在验收标准上——有明确的标准，检查才有依据。过程如下：
Generator（生成器）：产出初稿
↓
Reviewer（审查器）：对照验收标准逐条检查，输出结构化的问题清单
↓
Revisor（修正器）：按问题清单修正
↓
未通过就再来一轮，通过才交付
Reflection 模式适合有明确验收标准的任务：代码要能通过单测，JSON 要符合 Schema，文案要满足固定要求。
举一个文档生成的例子：让 Agent 给餐饮新人写一份开店前必读提醒。初稿生成时只给了笼统要求（语气亲切、不用列数字），审查环节对照勇哥的四条硬标准检查。这模拟的是实际开发里常见的问题：生成时的要求和验收时的标准不一致。
# Reflection 模式：生成 -> 按明确标准审查 -> 修正，演示交付前的检查环节
# 主题：勇哥说餐饮——给粉丝写"开店前必读提醒"，初稿没对齐勇哥的标准，被审查环节拦下来修正
from dotenv import load_dotenv
import os
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
load_dotenv()
llm = ChatOpenAI(
model="qwen3.7-max-2026-06-08",
api_key=os.getenv("API_KEY"),
base_url=os.getenv("BASE_URL")
)
# 1. 审查结果的输出结构：passed 决定是否放行，issues 是具体问题清单
# Reflection 的核心不是让模型说"我再检查一下"，而是把检查结果变成程序可判断的结构
class ReviewResult(BaseModel):
passed: bool = Field(description="初稿是否全部满足审查标准")
issues: list[str] = Field(description="不满足标准的具体问题列表，全部满足时为空列表")
# 2. 勇哥的验收标准：检查必须有明确抓手，没有标准的反思就是自我安慰
REVIEW_STANDARDS = """1. 必须包含三条成本红线：房租不超营业额 15%、人工不超 20%、食材不超 30%
2. 必须包含一条明确的止损建议（连续亏损多久该怎么办）
3. 语气必须直接、不留情面，符合勇哥"刀子嘴"人设，不能出现"建议您酌情考虑"这类客服腔
4. 不得出现任何具体品牌名，避免拉踩惹麻烦"""
def generate_draft(task: str) -> str:
"""生成初稿。故意不对齐验收标准，模拟真实开发里"生成和验收两张皮"的场景。
Args:
task: 内容生成任务描述
Returns:
模型生成的初稿文本
"""
# 注意：这里刻意只给了笼统要求（语气亲切、不用列数字）
# 初稿大概率过不了审查，正好演示 Reflection 怎么把问题拦下来
prompt = f"""请完成以下写作任务，300 字左右，语气亲切温和，读起来舒服即可，不用列具体数字。
任务：{task}"""
return llm.invoke(prompt).content
def review_draft(task: str, draft: str) -> ReviewResult:
"""按验收标准审查初稿，返回结构化审查结果。
Args:
task: 原始写作任务
draft: 待审查的初稿文本
Returns:
ReviewResult 对象：passed 是否通过，issues 问题列表
"""
reviewer_prompt = f"""你是严格的审稿人，对照下面的验收标准逐条审查初稿，一条都不能放过。
写作任务：{task}
验收标准：
{REVIEW_STANDARDS}
待审查初稿：
{draft}"""
# with_structured_output 保证审查结果可被程序判断，而不是一段没法用的自由文本
return llm.with_structured_output(ReviewResult).invoke(reviewer_prompt)
def revise_draft(task: str, draft: str, issues: list[str]) -> str:
"""按审查发现的问题修正初稿。
Args:
task: 原始写作任务
draft: 未通过审查的初稿文本
issues: 审查环节发现的问题列表
Returns:
修正后的终稿文本
"""
issue_list = "\n".join(f"- {issue}" for issue in issues)
prompt = f"""你是文章修改助手，按照审查问题逐条修正下面的初稿，其他内容保持不变。
写作任务：{task}
验收标准：
{REVIEW_STANDARDS}
审查发现的问题：
{issue_list}
待修正初稿：
{draft}"""
return llm.invoke(prompt).content
def run_reflection(task: str, max_rounds: int = 3) -> str:
"""跑完整 Reflection 流程：生成 -> 审查 -> 修正，未通过可多轮修正。
Args:
task: 内容生成任务描述
max_rounds: 最大审查修正轮数，防止无限循环
Returns:
通过审查的终稿文本
"""
# 第一步：生成初稿
draft = generate_draft(task)
print("=== 初稿（未对齐标准随手生成）===")
print(draft)
# 第二步：按标准审查，发现问题就修正，直到通过或达到轮数上限
for round_no in range(1, max_rounds + 1):
review = review_draft(task, draft)
print(f"\n=== 第 {round_no} 轮审查 ===")
print(f"是否通过：{'通过' if review.passed else '不通过'}")
for issue in review.issues:
print(f"发现问题：{issue}")
# 全部标准满足，放行交付
if review.passed:
print("\n审查通过，交付终稿。")
return draft
# 第三步：按问题清单修正，进入下一轮审查
draft = revise_draft(task, draft, review.issues)
print(f"\n=== 修正后稿件（第 {round_no} 轮）===")
print(draft)
print("\n达到最大审查轮数，按当前版本交付。")
return draft
# 3. 跑一次完整流程：初稿是"温和客服腔"，审查按勇哥的四条硬标准把它拦下来
if __name__ == "__main__":
run_reflection("给第一次开餐饮店的新人写一份《开店前必读提醒》")
运行效果：
=== 初稿（未对齐标准随手生成）===
亲爱的朋友，开店是一件令人期待的事情呢～建议您提前做好市场调研，
合理规划预算，保持良好的心态，相信用心经营一定会收获回报的哦……
=== 第 1 轮审查 ===
是否通过：不通过
发现问题：未包含三条成本红线数字（房租 15%、人工 20%、食材 30%）
发现问题：缺少明确的止损建议
发现问题：语气亲切温和，出现"建议您""呢～"等客服腔，不符合勇哥直接不留情面的人设
=== 修正后稿件（第 1 轮）===
开店前先记住三条红线：房租别超营业额 15%，人工别超 20%，食材别超 30%，
超了就是在给房东和员工打工。连续三个月亏损看不到拐点，趁早关店止损，
别死在"再坚持一下"……
=== 第 2 轮审查 ===
是否通过：通过
审查通过，交付终稿。
初稿是温和的客服腔，四条标准都不满足，被审查环节拦下，修正一轮后通过。如果没有校验环节，初稿就直接交付出去了。
Reflection 模式的一些注意点：
1、审查结果用结构化输出。ReviewResult 把是否通过变成程序能判断的布尔值，问题清单直接传给修正环节。审查器输出自由文本的话，循环逻辑写不起来。
2、修正时带上问题清单和原标准。
3、审查标准来自业务。这四条标准是根据餐饮行业规则定的，模型只负责执行检查，不能指望模型自己生成验收标准。
三种模式在实际项目里经常组合：Planner 拆任务保证不漏项，执行时用 ReAct 根据结果决定下一步，交付前用 Reflection 检查。
四、总结
- ReAct：Thought → Action → Observation 循环，边想边做，适合路径不确定的排查类任务，必须配步数限制。
- Plan-and-Execute：Planner 拆任务，Executor 逐步执行，适合复杂多步任务，计划要可调整。
- Reflection：生成、审查、修正三步，交付前按标准检查，验收标准要明确，审查结果要结构化。
三种模式都包含以下几个步骤： 工具调用 、结构化输出、循环控制。区别只是花心思的环节不同：ReAct 在每一步的决策上，Plan-and-Execute 在开工前的拆解上，Reflection 在交付前的检查上。
下一篇讲 Agent 记忆 机制。任务一长、对话一多，Agent 就开始丢失信息，所以它需要一套自己的记忆系统。
五、LuckReport 项目推荐
导航：LuckReport专栏
1、项目简介
Luck-Report 是一款基于开源项目 UReport2 重构的 Java 高性能报表引擎，通过迭代单元格可以实现任意复杂的中国式报表。相较于 UReport2，在技术架构上进行了全新升级，后端基于 SpringBoot 框架开发、前端采用 Vue 框架构建，技术选型贴合当下主流项目开发标准，可精准适配各类实际开发需求。
Luck-Report 提供了全新的基于网页的报表设计器，可以在 Chrome、Firefox、Edge 等各种主流浏览器运行（IE 浏览器除外）。使用 Luck-Report，打开浏览器即可完成各种复杂报表的设计制作。
Luck-Report 基于 Apache-2.0 开源协议 开源
