AgentEra
[Showcase]5 common patterns of Agent / LLM Agentic flow #263
Maplemx
started this conversation in
Show and tell
Replies: 4 comments
Todo List & Concurrency Executor"""
To-Do List + TriggerFlow 并发规划
展示模型拆解任务 → 依赖分析 → 并发执行 → 结果收集的完整链路
"""
import os
import asyncio
import time
from dotenv import load_dotenv, find_dotenv
from agently import Agently, TriggerFlow, TriggerFlowRuntimeData
load_dotenv(find_dotenv())
Agently.set_settings("OpenAICompatible", {
"base_url": os.environ.get("DEEPSEEK_BASE_URL", "https://api.deepseek.com"),
"model": os.environ.get("DEEPSEEK_DEFAULT_MODEL", "deepseek-v4-flash"),
"auth": os.environ.get("DEEPSEEK_API_KEY", ""),
})
# ── 模拟任务执行耗时 ────────────────────────────────────────────
# 不同任务类型有不同的模拟耗时（秒），让并发效果可见
TASK_DURATIONS = {
"t1": 1.2, "t2": 0.8, "t3": 1.0, "t4": 0.6, "t5": 0.9,
"t6": 1.1, "t7": 0.7, "t8": 0.5,
"default": 0.8,
}
# ── Phase 1: 模型生成 To-Do List ────────────────────────────────
def generate_todo_list(task: str) -> dict:
response = (
Agently.create_agent()
.input(task)
.instruct([
"把任务拆解成可独立执行的子任务",
"每个子任务要具体、可验证",
"在 depends_on 里标注依赖关系（被依赖的任务必须先完成）",
])
.output({
"tasks": [{
"id": ("str", "唯一标识，如 't1'"),
"title": ("str", "子任务标题"),
"description": ("str", "具体要做什么"),
"depends_on": (["str"], "依赖的子任务 id，无依赖则为空列表"),
}],
"summary": ("str", "整体拆解思路"),
})
.get_response()
)
return response.get_data(ensure_keys=["tasks", "summary"])
def analyze_dependencies(tasks: list) -> tuple[list, list]:
"""
分析依赖关系，将任务分为两组：
- independent: 无依赖，可并发执行
- dependent: 有依赖，必须等前置任务完成后顺序执行
"""
independent = [t for t in tasks if not t.get("depends_on")]
dependent = [t for t in tasks if t.get("depends_on")]
return independent, dependent
# ── Phase 2 & 3: TriggerFlow 并发执行 ────────────────────────────
def build_concurrent_executor():
"""构建一个支持 for_each 并发执行的 TriggerFlow"""
flow = TriggerFlow(name="todo-concurrent-executor")
async def execute_one_task(data: TriggerFlowRuntimeData):
task = data.input
task_id = task.get("id", "unknown")
duration = TASK_DURATIONS.get(task_id, TASK_DURATIONS["default"])
print(f" [{task_id}] 开始: {task.get('title', '')} (模拟耗时 {duration}s)")
await asyncio.sleep(duration)
print(f" [{task_id}] 完成")
return {
"id": task_id,
"title": task.get("title", ""),
"result": f"{task.get('title', '')} — 已完成",
"duration": duration,
}
async def collect_results(data: TriggerFlowRuntimeData):
results = data.input # for_each 自动收集为 list
await data.async_set_state("execution_results", results)
flow.for_each(concurrency=3).to(execute_one_task).end_for_each().to(collect_results)
return flow
# ── 顺序执行对照 ────────────────────────────────────────────────
async def execute_sequential(tasks: list) -> tuple[float, list]:
"""顺序执行所有任务，用于和并发执行对比耗时"""
results = []
started = time.perf_counter()
for task in tasks:
task_id = task.get("id", "unknown")
duration = TASK_DURATIONS.get(task_id, TASK_DURATIONS["default"])
print(f" [{task_id}] 顺序执行: {task.get('title', '')} (耗时 {duration}s)")
await asyncio.sleep(duration)
results.append({
"id": task_id,
"title": task.get("title", ""),
"result": f"{task.get('title', '')} — 已完成",
})
elapsed = time.perf_counter() - started
return elapsed, results
# ── Main ────────────────────────────────────────────────────────
async def main():
# 1. 模型生成 To-Do List
print("=" * 50)
print("Phase 1: 模型拆解任务")
print("=" * 50)
result = generate_todo_list(
"搭建一个多语言客服机器人：支持中文和英文，能回答常见问题、查订单状态、转人工"
)
print(f"拆解思路：{result['summary']}\n")
for t in result["tasks"]:
deps = f"（依赖：{', '.join(t['depends_on'])}）" if t.get("depends_on") else ""
print(f" [{t['id']}] {t['title']} {deps}")
print(f" {t['description']}")
# 2. 依赖分析
print(f"\n{'=' * 50}")
print("Phase 2: 依赖分析")
print("=" * 50)
independent, dependent = analyze_dependencies(result["tasks"])
print(f"可并发执行（无依赖）：{len(independent)} 个 — {[t['id'] for t in independent]}")
print(f"需排队执行（有依赖）：{len(dependent)} 个 — {[t['id'] for t in dependent]}")
# 3. 顺序执行 vs 并发执行对比
if len(independent) >= 2:
print(f"\n{'=' * 50}")
print("Phase 3a: 顺序执行（对照）")
print("=" * 50)
seq_elapsed, seq_results = await execute_sequential(independent)
seq_total = sum(
TASK_DURATIONS.get(r["id"], TASK_DURATIONS["default"])
for r in seq_results
)
print(f"\n顺序执行 {len(independent)} 个任务：总耗时 {seq_elapsed:.2f}s"
f"（各任务耗时之和 ≈ {seq_total:.1f}s）")
print(f"\n{'=' * 50}")
print("Phase 3b: TriggerFlow for_each 并发执行")
print("=" * 50)
flow = build_concurrent_executor()
started = time.perf_counter()
state = flow.start(independent)
conc_elapsed = time.perf_counter() - started
print(f"\n并发执行 {len(independent)} 个任务：总耗时 {conc_elapsed:.2f}s"
f"（concurrency=3，耗时 ≈ 最长单个任务 + 调度开销）")
# 对比
speedup = seq_elapsed / conc_elapsed if conc_elapsed > 0 else 0
print(f"\n加速比：{speedup:.1f}x"
f"（顺序 {seq_elapsed:.2f}s → 并发 {conc_elapsed:.2f}s）")
print(f"\n执行结果：")
for r in state.get("execution_results", []):
print(f" [{r['id']}] {r['result']}")
# 4. 处理有依赖的任务（顺序执行）
if dependent:
print(f"\n{'=' * 50}")
print("Phase 4: 有依赖的任务顺序执行")
print("=" * 50)
for t in dependent:
print(f" [{t['id']}] {t['title']} ← 等待前置: {t.get('depends_on')}")
dep_elapsed, dep_results = await execute_sequential(dependent)
print(f"有依赖任务执行完成，耗时 {dep_elapsed:.2f}s")
if __name__ == "__main__":
asyncio.run(main()) |
0 replies
Plan Mode"""
Plan 模式 —— 多轮澄清，只产出计划
Plan 模式不是「确认后执行」。
它的边界是：先询问、补齐约束，再给出行动计划；到计划为止，不进入执行阶段。
"""
import os
from dotenv import load_dotenv, find_dotenv
from agently import Agently, TriggerFlow, TriggerFlowRuntimeData
load_dotenv(find_dotenv())
Agently.set_settings("OpenAICompatible", {
"base_url": os.environ.get("DEEPSEEK_BASE_URL", "https://api.deepseek.com"),
"model": os.environ.get("DEEPSEEK_DEFAULT_MODEL", "deepseek-v4-flash"),
"auth": os.environ.get("DEEPSEEK_API_KEY", ""),
})
# ── Phase 1: 判断是否能进入计划阶段 ─────────────────────────────
def assess_plan_readiness(task: str, clarifications: list) -> dict:
response = (
Agently.create_agent()
.input({
"原始任务": task,
"已经补充的信息": clarifications,
})
.instruct([
"你正在模拟 Plan 模式，不允许执行任务",
"判断当前信息是否足以产出可执行计划",
"如果关键信息缺失，提出 1-3 个最重要的澄清问题",
"不要一次性追问细枝末节；能合理假设的内容留到计划的 assumptions",
"只有当目标、范围、约束、验收标准基本清楚时，ready 才能为 true",
])
.output({
"ready": ("bool", "当前信息是否足以生成行动计划"),
"reason": ("str", "为什么可以进入计划，或为什么还需要澄清"),
"questions": [{
"question": ("str", "需要用户回答的澄清问题"),
"why_needed": ("str", "这个问题会影响计划中的什么决策"),
}],
})
.get_response()
)
return response.get_data(ensure_keys=["ready", "reason", "questions"])
# ── Phase 2: 只生成计划，不执行 ─────────────────────────────────
def generate_action_plan(task: str, clarifications: list) -> dict:
response = (
Agently.create_agent()
.input({
"原始任务": task,
"澄清后的约束": clarifications,
})
.instruct([
"你正在模拟 Plan 模式，只输出行动计划，不执行任何操作",
"计划要能让用户判断是否授权后续执行",
"明确列出每一步的目的、具体动作、产出物和验收标准",
"标注风险、依赖、假设；不要声称已经完成任何步骤",
])
.output({
"goal": ("str", "最终目标"),
"assumptions": (["str"], "计划中采用的假设"),
"steps": [{
"step": ("int", "步骤编号"),
"objective": ("str", "这一步的目的"),
"action": ("str", "后续如果执行，需要做的具体动作"),
"deliverable": ("str", "这一步完成后应产出的结果"),
"check": ("str", "如何验证这一步完成"),
}],
"risks": [{
"risk": ("str", "潜在风险"),
"mitigation": ("str", "计划中的规避或缓解方式"),
}],
"done_criteria": (["str"], "整体完成标准"),
})
.get_response()
)
return response.get_data(ensure_keys=["goal", "steps", "done_criteria"])
def print_plan(plan: dict):
print("\n最终行动计划（只计划，不执行）")
print("=" * 50)
print(f"目标：{plan.get('goal', '')}")
assumptions = plan.get("assumptions") or []
if assumptions:
print("\n假设：")
for item in assumptions:
print(f" - {item}")
print("\n步骤：")
for s in plan.get("steps", []):
print(f" {s['step']}. {s['objective']}")
print(f" 动作：{s['action']}")
print(f" 产出：{s['deliverable']}")
print(f" 验收：{s['check']}")
risks = plan.get("risks") or []
if risks:
print("\n风险与缓解：")
for item in risks:
print(f" - 风险：{item['risk']}")
print(f" 缓解：{item['mitigation']}")
print("\n完成标准：")
for item in plan.get("done_criteria", []):
print(f" - {item}")
print("\n[Plan 模式停止点] 已产出计划，未执行任何文件、数据库或外部系统操作。")
# ── Phase 3: TriggerFlow 澄清循环 ───────────────────────────────
def build_plan_mode_flow(max_rounds: int = 3):
"""
PLAN 模式流程：
Assess → Clarify → Assess → ... → Plan
- Assess: 判断信息是否足够
- Clarify: 信息不足时继续向用户提问
- Plan: 信息足够后只输出行动计划，不进入执行
"""
flow = TriggerFlow(name="plan-mode-clarify-only")
async def assess(data: TriggerFlowRuntimeData):
state = data.input
task = state["task"]
clarifications = state.get("clarifications", [])
round_num = state.get("round", 0)
decision = assess_plan_readiness(task, clarifications)
print(f"\n[第 {round_num + 1} 轮判断]")
print(f"判断：{'信息足够' if decision['ready'] else '需要继续澄清'}")
print(f"原因：{decision.get('reason', '')}")
next_state = {**state, "decision": decision}
if decision["ready"] or round_num >= max_rounds:
if not decision["ready"]:
print(f"\n达到最大澄清轮次 {max_rounds}，基于现有信息生成计划。")
data.emit_nowait("MakePlan", next_state)
else:
data.emit_nowait("Clarify", next_state)
async def clarify(data: TriggerFlowRuntimeData):
state = data.input
questions = state.get("decision", {}).get("questions", [])
clarifications = list(state.get("clarifications", []))
round_num = state.get("round", 0) + 1
print(f"\n[第 {round_num} 轮澄清]")
for index, q in enumerate(questions, start=1):
print(f"\n问题 {index}：{q['question']}")
print(f"为什么问：{q.get('why_needed', '')}")
answer = input("你的回答：").strip()
clarifications.append({
"question": q["question"],
"answer": answer,
})
data.emit_nowait("Assess", {
**state,
"clarifications": clarifications,
"round": round_num,
})
async def make_plan(data: TriggerFlowRuntimeData):
state = data.input
plan = generate_action_plan(state["task"], state.get("clarifications", []))
print_plan(plan)
await data.async_set_state("task", state["task"], emit=False)
await data.async_set_state("clarifications", state.get("clarifications", []), emit=False)
await data.async_set_state("clarification_rounds", state.get("round", 0), emit=False)
await data.async_set_state("plan", plan, emit=False)
await data.async_set_state("result", "已生成行动计划，未执行", emit=False)
flow.to(assess)
flow.when("Assess").to(assess)
flow.when("Clarify").to(clarify)
flow.when("MakePlan").to(make_plan)
return flow
if __name__ == "__main__":
flow = build_plan_mode_flow(max_rounds=3)
task = input("请输入一个需要先规划的任务：").strip()
if not task:
task = "帮我把课程项目上线到生产环境"
print(f"\n任务：{task}")
state = flow.start({
"task": task,
"clarifications": [],
"round": 0,
})
print(f"\n结果：{state.get('result', '')}") |
0 replies
Reflection"""
Reflection 模式 —— TriggerFlow signal-driven 迭代改进
"""
import os
from dotenv import load_dotenv, find_dotenv
from agently import Agently, TriggerFlow, TriggerFlowRuntimeData
load_dotenv(find_dotenv())
Agently.set_settings("OpenAICompatible", {
"base_url": os.environ.get("DEEPSEEK_BASE_URL", "https://api.deepseek.com"),
"model": os.environ.get("DEEPSEEK_DEFAULT_MODEL", "deepseek-v4-flash"),
"auth": os.environ.get("DEEPSEEK_API_KEY", ""),
})
def build_reflection_flow(max_rounds: int = 3):
"""
signal-driven reflection loop：
GENERATE → EVALUATE → (不通过) → REVISE → EVALUATE → ... → 通过/达到上限
和第06课 ReAct Loop 结构同构，但循环的是「内容改进」而非「工具调用」
"""
flow = TriggerFlow(name="reflection-loop")
async def generate_draft(data: TriggerFlowRuntimeData):
"""生成初稿，然后进入评估循环"""
state = data.input
task = state.get("task", "")
print("[初稿生成]")
draft = (
Agently.create_agent()
.input(task)
.output({"result": ("str", "根据任务要求生成的内容")})
.get_response()
.get_data(ensure_keys=["result"])
)["result"]
print(f" 初稿（{len(draft)} 字符）")
new_state = {**state, "draft": draft, "round": 0}
data.emit_nowait("Evaluate", new_state)
async def evaluate(data: TriggerFlowRuntimeData):
state = data.input
round_num = state.get("round", 0) + 1
draft = state.get("draft", "")
task = state.get("task", "")
criteria = state.get("criteria", [])
if round_num > max_rounds:
print(f"\n[第 {round_num} 轮] 达到最大轮次 {max_rounds}，强制终止")
await data.async_set_state("final_result", draft, emit=False)
await data.async_set_state("total_rounds", max_rounds, emit=False)
return
print(f"\n[第 {round_num} 轮评估]")
critique = (
Agently.create_agent()
.input({"任务要求": task, "当前版本": draft})
.info({"评估标准": criteria})
.instruct(["对照评估标准，检查当前版本是否满足所有要求"])
.output({
"passed": ("bool", "是否满足所有标准"),
"score": ("float", "0-1 综合质量分"),
"issues": (["str"], "不满足的点，通过则为空列表"),
"suggestions": (["str"], "具体改进建议，通过则为空列表"),
})
.get_response()
.get_data(ensure_keys=["passed", "score"])
)
print(f" 分数：{critique['score']:.2f} 通过：{critique['passed']}")
if critique.get("issues"):
for issue in critique["issues"]:
print(f" 问题：{issue}")
new_state = {
**state,
"round": round_num,
"last_critique": critique,
}
if critique["passed"]:
print(f" 通过！最终分数 {critique['score']:.2f}")
await data.async_set_state("final_result", draft, emit=False)
await data.async_set_state("total_rounds", round_num, emit=False)
return
# 不通过 → 发信号进入改进步骤
data.emit_nowait("Revise", new_state)
async def revise(data: TriggerFlowRuntimeData):
state = data.input
task = state.get("task", "")
draft = state.get("draft", "")
suggestions = state.get("last_critique", {}).get("suggestions", [])
round_num = state.get("round", 1)
print(f"\n[第 {round_num} 轮改进] 根据 {len(suggestions)} 条建议修改...")
improved = (
Agently.create_agent()
.input({"原始任务": task, "当前版本": draft})
.info({"需要改进的点": suggestions})
.instruct([
"根据改进建议修改当前版本",
"保留已满足要求的部分，只修改不满足的",
])
.output({"result": ("str", "改进后的内容")})
.get_response()
.get_data(ensure_keys=["result"])
)
new_draft = improved["result"]
print(f" 改进完成（{len(draft)} → {len(new_draft)} 字符）")
# 更新 draft，发信号回到评估步骤
data.emit_nowait("Evaluate", {**state, "draft": new_draft})
flow.to(generate_draft)
flow.when("Evaluate").to(evaluate)
flow.when("Revise").to(revise)
return flow
if __name__ == "__main__":
flow = build_reflection_flow(max_rounds=3)
state = flow.start({
"task": "为一个 AI 开发课程写一段招生介绍，不超过 80 字，面向有编程基础的工程师",
"criteria": [
"字数不超过 80 字",
"明确说明课程的核心学习内容",
"语气简洁，适合工程师读者，不夸张",
"包含一个具体的学员收益",
],
})
print(f"\n{'=' * 50}")
print(f"总轮次：{state.get('total_rounds', '?')}")
print(f"最终版本：")
print(state.get("final_result", state.get("draft", "")))
# ─────────────────────────────────────────────────────────────
# 示例输出（实际运行结果，模型非确定性，每次会有差异）
# ─────────────────────────────────────────────────────────────
# [初稿生成]
# 初稿（56 字符）
#
# [第 1 轮评估]
# 分数：0.60 通过：False
# 问题：字数超过80字（当前约90字）
# 问题：未明确说明课程核心学习内容，仅笼统描述为'AI开发核心技能'
# 问题：学员收益不具体，'升职加薪必备'过于夸张且空泛
#
# [第 1 轮改进] 根据 3 条建议修改...
# 改进完成（56 → 52 字符）
#
# [第 2 轮评估]
# 分数：0.95 通过：True
# 通过！最终分数 0.95
#
# ==================================================
# 总轮次：2
# 最终版本：
# 专为有编程基础的工程师设计，系统学习大模型微调、深度学习模型部署，
# 实战驱动，助你独立完成企业级AI项目。
#
# 观察要点：
# - 通常 2-3 轮内收敛；若评分卡在 ≈0.85 反复改不动，多半是 criteria 之间互相冲突
# - max_rounds 是"安全阀"，达到上限即使没通过也强制退出，避免死循环
# - 信号驱动循环：evaluate 内部 emit "Revise"，revise 内部 emit "Evaluate"，
# 和第06课 ReAct Loop 的 emit("ToolUse")/emit("Reply") 同构 |
0 replies
Wait for Confirm"""
Confirm 模式 —— 计划后的审批与放行
Confirm 模式不是生成计划本身，而是在已有计划或拟执行操作进入执行前，
根据风险等级、可回滚性和人工确认结果决定是否放行。
"""
import os
from dotenv import load_dotenv, find_dotenv
from agently import Agently, TriggerFlow, TriggerFlowRuntimeData
load_dotenv(find_dotenv())
Agently.set_settings("OpenAICompatible", {
"base_url": os.environ.get("DEEPSEEK_BASE_URL", "https://api.deepseek.com"),
"model": os.environ.get("DEEPSEEK_DEFAULT_MODEL", "deepseek-v4-flash"),
"auth": os.environ.get("DEEPSEEK_API_KEY", ""),
})
# ── Phase 1: 生成待确认的执行方案 ───────────────────────────────
def generate_execution_proposal(task: str) -> dict:
response = (
Agently.create_agent()
.input(task)
.instruct([
"生成一个待确认的执行方案，但不要真的执行",
"明确列出每一步会产生什么影响",
"评估整体风险等级和可回滚性",
"高风险操作要说明为什么需要人工确认",
])
.output({
"steps": [{
"step": ("int", "步骤编号"),
"action": ("str", "如果获批，后续会执行的具体操作"),
"impact": ("str", "这一步会改变什么"),
}],
"risk_level": ("'low' | 'medium' | 'high'", "整体风险等级"),
"risk_reason": ("str", "风险评估依据"),
"reversible": ("bool", "操作是否可回滚"),
})
.get_response()
)
return response.get_data(ensure_keys=["steps", "risk_level"])
# ── Phase 2: TriggerFlow 审批策略路由 ───────────────────────────
def build_confirm_flow():
"""
根据 risk_level + reversible 自动决定确认策略：
- low: 自动放行
- medium + reversible: 自动放行但打印警告
- medium + irreversible 或 high: 等待人工确认
"""
flow = TriggerFlow(name="confirm-with-policy")
async def classify_and_route(data: TriggerFlowRuntimeData):
task = data.input
proposal = generate_execution_proposal(task)
await data.async_set_state("proposal", proposal, emit=False)
await data.async_set_state("task", task, emit=False)
risk = proposal["risk_level"]
rev = proposal.get("reversible", False)
print(f"\n风险等级：{risk.upper()}")
print(f"风险说明：{proposal.get('risk_reason', '')}")
print(f"可回滚：{'是' if rev else '否'}")
print("\n待确认步骤：")
for s in proposal["steps"]:
print(f" {s['step']}. {s['action']}")
print(f" 影响：{s['impact']}")
if risk == "low":
print("\n[确认策略] 低风险 → 自动放行")
await data.async_set_state("auto_approved", True, emit=False)
elif risk == "medium" and rev:
print("\n[确认策略] 中风险但可回滚 → 自动放行（已记录）")
await data.async_set_state("auto_approved", True, emit=False)
else:
print("\n[确认策略] 高风险或不可回滚 → 等待人工确认")
await data.async_set_state("auto_approved", False, emit=False)
async def auto_approve(data: TriggerFlowRuntimeData):
proposal = data.get_state("proposal", {})
print("\n已自动放行。")
for s in proposal.get("steps", []):
print(f" 已放行步骤 {s['step']}: {s['action']}")
await data.async_set_state(
"result",
f"方案已自动放行，共 {len(proposal.get('steps', []))} 步",
)
async def wait_confirm(data: TriggerFlowRuntimeData):
proposal = data.get_state("proposal", {})
task = data.get_state("task", "")
print(f"\n此操作需要人工确认：{task}")
confirm = input("确认放行后续执行？(y/n) ").strip().lower()
if confirm == "y":
print("已确认放行。")
for s in proposal.get("steps", []):
print(f" 已放行步骤 {s['step']}: {s['action']}")
await data.async_set_state(
"result",
f"方案已人工确认放行，共 {len(proposal.get('steps', []))} 步",
)
else:
print("已拒绝放行。")
await data.async_set_state("result", "用户拒绝放行")
flow.to(classify_and_route).if_condition(
lambda data: data.get_state("auto_approved", False)
).to(auto_approve).else_condition().to(wait_confirm).end_condition()
return flow
if __name__ == "__main__":
flow = build_confirm_flow()
demos = [
("low", "在 README.md 里补充项目的安装说明"),
("high", "将 users 表的 email 字段从 VARCHAR(100) 改为 VARCHAR(255)，并添加唯一索引"),
]
for label, task in demos:
print(f"\n{'=' * 60}")
print(f"任务（期望风险={label}）：{task}")
print(f"{'=' * 60}")
state = flow.start(task)
print(f"\n结果：{state.get('result', '')}") |
0 replies
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Uh oh!
There was an error while loading. Please reload this page.
Routing
All reactions