AI Agent工作流自动化完全教程
本教程系统讲解AI Agent工作流的设计模式、编排框架、企业级架构与实战案例。你将掌握从简单的线性工作流到复杂的多Agent协作系统的完整构建方法。
目录
- Agent工作流设计模式
- LangChain/LangGraph工作流编排
- 定时任务与触发器
- 条件分支与循环
- 错误处理与重试
- 人工审批节点
- 多系统集成
- 监控与日志
- 企业级Agent工作流架构
- 实战案例
1. Agent工作流设计模式
1.1 什么是Agent工作流
Agent工作流是将AI模型的推理能力与确定性的业务逻辑相结合的自动化流程。与单纯的LLM调用不同，Agent工作流具有状态管理、条件分支、错误恢复和多步骤协调的能力。
用户请求
│
▼
┌─────────┐ ┌──────────┐ ┌──────────┐
│ 意图识别 │───▶│ 任务规划 │───▶│ 工具选择 │
└─────────┘ └──────────┘ └──────────┘
│
┌────────────────────────────────┘
▼
┌─────────┐ ┌──────────┐ ┌──────────┐
│ 工具执行 │───▶│ 结果验证 │───▶│ 响应生成 │
└─────────┘ └──────────┘ └──────────┘
│ │
│ ┌────┴────┐
│ │ 不通过 │
│ ▼ │
└──▶ 重新执行 ◀──────┘
1.2 六种核心设计模式
模式一：顺序链（Sequential Chain）
最简单的模式，节点按顺序依次执行，前一个节点的输出作为后一个节点的输入。
from dataclasses import dataclass, field
from typing import Any, Callable
@dataclass
class Step:
"""工作流步骤"""
name: str
handler: Callable[[dict], dict]
timeout: int = 30
@dataclass
class SequentialWorkflow:
"""顺序链工作流"""
name: str
steps: list[Step] = field(default_factory=list)
def add_step(self, name: str, handler: Callable) -> "SequentialWorkflow":
self.steps.append(Step(name=name, handler=handler))
return self
def execute(self, initial_input: dict) -> dict:
context = initial_input.copy()
for step in self.steps:
print(f" ▶ 执行步骤: {step.name}")
try:
result = step.handler(context)
context.update(result)
context["_last_step"] = step.name
except Exception as e:
context["_error"] = str(e)
context["_failed_at"] = step.name
break
return context
# 使用示例：内容生成流水线
workflow = SequentialWorkflow(name="内容生成流水线")
workflow.add_step("需求分析", lambda ctx: {
"topic": f"关于'{ctx['user_input']}'的深度分析",
"style": "专业但易懂"
})
workflow.add_step("大纲生成", lambda ctx: {
"outline": f"1. 引言\n2. 核心概念\n3. 实践方法\n4. 总结"
})
workflow.add_step("内容撰写", lambda ctx: {
"draft": f"根据大纲撰写关于{ctx['topic']}的文章"
})
workflow.add_step("质量检查", lambda ctx: {
"quality_score": 85,
"approved": ctx.get("quality_score", 0) >= 80
})
result = workflow.execute({"user_input": "AI Agent工作流"})
模式二：路由分发（Router Pattern）
根据输入内容将请求路由到不同的处理分支。
class RouterWorkflow:
"""路由分发工作流"""
def __init__(self, name: str):
self.name = name
self.routes: dict[str, Callable] = {}
self.classifier: Callable = None
self.default_route: str = "general"
def set_classifier(self, classifier: Callable):
"""设置意图分类器"""
self.classifier = classifier
def add_route(self, intent: str, handler: Callable):
"""添加路由"""
self.routes[intent] = handler
return self
def execute(self, input_data: dict) -> dict:
# 分类意图
if self.classifier:
intent = self.classifier(input_data)
else:
intent = input_data.get("intent", self.default_route)
# 路由到对应处理器
handler = self.routes.get(intent, self.routes.get(self.default_route))
if not handler:
return {"error": f"未找到路由: {intent}"}
print(f" 📍 路由到: {intent}")
return handler(input_data)
# 示例：客服路由
router = RouterWorkflow("客服路由")
router.set_classifier(lambda ctx: ctx.get("intent", "general"))
router.add_route("complaint", lambda ctx: {"response": "我理解您的不满，让我为您处理..."})
router.add_route("inquiry", lambda ctx: {"response": "让我为您查询相关信息..."})
router.add_route("general", lambda ctx: {"response": "请问有什么可以帮您？"})
模式三：并行扇出/汇聚（Fan-out/Fan-in）
多个节点并行执行，结果汇聚后继续处理。
import concurrent.futures
from typing import Callable
class ParallelWorkflow:
"""并行扇出/汇聚工作流"""
def __init__(self):
self.parallel_tasks: list[tuple[str, Callable]] = []
self.aggregator: Callable = None
def add_parallel_task(self, name: str, handler: Callable):
self.parallel_tasks.append((name, handler))
return self
def set_aggregator(self, handler: Callable):
self.aggregator = handler
return self
def execute(self, input_data: dict) -> dict:
results = {}
# 扇出：并行执行所有任务
with concurrent.futures.ThreadPoolExecutor(max_workers=len(self.parallel_tasks)) as executor:
futures = {}
for name, handler in self.parallel_tasks:
futures[executor.submit(handler, input_data)] = name
for future in concurrent.futures.as_completed(futures):
name = futures[future]
try:
results[name] = future.result()
print(f" ✓ {name} 完成")
except Exception as e:
results[name] = {"error": str(e)}
print(f" ✗ {name} 失败: {e}")
# 汇聚：聚合所有结果
if self.aggregator:
return self.aggregator(results)
return results
模式四：状态机（State Machine）
适用于需要复杂状态转换的场景，如订单处理、审批流程等。
from enum import Enum
from dataclasses import dataclass, field
class OrderState(Enum):
CREATED = "created"
VALIDATING = "validating"
PAYMENT_PENDING = "payment_pending"
PROCESSING = "processing"
SHIPPING = "shipping"
DELIVERED = "delivered"
CANCELLED = "cancelled"
ERROR = "error"
@dataclass
class StateMachine:
"""通用状态机"""
initial_state: str
transitions: dict[tuple[str, str], Callable] = field(default_factory=dict)
current_state: str = ""
def __post_init__(self):
self.current_state = self.initial_state
def add_transition(self, from_state: str, event: str,
to_state: str, handler: Callable = None):
"""添加状态转换规则"""
self.transitions[(from_state, event)] = {
"to": to_state,
"handler": handler
}
return self
def trigger(self, event: str, context: dict) -> dict:
"""触发事件，执行状态转换"""
key = (self.current_state, event)
if key not in self.transitions:
raise ValueError(
f"无效的状态转换: {self.current_state} + {event}"
)
transition = self.transitions[key]
old_state = self.current_state
# 执行转换处理器
if transition["handler"]:
context = transition["handler"](context)
self.current_state = transition["to"]
print(f" 🔄 {old_state} → {self.current_state} (事件: {event})")
context["_state"] = self.current_state
return context
# 订单处理状态机
order_sm = StateMachine(initial_state=OrderState.CREATED.value)
order_sm.add_transition("created", "validate", "validating")
order_sm.add_transition("validating", "valid", "payment_pending")
order_sm.add_transition("validating", "invalid", "cancelled")
order_sm.add_transition("payment_pending", "pay_success", "processing")
order_sm.add_transition("payment_pending", "pay_fail", "error")
order_sm.add_transition("processing", "ship", "shipping")
order_sm.add_transition("shipping", "deliver", "delivered")
模式五：反思循环（Reflection Loop）
Agent生成结果后自我评估，不满足条件则重新生成。
class ReflectionWorkflow:
"""反思循环工作流"""
def __init__(self, max_iterations: int = 3):
self.max_iterations = max_iterations
def execute(self, task: str, generator: Callable,
evaluator: Callable) -> dict:
"""
Args:
task: 任务描述
generator: 生成函数 (task, feedback) -> result
evaluator: 评估函数 (result) -> (score, feedback)
"""
feedback = ""
for i in range(self.max_iterations):
print(f"\n 📝 第{i+1}次生成...")
result = generator(task, feedback)
print(f" 🔍 评估中...")
score, feedback = evaluator(result)
print(f" 📊 评分: {score}/100")
if score >= 80:
return {
"result": result,
"iterations": i + 1,
"final_score": score
}
return {
"result": result,
"iterations": self.max_iterations,
"final_score": score,
"warning": "达到最大迭代次数"
}
模式六：人机协作（Human-in-the-Loop）
关键决策节点引入人工审批。
from enum import Enum
import uuid
import time
class ApprovalStatus(Enum):
PENDING = "pending"
APPROVED = "approved"
REJECTED = "rejected"
TIMEOUT = "timeout"
class HumanApprovalNode:
"""人工审批节点"""
def __init__(self, timeout_seconds: int = 3600):
self.pending_approvals: dict[str, dict] = {}
self.timeout = timeout_seconds
def request_approval(self, context: dict,
question: str,
options: list[str] = None) -> str:
"""发起审批请求，返回审批ID"""
approval_id = str(uuid.uuid4())[:8]
self.pending_approvals[approval_id] = {
"context": context,
"question": question,
"options": options or ["approve", "reject"],
"status": ApprovalStatus.PENDING,
"created_at": time.time()
}
print(f"\n ⏳ 等待人工审批 [{approval_id}]")
print(f" ❓ {question}")
return approval_id
def submit_decision(self, approval_id: str,
decision: str, reason: str = "") -> dict:
"""提交审批决定"""
if approval_id not in self.pending_approvals:
return {"error": "审批ID不存在"}
approval = self.pending_approvals[approval_id]
approval["status"] = ApprovalStatus.APPROVED if decision == "approve" \
else ApprovalStatus.REJECTED
approval["reason"] = reason
approval["decided_at"] = time.time()
print(f" {'✅' if decision == 'approve' else '❌'} "
f"审批结果: {decision}")
return approval
def check_status(self, approval_id: str) -> ApprovalStatus:
"""检查审批状态（支持超时检测）"""
approval = self.pending_approvals.get(approval_id)
if not approval:
return None
if approval["status"] == ApprovalStatus.PENDING:
elapsed = time.time() - approval["created_at"]
if elapsed > self.timeout:
approval["status"] = ApprovalStatus.TIMEOUT
return approval["status"]
1.3 设计模式选择指南
| 模式 | 适用场景 | 复杂度 | 可靠性 |
|---|---|---|---|
| 顺序链 | 线性流水线、数据处理管道 | 低 | 高 |
| 路由分发 | 多意图客服、多类型请求处理 | 中 | 高 |
| 并行扇出/汇聚 | 多源数据聚合、并行分析 | 中 | 中 |
| 状态机 | 订单/审批/工单等有状态流程 | 高 | 高 |
| 反思循环 | 内容生成、代码优化 | 中 | 中 |
| 人机协作 | 关键决策、合规审核 | 高 | 最高 |
2. LangChain/LangGraph工作流编排
2.1 LangChain基础
LangChain是最流行的LLM应用开发框架之一，提供了链（Chain）、代理（Agent）和工具（Tool）的抽象。
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_anthropic import ChatAnthropic
# 基础链
llm = ChatAnthropic(model="claude-sonnet-4-20250514")
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个专业的技术文档翻译专家。"),
("human", "请将以下内容翻译成{language}：\n\n{content}")
])
chain = prompt | llm | StrOutputParser()
result = chain.invoke({
"language": "英文",
"content": "人工智能正在改变我们的工作方式。"
})
print(result)
2.2 LangChain表达式语言（LCEL）
LCEL是LangChain的核心编排语法，支持链式组合、并行执行和条件逻辑。
from langchain_core.runnables import (
RunnablePassthrough, RunnableParallel, RunnableLambda
)
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain_anthropic import ChatAnthropic
from pydantic import BaseModel, Field
llm = ChatAnthropic(model="claude-sonnet-4-20250514")
# 1. 顺序链（LCEL管道）
analysis_prompt = ChatPromptTemplate.from_messages([
("system", "分析用户需求，提取关键信息。"),
("human", "{input}")
])
class AnalysisResult(BaseModel):
topic: str = Field(description="主题")
keywords: list[str] = Field(description="关键词列表")
intent: str = Field(description="用户意图")
analysis_chain = (
analysis_prompt
| llm.with_structured_output(AnalysisResult)
)
# 2. 并行链：同时执行多个分析
parallel_chain = RunnableParallel(
analysis=analysis_chain,
sentiment=ChatPromptTemplate.from_messages([
("system", "判断用户情感，返回：positive/negative/neutral"),
("human", "{input}")
]) | llm | StrOutputParser(),
language=ChatPromptTemplate.from_messages([
("system", "检测用户使用的语言，返回语言名称"),
("human", "{input}")
]) | llm | StrOutputParser()
)
# 3. 完整流水线
def format_output(data: dict) -> str:
return (
f"主题: {data['analysis'].topic}\n"
f"关键词: {', '.join(data['analysis'].keywords)}\n"
f"意图: {data['analysis'].intent}\n"
f"情感: {data['sentiment']}\n"
f"语言: {data['language']}"
)
full_pipeline = parallel_chain | RunnableLambda(format_output)
result = full_pipeline.invoke({
"input": "我最近在学习机器学习，感觉深度学习部分比较难理解"
})
print(result)
2.3 LangGraph状态图工作流
LangGraph是LangChain生态中的图编排框架，适合构建复杂的Agent工作流。
from langgraph.graph import StateGraph, END, START
from langgraph.checkpoint.memory import MemorySaver
from typing import TypedDict, Annotated
import operator
# 定义状态类型
class AgentState(TypedDict):
messages: Annotated[list, operator.add] # 消息历史（自动追加）
current_step: str # 当前步骤
analysis_result: dict # 分析结果
tool_outputs: list[str] # 工具输出
final_answer: str # 最终回答
iteration: int # 迭代次数
# 定义节点函数
def analyze_input(state: AgentState) -> dict:
"""分析用户输入"""
print(" 🔍 分析输入...")
messages = state["messages"]
user_msg = messages[-1]["content"] if messages else ""
# 这里可以调用LLM进行意图识别
return {
"current_step": "analyze",
"analysis_result": {
"intent": "search_and_answer",
"complexity": "medium"
}
}
def select_tools(state: AgentState) -> dict:
"""根据分析结果选择工具"""
print(" 🔧 选择工具...")
intent = state["analysis_result"].get("intent", "")
tools = []
if "search" in intent:
tools.append("web_search")
if "calculate" in intent:
tools.append("calculator")
if not tools:
tools.append("knowledge_base")
return {
"current_step": "select_tools",
"tool_outputs": tools
}
def execute_tools(state: AgentState) -> dict:
"""执行选定的工具"""
print(" ⚡ 执行工具...")
tools = state.get("tool_outputs", [])
results = [f"工具{tool}执行结果" for tool in tools]
return {
"current_step": "execute_tools",
"tool_outputs": results,
"iteration": state.get("iteration", 0) + 1
}
def generate_answer(state: AgentState) -> dict:
"""生成最终回答"""
print(" 💬 生成回答...")
return {
"current_step": "done",
"final_answer": "根据工具执行结果生成的回答"
}
def should_continue(state: AgentState) -> str:
"""条件路由：是否需要继续执行"""
if state.get("iteration", 0) >= 3:
return "generate"
if state.get("current_step") == "execute_tools":
# 检查工具执行结果是否满意
return "generate"
return "continue"
# 构建状态图
graph = StateGraph(AgentState)
# 添加节点
graph.add_node("analyze", analyze_input)
graph.add_node("select_tools", select_tools)
graph.add_node("execute_tools", execute_tools)
graph.add_node("generate", generate_answer)
# 添加边
graph.add_edge(START, "analyze")
graph.add_edge("analyze", "select_tools")
graph.add_edge("select_tools", "execute_tools")
graph.add_conditional_edges(
"execute_tools",
should_continue,
{
"continue": "select_tools", # 循环：继续选择工具
"generate": "generate" # 退出：生成回答
}
)
graph.add_edge("generate", END)
# 编译并运行
app = graph.compile()
result = app.invoke({
"messages": [{"role": "user", "content": "帮我查一下最新的AI趋势"}],
"iteration": 0
})
2.4 带检查点的持久化工作流
from langgraph.checkpoint.sqlite import SqliteSaver
# 使用SQLite持久化检查点（支持断点续执行）
memory = SqliteSaver.from_conn_string(":checkpoints:")
persistent_app = graph.compile(
checkpointer=memory,
interrupt_before=["execute_tools"] # 在工具执行前暂停（用于人工审批）
)
# 运行工作流（会在interrupt点暂停）
config = {"configurable": {"thread_id": "session-001"}}
result = persistent_app.invoke(
{"messages": [{"role": "user", "content": "查询订单状态"}]},
config=config
)
# 检查是否被中断
state = persistent_app.get_state(config)
if state.next: # 存在待执行的节点
# 人工审批后继续
persistent_app.invoke(None, config=config)
3. 定时任务与触发器
3.1 任务调度架构
┌──────────────────────────────────────────────┐
│ 调度器层 │
│ ┌─────────┐ ┌──────────┐ ┌─────────────┐ │
│ │ Cron表达式│ │ 事件触发 │ │ 消息队列触发 │ │
│ └────┬────┘ └────┬─────┘ └──────┬──────┘ │
│ └────────────┼───────────────┘ │
│ ▼ │
│ ┌──────────────┐ │
│ │ 任务分发器 │ │
│ └──────┬───────┘ │
└───────────────────┼──────────────────────────┘
▼
┌──────────────────────────────────────────────┐
│ 执行器层 │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ Agent执行 │ │ 数据处理 │ │ 通知发送 │ │
│ └──────────┘ └──────────┘ └──────────┘ │
└──────────────────────────────────────────────┘
3.2 Cron调度实现
import schedule
import time
import threading
from datetime import datetime, timedelta
from typing import Callable
class AgentScheduler:
"""Agent任务调度器"""
def __init__(self):
self.jobs: dict[str, dict] = {}
self._running = False
self._thread = None
def add_cron_job(self, name: str, task: Callable,
schedule_expr: str, **kwargs):
"""
添加Cron风格的定时任务
Args:
name: 任务名称
task: 任务函数
schedule_expr: 调度表达式，如 "daily:09:00", "hourly", "weekly:mon:10:00"
"""
parts = schedule_expr.split(":")
trigger_type = parts[0]
job_schedule = schedule.every(1)
if trigger_type == "daily":
time_str = parts[1] if len(parts) > 1 else "00:00"
job_schedule = schedule.every().day.at(time_str)
elif trigger_type == "hourly":
job_schedule = schedule.every().hour
elif trigger_type == "weekly":
day = parts[1] if len(parts) > 1 else "monday"
time_str = parts[2] if len(parts) > 2 else "09:00"
day_method = getattr(schedule.every(), day)
job_schedule = day_method.at(time_str)
elif trigger_type == "interval":
minutes = int(parts[1]) if len(parts) > 1 else 30
job_schedule = schedule.every(minutes).minutes
job_schedule.do(self._execute_job, name=name, task=task, **kwargs)
self.jobs[name] = {
"schedule": schedule_expr,
"task": task,
"last_run": None,
"run_count": 0,
"errors": []
}
print(f" ⏰ 已添加定时任务: {name} ({schedule_expr})")
def _execute_job(self, name: str, task: Callable, **kwargs):
"""执行任务并记录状态"""
job = self.jobs[name]
print(f"\n 🚀 执行定时任务: {name} @ {datetime.now()}")
try:
result = task(**kwargs)
job["last_run"] = datetime.now()
job["run_count"] += 1
job["last_result"] = result
print(f" ✅ 任务完成: {name}")
except Exception as e:
job["errors"].append({
"time": datetime.now(),
"error": str(e)
})
print(f" ❌ 任务失败: {name} - {e}")
def start(self):
"""启动调度器（后台线程）"""
self._running = True
self._thread = threading.Thread(target=self._run_loop, daemon=True)
self._thread.start()
print(" 🟢 调度器已启动")
def _run_loop(self):
while self._running:
schedule.run_pending()
time.sleep(1)
def stop(self):
self._running = False
print(" 🔴 调度器已停止")
def get_status(self) -> dict:
return {
name: {
"schedule": info["schedule"],
"last_run": str(info["last_run"]),
"run_count": info["run_count"],
"error_count": len(info["errors"])
}
for name, info in self.jobs.items()
}
# 使用示例
scheduler = AgentScheduler()
def daily_report():
"""每日报告任务"""
return "日报生成完成"
def check_alerts():
"""检查告警"""
return "无新告警"
scheduler.add_cron_job("daily_report", daily_report, "daily:09:00")
scheduler.add_cron_job("alert_check", check_alerts, "interval:30")
scheduler.start()
3.3 事件驱动触发器
from abc import ABC, abstractmethod
from typing import Callable, Any
import threading
from collections import defaultdict
class EventBus:
"""事件总线"""
def __init__(self):
self._handlers: dict[str, list[Callable]] = defaultdict(list)
self._filters: dict[str, list[Callable]] = defaultdict(list)
def on(self, event: str, handler: Callable,
filter_fn: Callable = None):
"""注册事件处理器"""
self._handlers[event].append(handler)
if filter_fn:
self._filters[event].append(filter_fn)
return self
def emit(self, event: str, data: dict = None):
"""触发事件"""
print(f" 📡 事件触发: {event}")
data = data or {}
for handler in self._handlers.get(event, []):
# 检查过滤器
filters = self._filters.get(event, [])
if filters and not all(f(data) for f in filters):
continue
# 异步执行处理器
threading.Thread(
target=self._safe_execute,
args=(handler, data),
daemon=True
).start()
def _safe_execute(self, handler: Callable, data: dict):
try:
handler(data)
except Exception as e:
print(f" ❌ 事件处理失败: {e}")
# 使用示例
event_bus = EventBus()
# 注册处理器
event_bus.on("new_order", lambda d: print(f" 处理新订单: {d.get('order_id')}"))
event_bus.on("new_order", lambda d: print(f" 发送订单通知"))
event_bus.on("user_signup", lambda d: print(f" 发送欢迎邮件给: {d.get('email')}"),
filter_fn=lambda d: d.get("email") is not None)
# 触发事件
event_bus.emit("new_order", {"order_id": "ORD-001", "amount": 299.0})
event_bus.emit("user_signup", {"email": "user@example.com", "name": "张三"})
4. 条件分支与循环
4.1 条件分支引擎
from typing import Any, Callable
from dataclasses import dataclass
@dataclass
class Condition:
"""条件定义"""
name: str
check: Callable[[dict], bool]
true_branch: str
false_branch: str
class ConditionalWorkflow:
"""支持复杂条件分支的工作流"""
def __init__(self):
self.nodes: dict[str, dict] = {}
self.conditions: dict[str, Condition] = {}
def add_action(self, name: str, handler: Callable):
"""添加动作节点"""
self.nodes[name] = {"type": "action", "handler": handler}
return self
def add_condition(self, name: str, check: Callable,
true_branch: str, false_branch: str):
"""添加条件节点"""
self.conditions[name] = Condition(
name=name,
check=check,
true_branch=true_branch,
false_branch=false_branch
)
self.nodes[name] = {"type": "condition"}
return self
def add_end(self, name: str):
"""添加结束节点"""
self.nodes[name] = {"type": "end"}
return self
def execute(self, start_node: str, context: dict) -> dict:
"""执行工作流"""
current = start_node
path = []
while current:
path.append(current)
node = self.nodes.get(current)
if not node:
raise ValueError(f"节点不存在: {current}")
if node["type"] == "end":
print(f" 🏁 到达结束节点: {current}")
break
elif node["type"] == "action":
print(f" ▶ 执行动作: {current}")
result = node["handler"](context)
if isinstance(result, dict):
context.update(result)
# 动作节点后需要指定next
current = context.pop("_next", None)
elif node["type"] == "condition":
condition = self.conditions[current]
result = condition.check(context)
print(f" ❓ 条件 '{current}': {'True' if result else 'False'}")
current = condition.true_branch if result else condition.false_branch
context["_execution_path"] = path
return context
# 使用示例：内容审核工作流
workflow = ConditionalWorkflow()
workflow.add_action("receive_content", lambda ctx: {
"content": ctx.get("user_input", ""),
"_next": "check_length"
})
workflow.add_condition(
"check_length",
check=lambda ctx: len(ctx.get("content", "")) > 10,
true_branch="analyze_sentiment",
false_branch="reject_short"
)
workflow.add_action("analyze_sentiment", lambda ctx: {
"sentiment": "positive", # 实际应调用LLM
"toxicity": 0.1,
"_next": "check_toxicity"
})
workflow.add_condition(
"check_toxicity",
check=lambda ctx: ctx.get("toxicity", 1.0) < 0.5,
true_branch="approve",
false_branch="reject_toxic"
)
workflow.add_action("approve", lambda ctx: {"status": "approved", "_next": "end"})
workflow.add_action("reject_short", lambda ctx: {"status": "rejected", "reason": "内容过短", "_next": "end"})
workflow.add_action("reject_toxic", lambda ctx: {"status": "rejected", "reason": "内容不合规", "_next": "end"})
workflow.add_end("end")
result = workflow.execute("receive_content", {"user_input": "这是一段需要审核的内容"})
print(f"结果: {result['status']}")
print(f"执行路径: {' → '.join(result['_execution_path'])}")
4.2 循环控制
class LoopWorkflow:
"""支持循环的工作流"""
def __init__(self, max_iterations: int = 10):
self.max_iterations = max_iterations
def while_loop(self, condition: Callable[[dict], bool],
body: Callable[[dict], dict],
context: dict) -> dict:
"""While循环"""
iteration = 0
while condition(context) and iteration < self.max_iterations:
iteration += 1
print(f" 🔄 循环第{iteration}次")
context = body(context)
context["_iteration"] = iteration
if iteration >= self.max_iterations:
context["_warning"] = f"达到最大循环次数({self.max_iterations})"
return context
def for_each(self, items: list,
handler: Callable[[dict, Any], dict],
context: dict) -> dict:
"""遍历循环"""
results = []
for i, item in enumerate(items):
print(f" 📋 处理第{i+1}/{len(items)}项")
ctx = {**context, "_current_item": item, "_index": i}
result = handler(ctx, item)
results.append(result)
context["_results"] = results
return context
# 示例：迭代优化循环
loop = LoopWorkflow(max_iterations=5)
def check_quality(ctx: dict) -> bool:
return ctx.get("quality_score", 0) < 90
def improve_content(ctx: dict) -> dict:
current_score = ctx.get("quality_score", 60)
return {
"quality_score": current_score + 10,
"content": f"改进后的版本(v{ctx.get('_iteration', 0) + 1})"
}
result = loop.while_loop(
condition=check_quality,
body=improve_content,
context={"quality_score": 60, "content": "初始版本"}
)
print(f"最终质量分数: {result['quality_score']}")
5. 错误处理与重试
5.1 分层错误处理
import time
import logging
from enum import Enum
from dataclasses import dataclass, field
from typing import Callable, Optional
logger = logging.getLogger(__name__)
class ErrorSeverity(Enum):
LOW = "low" # 可忽略，继续执行
MEDIUM = "medium" # 需要重试
HIGH = "high" # 需要降级处理
CRITICAL = "critical" # 需要立即终止
@dataclass
class RetryPolicy:
"""重试策略"""
max_retries: int = 3
base_delay: float = 1.0
max_delay: float = 60.0
exponential_base: float = 2.0
retryable_errors: list = field(default_factory=lambda: [
ConnectionError, TimeoutError
])
@dataclass
class ErrorHandler:
"""工作流错误处理器"""
retry_policy: RetryPolicy = field(default_factory=RetryPolicy)
fallback_handler: Optional[Callable] = None
error_callbacks: list = field(default_factory=list)
def handle_with_retry(self, func: Callable, *args, **kwargs):
"""带重试的执行"""
last_error = None
for attempt in range(self.retry_policy.max_retries + 1):
try:
return func(*args, **kwargs)
except tuple(self.retry_policy.retryable_errors) as e:
last_error = e
delay = min(
self.retry_policy.base_delay * (
self.retry_policy.exponential_base ** attempt
),
self.retry_policy.max_delay
)
logger.warning(
f"第{attempt+1}次尝试失败: {e}，"
f"{delay}秒后重试"
)
time.sleep(delay)
except Exception as e:
# 非可重试错误
last_error = e
severity = self._classify_error(e)
if severity == ErrorSeverity.CRITICAL:
raise
elif severity == ErrorSeverity.HIGH and self.fallback_handler:
logger.error(f"使用降级处理: {e}")
return self.fallback_handler(*args, **kwargs)
else:
raise
# 所有重试都失败
if self.fallback_handler:
logger.error(f"重试耗尽，使用降级处理")
return self.fallback_handler(*args, **kwargs)
raise last_error
def _classify_error(self, error: Exception) -> ErrorSeverity:
"""分类错误严重程度"""
if isinstance(error, (PermissionError, SecurityError)):
return ErrorSeverity.CRITICAL
if isinstance(error, (ValueError, TypeError)):
return ErrorSeverity.HIGH
if isinstance(error, TimeoutError):
return ErrorSeverity.MEDIUM
return ErrorSeverity.LOW
class SecurityError(Exception):
pass
# 使用示例
def api_fallback(*args, **kwargs):
"""降级处理：返回缓存数据或默认值"""
return {"result": "降级响应：服务暂时不可用，请稍后重试"}
error_handler = ErrorHandler(
retry_policy=RetryPolicy(max_retries=3, base_delay=1.0),
fallback_handler=api_fallback
)
# 执行
result = error_handler.handle_with_retry(
lambda: {"result": "正常响应"} # 模拟API调用
)
5.2 工作流级别的断路器
import time
from threading import Lock
class CircuitBreaker:
"""断路器模式"""
# 状态常量
CLOSED = "closed" # 正常状态
OPEN = "open" # 断开状态（拒绝请求）
HALF_OPEN = "half_open" # 半开状态（试探性放行）
def __init__(self, failure_threshold: int = 5,
recovery_timeout: int = 30,
success_threshold: int = 2):
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.success_threshold = success_threshold
self.state = self.CLOSED
self.failure_count = 0
self.success_count = 0
self.last_failure_time = 0
self._lock = Lock()
def execute(self, func: Callable, fallback: Callable = None):
"""通过断路器执行函数"""
with self._lock:
if self.state == self.OPEN:
if time.time() - self.last_failure_time > self.recovery_timeout:
self.state = self.HALF_OPEN
print(" 🔄 断路器: OPEN → HALF_OPEN")
else:
print(" 🚫 断路器: OPEN，请求被拒绝")
if fallback:
return fallback()
raise CircuitBreakerOpenError("断路器已打开")
try:
result = func()
with self._lock:
if self.state == self.HALF_OPEN:
self.success_count += 1
if self.success_count >= self.success_threshold:
self.state = self.CLOSED
self.failure_count = 0
self.success_count = 0
print(" ✅ 断路器: HALF_OPEN → CLOSED")
elif self.state == self.CLOSED:
self.failure_count = 0
return result
except Exception as e:
with self._lock:
self.failure_count += 1
self.last_failure_time = time.time()
if self.state == self.HALF_OPEN:
self.state = self.OPEN
print(" ❌ 断路器: HALF_OPEN → OPEN")
elif self.failure_count >= self.failure_threshold:
self.state = self.OPEN
print(f" ❌ 断路器: CLOSED → OPEN "
f"(连续{self.failure_count}次失败)")
raise
class CircuitBreakerOpenError(Exception):
pass
# 使用示例
breaker = CircuitBreaker(failure_threshold=3, recovery_timeout=60)
def unreliable_api():
"""模拟不稳定的API"""
import random
if random.random() < 0.5:
raise ConnectionError("API超时")
return {"status": "ok"}
for i in range(10):
try:
result = breaker.execute(
unreliable_api,
fallback=lambda: {"status": "降级响应"}
)
print(f" 请求{i+1}: {result}")
except CircuitBreakerOpenError:
print(f" 请求{i+1}: 断路器打开，跳过")
time.sleep(1)
6. 人工审批节点
6.1 审批工作流引擎
import uuid
import json
import time
from enum import Enum
from dataclasses import dataclass, field
from typing import Callable, Optional
from datetime import datetime
class ApprovalAction(Enum):
APPROVE = "approve"
REJECT = "reject"
REVISE = "revise" # 要求修改
ESCALATE = "escalate" # 升级审批
@dataclass
class ApprovalRequest:
id: str
workflow_id: str
node_name: str
context: dict
description: str
options: list[str]
created_at: datetime
status: str = "pending"
decision: Optional[str] = None
comment: Optional[str] = None
decided_at: Optional[datetime] = None
decided_by: Optional[str] = None
class ApprovalEngine:
"""人工审批引擎"""
def __init__(self):
self.pending: dict[str, ApprovalRequest] = {}
self.history: list[ApprovalRequest] = []
self.notification_handler: Optional[Callable] = None
def set_notification_handler(self, handler: Callable):
"""设置通知处理器（如邮件、Slack、钉钉）"""
self.notification_handler = handler
def create_request(self, workflow_id: str, node_name: str,
context: dict, description: str,
options: list[str] = None) -> str:
"""创建审批请求"""
req_id = f"APR-{uuid.uuid4().hex[:8].upper()}"
request = ApprovalRequest(
id=req_id,
workflow_id=workflow_id,
node_name=node_name,
context=context,
description=description,
options=options or [
ApprovalAction.APPROVE.value,
ApprovalAction.REJECT.value
],
created_at=datetime.now()
)
self.pending[req_id] = request
# 发送通知
if self.notification_handler:
self.notification_handler(request)
print(f" 📋 审批请求已创建: {req_id}")
print(f" 📝 {description}")
return req_id
def submit_decision(self, request_id: str, action: str,
comment: str = "", decided_by: str = "system"):
"""提交审批决定"""
if request_id not in self.pending:
raise ValueError(f"审批请求不存在: {request_id}")
request = self.pending.pop(request_id)
request.status = action
request.decision = action
request.comment = comment
request.decided_at = datetime.now()
request.decided_by = decided_by
self.history.append(request)
emoji = {"approve": "✅", "reject": "❌", "revise": "📝", "escalate": "⬆️"}
print(f" {emoji.get(action, '❓')} 审批决定: {action} ({request_id})")
return request
def get_pending_summary(self) -> list[dict]:
"""获取待审批列表"""
return [
{
"id": req.id,
"workflow": req.workflow_id,
"node": req.node_name,
"description": req.description,
"waiting_since": req.created_at.isoformat(),
"options": req.options
}
for req in self.pending.values()
]
# 使用示例
approval_engine = ApprovalEngine()
# 设置通知（发送到钉钉/Slack等）
def send_notification(request: ApprovalRequest):
print(f" 📧 通知已发送: 审批请求 {request.id} 等待处理")
approval_engine.set_notification_handler(send_notification)
# 在工作流中使用
class WorkflowWithApproval:
"""带审批节点的工作流"""
def __init__(self, engine: ApprovalEngine):
self.engine = engine
self.workflows: dict[str, dict] = {}
def execute_with_approval(self, workflow_id: str,
context: dict,
auto_steps: list[Callable],
approval_node: dict,
post_approval_steps: list[Callable]) -> dict:
"""
执行带审批的工作流
执行流程: 自动步骤 → 审批节点 → 审批后步骤
"""
# 1. 执行前置自动步骤
for i, step in enumerate(auto_steps):
print(f" ▶ 自动步骤 {i+1}")
result = step(context)
if isinstance(result, dict):
context.update(result)
# 2. 到达审批节点
request_id = self.engine.create_request(
workflow_id=workflow_id,
node_name=approval_node.get("name", "approval"),
context=context,
description=approval_node.get("description", "请审批"),
options=approval_node.get("options")
)
context["_approval_id"] = request_id
context["_approval_status"] = "pending"
# 3. 模拟等待审批（实际中这里是异步的）
# 在真实场景中，工作流会在此暂停，等待外部审批
print(f" ⏳ 工作流暂停，等待审批 {request_id}...")
return context
7. 多系统集成
7.1 统一集成框架
from abc import ABC, abstractmethod
from typing import Any
from dataclasses import dataclass, field
class SystemConnector(ABC):
"""系统连接器基类"""
@abstractmethod
def connect(self) -> bool:
pass
@abstractmethod
def health_check(self) -> bool:
pass
@abstractmethod
def execute(self, action: str, params: dict) -> Any:
pass
class IntegrationHub:
"""集成中心：统一管理系统连接器"""
def __init__(self):
self.connectors: dict[str, SystemConnector] = {}
self.action_mappings: dict[str, tuple[str, str]] = {} # action -> (connector, method)
def register(self, name: str, connector: SystemConnector):
"""注册系统连接器"""
connector.connect()
self.connectors[name] = connector
print(f" 🔌 已注册系统: {name}")
return self
def map_action(self, action_name: str, connector_name: str, method: str):
"""映射动作到具体的系统方法"""
self.action_mappings[action_name] = (connector_name, method)
return self
def execute(self, action: str, params: dict) -> Any:
"""通过动作名执行系统调用"""
if action not in self.action_mappings:
raise ValueError(f"未映射的动作: {action}")
connector_name, method = self.action_mappings[action]
connector = self.connectors.get(connector_name)
if not connector:
raise ValueError(f"系统未注册: {connector_name}")
if not connector.health_check():
raise ConnectionError(f"系统不可用: {connector_name}")
return connector.execute(method, params)
# 示例：CRM连接器
class CRMConnector(SystemConnector):
def __init__(self, api_url: str):
self.api_url = api_url
def connect(self) -> bool:
print(f" 连接CRM: {self.api_url}")
return True
def health_check(self) -> bool:
return True
def execute(self, action: str, params: dict) -> Any:
if action == "get_customer":
return {"id": params.get("customer_id"), "name": "张三", "level": "VIP"}
elif action == "update_ticket":
return {"ticket_id": params.get("ticket_id"), "status": "updated"}
return {}
# 示例：邮件系统连接器
class EmailConnector(SystemConnector):
def __init__(self, smtp_host: str):
self.smtp_host = smtp_host
def connect(self) -> bool:
print(f" 连接邮件服务: {self.smtp_host}")
return True
def health_check(self) -> bool:
return True
def execute(self, action: str, params: dict) -> Any:
if action == "send":
print(f" 📧 发送邮件: {params.get('to')} - {params.get('subject')}")
return {"sent": True, "message_id": "msg_001"}
return {}
# 使用示例
hub = IntegrationHub()
hub.register("crm", CRMConnector("https://crm.example.com/api"))
hub.register("email", EmailConnector("smtp.example.com"))
hub.map_action("lookup_customer", "crm", "get_customer")
hub.map_action("send_email", "email", "send")
# 在Agent工作流中使用
customer = hub.execute("lookup_customer", {"customer_id": "C001"})
hub.execute("send_email", {
"to": "user@example.com",
"subject": f"欢迎回来，{customer['name']}！",
"body": "感谢您继续使用我们的服务。"
})
7.2 MCP协议集成
from dataclasses import dataclass
from typing import Any
@dataclass
class MCPTool:
"""MCP工具定义"""
name: str
description: str
input_schema: dict
handler: Any # Callable
class MCPServer:
"""简化的MCP服务器实现"""
def __init__(self, name: str):
self.name = name
self.tools: dict[str, MCPTool] = {}
def register_tool(self, name: str, description: str,
input_schema: dict, handler: Callable):
"""注册MCP工具"""
self.tools[name] = MCPTool(
name=name,
description=description,
input_schema=input_schema,
handler=handler
)
return self
def list_tools(self) -> list[dict]:
"""列出所有工具（MCP tools/list）"""
return [
{
"name": tool.name,
"description": tool.description,
"inputSchema": tool.input_schema
}
for tool in self.tools.values()
]
def call_tool(self, name: str, arguments: dict) -> Any:
"""调用工具（MCP tools/call）"""
if name not in self.tools:
return {"error": f"工具不存在: {name}"}
tool = self.tools[name]
try:
return {"content": [{"type": "text", "text": str(tool.handler(**arguments))}]}
except Exception as e:
return {"error": str(e)}
# 创建MCP工具服务器
mcp_server = MCPServer("workflow-tools")
mcp_server.register_tool(
name="query_database",
description="查询数据库中的数据",
input_schema={
"type": "object",
"properties": {
"sql": {"type": "string", "description": "SQL查询语句"},
"database": {"type": "string", "description": "数据库名"}
},
"required": ["sql"]
},
handler=lambda sql, database="default": f"查询结果: [{sql}] 的执行结果"
)
mcp_server.register_tool(
name="send_notification",
description="发送通知消息",
input_schema={
"type": "object",
"properties": {
"channel": {"type": "string", "description": "通知渠道"},
"message": {"type": "string", "description": "消息内容"}
},
"required": ["channel", "message"]
},
handler=lambda channel, message: f"通知已发送到 {channel}"
)
8. 监控与日志
8.1 结构化日志
import json
import logging
import time
import uuid
from datetime import datetime
from contextlib import contextmanager
from typing import Optional
class StructuredLogger:
"""结构化日志器"""
def __init__(self, name: str, log_file: str = None):
self.name = name
self.logger = logging.getLogger(name)
self.logger.setLevel(logging.INFO)
# 控制台输出
console_handler = logging.StreamHandler()
console_handler.setFormatter(
logging.Formatter('%(asctime)s [%(levelname)s] %(message)s')
)
self.logger.addHandler(console_handler)
# 文件输出（JSON格式）
if log_file:
file_handler = logging.FileHandler(log_file)
file_handler.setFormatter(logging.Formatter('%(message)s'))
self.logger.addHandler(file_handler)
def log_event(self, event_type: str, data: dict = None,
level: str = "info", trace_id: str = None):
"""记录结构化事件"""
entry = {
"timestamp": datetime.now().isoformat(),
"logger": self.name,
"level": level,
"event_type": event_type,
"trace_id": trace_id or str(uuid.uuid4())[:8],
"data": data or {}
}
log_method = getattr(self.logger, level, self.logger.info)
log_method(json.dumps(entry, ensure_ascii=False))
return entry
@contextmanager
def trace(self, operation: str, trace_id: str = None):
"""上下文管理器：自动记录操作的开始和结束"""
trace_id = trace_id or str(uuid.uuid4())[:8]
start_time = time.time()
self.log_event(f"{operation}.start", {"trace_id": trace_id},
trace_id=trace_id)
try:
yield trace_id
duration = time.time() - start_time
self.log_event(f"{operation}.success",
{"duration_ms": round(duration * 1000, 2)},
trace_id=trace_id)
except Exception as e:
duration = time.time() - start_time
self.log_event(f"{operation}.error",
{"error": str(e), "duration_ms": round(duration * 1000, 2)},
level="error", trace_id=trace_id)
raise
# 使用示例
logger = StructuredLogger("workflow-engine", "workflow.log")
# 记录事件
logger.log_event("workflow.started", {"workflow_id": "WF-001", "input": "用户查询"})
# 追踪操作
with logger.trace("llm_call") as trace_id:
time.sleep(0.5) # 模拟API调用
logger.log_event("llm.tokens_used",
{"input": 500, "output": 200},
trace_id=trace_id)
8.2 工作流指标收集
import time
from collections import defaultdict
from dataclasses import dataclass, field
from threading import Lock
@dataclass
class WorkflowMetrics:
"""工作流指标收集器"""
_lock: Lock = field(default_factory=Lock)
_counters: dict = field(default_factory=lambda: defaultdict(int))
_timers: dict = field(default_factory=lambda: defaultdict(list))
_gauges: dict = field(default_factory=lambda: defaultdict(float))
def increment(self, name: str, value: int = 1, tags: dict = None):
"""递增计数器"""
key = self._make_key(name, tags)
with self._lock:
self._counters[key] += value
def record_duration(self, name: str, duration_ms: float, tags: dict = None):
"""记录耗时"""
key = self._make_key(name, tags)
with self._lock:
self._timers[key].append(duration_ms)
def set_gauge(self, name: str, value: float, tags: dict = None):
"""设置仪表盘值"""
key = self._make_key(name, tags)
with self._lock:
self._gauges[key] = value
@contextmanager
def timer(self, name: str, tags: dict = None):
"""计时上下文管理器"""
start = time.time()
try:
yield
finally:
duration = (time.time() - start) * 1000
self.record_duration(name, duration, tags)
def get_summary(self) -> dict:
"""获取指标汇总"""
with self._lock:
summary = {
"counters": dict(self._counters),
"timers": {},
"gauges": dict(self._gauges)
}
for key, durations in self._timers.items():
if durations:
summary["timers"][key] = {
"count": len(durations),
"avg_ms": round(sum(durations) / len(durations), 2),
"min_ms": round(min(durations), 2),
"max_ms": round(max(durations), 2),
"p95_ms": round(sorted(durations)[int(len(durations) * 0.95)], 2)
}
return summary
def _make_key(self, name: str, tags: dict = None) -> str:
if tags:
tag_str = ",".join(f"{k}={v}" for k, v in sorted(tags.items()))
return f"{name}{{{tag_str}}}"
return name
# 使用示例
metrics = WorkflowMetrics()
# 记录工作流执行
metrics.increment("workflow.executed", tags={"workflow": "customer_service"})
metrics.increment("workflow.success", tags={"workflow": "customer_service"})
with metrics.timer("workflow.duration", tags={"workflow": "customer_service"}):
time.sleep(0.3) # 模拟工作流执行
metrics.set_gauge("workflow.active_count", 5)
# 查看指标
summary = metrics.get_summary()
print(json.dumps(summary, indent=2, ensure_ascii=False))
9. 企业级Agent工作流架构
9.1 整体架构设计
┌─────────────────────────────────────────────────────────────────┐
│ 接入层 │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ REST API │ │ WebSocket│ │ 消息队列 │ │ Webhook │ │
│ └────┬─────┘ └────┬─────┘ └────┬─────┘ └────┬─────┘ │
│ └──────────────┼────────────┼──────────────┘ │
│ ▼ │
│ ┌──────────────┐ │
│ │ API Gateway │ (认证/限流/路由) │
│ └──────┬───────┘ │
└─────────────────────┼───────────────────────────────────────────┘
▼
┌─────────────────────────────────────────────────────────────────┐
│ 编排层 │
│ ┌──────────────────────────────────────────────────────────┐ │
│ │ 工作流引擎 (LangGraph / 自研) │ │
│ │ ┌─────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │
│ │ │ 状态管理 │ │ 条件路由 │ │ 并行执行 │ │ 审批节点 │ │ │
│ │ └─────────┘ └──────────┘ └──────────┘ └──────────┘ │ │
│ └──────────────────────────────────────────────────────────┘ │
│ │ │
│ ┌───────────────────┼───────────────────────────────────────┐ │
│ │ Agent 池 (可扩展) │ │
│ │ ┌────────┐ ┌────────┐ ┌────────┐ ┌────────┐ │ │
│ │ │ 分析师 │ │ 执行者 │ │ 审核员 │ │ 协调者 │ │ │
│ │ └────────┘ └────────┘ └────────┘ └────────┘ │ │
│ └───────────────────────────────────────────────────────────┘ │
└─────────────────────┼───────────────────────────────────────────┘
▼
┌─────────────────────────────────────────────────────────────────┐
│ 服务层 │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ LLM API │ │ 数据库 │ │ 缓存 │ │ 搜索引擎 │ │
│ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ 文件存储 │ │ 邮件服务 │ │ 监控告警 │ │ 第三方API │ │
│ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │
└─────────────────────────────────────────────────────────────────┘
9.2 Agent注册与管理
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Any, Optional
from enum import Enum
class AgentStatus(Enum):
IDLE = "idle"
BUSY = "busy"
ERROR = "error"
OFFLINE = "offline"
@dataclass
class AgentCapability:
"""Agent能力描述"""
name: str
description: str
input_types: list[str]
output_types: list[str]
max_concurrent: int = 1
class BaseAgent(ABC):
"""Agent基类"""
def __init__(self, agent_id: str, name: str):
self.agent_id = agent_id
self.name = name
self.status = AgentStatus.IDLE
self.capabilities: list[AgentCapability] = []
self.current_tasks: list[str] = []
@abstractmethod
def process(self, task: dict) -> dict:
"""处理任务"""
pass
@abstractmethod
def get_capabilities(self) -> list[AgentCapability]:
"""返回Agent能力列表"""
pass
class AgentRegistry:
"""Agent注册中心"""
def __init__(self):
self.agents: dict[str, BaseAgent] = {}
self.capability_index: dict[str, list[str]] = defaultdict(list)
def register(self, agent: BaseAgent):
"""注册Agent"""
self.agents[agent.agent_id] = agent
for cap in agent.get_capabilities():
self.capability_index[cap.name].append(agent.agent_id)
print(f" 🤖 已注册Agent: {agent.name} ({agent.agent_id})")
def find_agent(self, capability: str,
exclude_busy: bool = True) -> Optional[BaseAgent]:
"""查找具备指定能力的可用Agent"""
agent_ids = self.capability_index.get(capability, [])
for agent_id in agent_ids:
agent = self.agents[agent_id]
if exclude_busy and agent.status == AgentStatus.BUSY:
continue
return agent
return None
def get_status(self) -> dict:
"""获取所有Agent状态"""
return {
agent_id: {
"name": agent.name,
"status": agent.status.value,
"tasks": len(agent.current_tasks)
}
for agent_id, agent in self.agents.items()
}
# 示例Agent实现
class CustomerServiceAgent(BaseAgent):
def __init__(self):
super().__init__("cs-agent-001", "客服Agent")
def get_capabilities(self):
return [
AgentCapability(
name="customer_inquiry",
description="处理客户咨询",
input_types=["text"],
output_types=["text"]
)
]
def process(self, task: dict) -> dict:
self.status = AgentStatus.BUSY
try:
# 调用LLM处理客户问题
answer = f"已处理客户咨询: {task.get('message', '')}"
return {"status": "success", "answer": answer}
finally:
self.status = AgentStatus.IDLE
class DataAnalysisAgent(BaseAgent):
def __init__(self):
super().__init__("da-agent-001", "数据分析Agent")
def get_capabilities(self):
return [
AgentCapability(
name="data_analysis",
description="数据分析与可视化",
input_types=["text", "csv", "json"],
output_types=["text", "chart", "report"]
)
]
def process(self, task: dict) -> dict:
self.status = AgentStatus.BUSY
try:
return {"status": "success", "report": "分析报告内容"}
finally:
self.status = AgentStatus.IDLE
# 使用示例
registry = AgentRegistry()
registry.register(CustomerServiceAgent())
registry.register(DataAnalysisAgent())
# 查找合适的Agent处理任务
agent = registry.find_agent("customer_inquiry")
if agent:
result = agent.process({"message": "我想查询订单状态"})
print(result)
9.3 工作流持久化与恢复
import json
import os
from datetime import datetime
from pathlib import Path
class WorkflowPersistence:
"""工作流持久化管理"""
def __init__(self, storage_dir: str = "workflow_states"):
self.storage_dir = Path(storage_dir)
self.storage_dir.mkdir(parents=True, exist_ok=True)
def save_state(self, workflow_id: str, state: dict,
checkpoint_name: str = "latest"):
"""保存工作流状态"""
state_data = {
"workflow_id": workflow_id,
"checkpoint": checkpoint_name,
"saved_at": datetime.now().isoformat(),
"state": state
}
filepath = self.storage_dir / f"{workflow_id}_{checkpoint_name}.json"
with open(filepath, "w", encoding="utf-8") as f:
json.dump(state_data, f, ensure_ascii=False, indent=2)
print(f" 💾 状态已保存: {filepath}")
def load_state(self, workflow_id: str,
checkpoint_name: str = "latest") -> dict:
"""加载工作流状态"""
filepath = self.storage_dir / f"{workflow_id}_{checkpoint_name}.json"
if not filepath.exists():
raise FileNotFoundError(f"状态文件不存在: {filepath}")
with open(filepath, "r", encoding="utf-8") as f:
state_data = json.load(f)
print(f" 📂 状态已加载: {filepath}")
return state_data["state"]
def list_checkpoints(self, workflow_id: str) -> list[str]:
"""列出所有检查点"""
checkpoints = []
for f in self.storage_dir.glob(f"{workflow_id}_*.json"):
name = f.stem.replace(f"{workflow_id}_", "")
checkpoints.append(name)
return sorted(checkpoints)
10. 实战案例
10.1 案例一：智能客服系统
class CustomerServiceWorkflow:
"""
智能客服工作流
流程: 接收消息 → 意图识别 → 路由分发 → 处理 → 质检 → 回复
"""
def __init__(self, llm_client, knowledge_base, crm_system):
self.llm = llm_client
self.kb = knowledge_base
self.crm = crm_system
self.conversation_history: dict[str, list] = {}
def handle_message(self, customer_id: str, message: str) -> dict:
"""处理客户消息的完整流程"""
# 1. 获取对话历史
history = self.conversation_history.get(customer_id, [])
# 2. 意图识别
intent = self._classify_intent(message, history)
print(f" 🎯 识别意图: {intent}")
# 3. 路由到对应处理器
handlers = {
"order_inquiry": self._handle_order_inquiry,
"complaint": self._handle_complaint,
"product_question": self._handle_product_question,
"transfer_agent": self._handle_transfer,
"general": self._handle_general
}
handler = handlers.get(intent, self._handle_general)
result = handler(customer_id, message, history)
# 4. 质检：检查回复质量
quality = self._quality_check(result["reply"], message)
if quality["score"] < 70:
result["reply"] = self._improve_reply(
result["reply"], quality["issues"]
)
# 5. 更新历史
history.append({"role": "user", "content": message})
history.append({"role": "assistant", "content": result["reply"]})
self.conversation_history[customer_id] = history[-20:] # 保留最近20条
# 6. 记录到CRM
self.crm.execute("update_ticket", {
"customer_id": customer_id,
"intent": intent,
"message": message,
"reply": result["reply"]
})
return result
def _classify_intent(self, message: str, history: list) -> str:
"""意图识别"""
prompt = f"""根据以下对话内容，判断用户意图。
对话历史：{json.dumps(history[-4:], ensure_ascii=False)}
当前消息：{message}
可能的意图：order_inquiry, complaint, product_question, transfer_agent, general
只返回意图名称。"""
response = self.llm.generate(prompt)
intent = response.strip().lower()
valid_intents = ["order_inquiry", "complaint", "product_question",
"transfer_agent", "general"]
return intent if intent in valid_intents else "general"
def _handle_order_inquiry(self, customer_id: str, message: str,
history: list) -> dict:
"""处理订单查询"""
# 从CRM获取客户订单
orders = self.crm.execute("get_orders", {"customer_id": customer_id})
prompt = f"""你是客服助手。客户询问订单信息。
客户信息：{customer_id}
订单数据：{json.dumps(orders, ensure_ascii=False)}
客户消息：{message}
请提供清晰的订单状态回复。"""
reply = self.llm.generate(prompt)
return {"reply": reply, "intent": "order_inquiry", "escalated": False}
def _handle_complaint(self, customer_id: str, message: str,
history: list) -> dict:
"""处理投诉（需要特殊关注）"""
# 获取客户等级
customer = self.crm.execute("get_customer", {"customer_id": customer_id})
is_vip = customer.get("level") == "VIP"
prompt = f"""你是高级客服专员。客户正在投诉。
客户等级：{'VIP' if is_vip else '普通'}
投诉内容：{message}
请以同理心回应，表达歉意，并提供解决方案。
如果是VIP客户，提供额外补偿方案。"""
reply = self.llm.generate(prompt)
# VIP客户自动升级处理
escalated = is_vip
return {
"reply": reply,
"intent": "complaint",
"escalated": escalated,
"priority": "high" if is_vip else "medium"
}
def _handle_product_question(self, customer_id: str, message: str,
history: list) -> dict:
"""处理产品咨询"""
# 从知识库检索相关信息
kb_results = self.kb.search(message, top_k=3)
context = "\n".join([r["content"] for r in kb_results])
prompt = f"""你是产品专家。根据以下知识库内容回答客户问题。
知识库参考：
{context}
客户问题：{message}
请基于参考内容回答，如果参考内容不足以回答，请说明。"""
reply = self.llm.generate(prompt)
return {"reply": reply, "intent": "product_question", "escalated": False}
def _handle_transfer(self, customer_id: str, message: str,
history: list) -> dict:
"""转接人工客服"""
return {
"reply": "好的，正在为您转接人工客服，请稍候...",
"intent": "transfer_agent",
"escalated": True,
"action": "transfer_to_human"
}
def _handle_general(self, customer_id: str, message: str,
history: list) -> dict:
"""通用回复"""
prompt = f"""你是友好的客服助手。
对话历史：{json.dumps(history[-4:], ensure_ascii=False)}
客户消息：{message}
请提供友好、有帮助的回复。"""
reply = self.llm.generate(prompt)
return {"reply": reply, "intent": "general", "escalated": False}
def _quality_check(self, reply: str, original_message: str) -> dict:
"""回复质量检查"""
# 简化版本：检查长度和关键词
score = 100
issues = []
if len(reply) < 10:
score -= 30
issues.append("回复过短")
if len(reply) > 2000:
score -= 20
issues.append("回复过长")
if "抱歉" in original_message and "抱歉" not in reply:
score -= 10
issues.append("缺少同理心表达")
return {"score": score, "issues": issues}
def _improve_reply(self, reply: str, issues: list) -> str:
"""改进回复质量"""
prompt = f"""改进以下客服回复。
原始回复：{reply}
需要改进的问题：{', '.join(issues)}
请生成改进后的回复。"""
return self.llm.generate(prompt)
10.2 案例二：自动化数据分析流水线
class DataAnalysisWorkflow:
"""
自动化数据分析工作流
流程: 数据接入 → 清洗 → 探索分析 → 洞察生成 → 报告输出
"""
def __init__(self, llm_client):
self.llm = llm_client
def run_analysis(self, data_source: str,
analysis_goal: str) -> dict:
"""执行完整的数据分析流程"""
results = {"steps": []}
# 步骤1: 数据接入与概览
print(" 📊 步骤1: 数据接入")
data_overview = self._load_and_preview(data_source)
results["steps"].append({
"name": "数据接入",
"output": data_overview
})
# 步骤2: 数据清洗建议
print(" 🧹 步骤2: 数据清洗")
cleaning_plan = self._generate_cleaning_plan(data_overview)
results["steps"].append({
"name": "数据清洗",
"output": cleaning_plan
})
# 步骤3: 探索性分析
print(" 🔍 步骤3: 探索性分析")
exploration = self._exploratory_analysis(data_overview, analysis_goal)
results["steps"].append({
"name": "探索性分析",
"output": exploration
})
# 步骤4: 深度洞察
print(" 💡 步骤4: 洞察生成")
insights = self._generate_insights(exploration, analysis_goal)
results["steps"].append({
"name": "深度洞察",
"output": insights
})
# 步骤5: 生成报告
print(" 📝 步骤5: 报告生成")
report = self._generate_report(results, analysis_goal)
results["report"] = report
return results
def _load_and_preview(self, data_source: str) -> dict:
"""加载数据并生成预览"""
# 模拟数据加载
return {
"source": data_source,
"rows": 10000,
"columns": 15,
"column_types": {
"id": "int", "name": "string", "date": "datetime",
"amount": "float", "category": "string"
},
"missing_values": {"name": 5, "amount": 120},
"sample_data": "..."
}
def _generate_cleaning_plan(self, overview: dict) -> dict:
"""生成数据清洗计划"""
prompt = f"""基于以下数据概览，生成数据清洗计划：
数据规模：{overview['rows']}行 x {overview['columns']}列
缺失值：{json.dumps(overview['missing_values'])}
列类型：{json.dumps(overview['column_types'])}
请提供具体的清洗步骤建议。"""
plan = self.llm.generate(prompt)
return {"plan": plan}
def _exploratory_analysis(self, overview: dict, goal: str) -> dict:
"""生成探索性分析方案"""
prompt = f"""你是一位资深数据分析师。根据以下信息设计探索性分析方案。
数据概览：{json.dumps(overview, ensure_ascii=False)}
分析目标：{goal}
请提供：
1. 需要计算的关键统计指标
2. 建议的可视化图表类型
3. 需要验证的假设
4. 可能的分析维度"""
analysis = self.llm.generate(prompt)
return {"analysis_plan": analysis}
def _generate_insights(self, exploration: dict, goal: str) -> dict:
"""从分析结果中提取洞察"""
prompt = f"""基于探索性分析结果，提取关键商业洞察。
分析结果：{exploration['analysis_plan'][:500]}
分析目标：{goal}
请提供：
1. 3-5个最重要的发现
2. 每个发现的商业含义
3. 可行的行动建议
4. 需要进一步分析的方向"""
insights = self.llm.generate(prompt)
return {"insights": insights}
def _generate_report(self, results: dict, goal: str) -> str:
"""生成最终分析报告"""
steps_summary = "\n".join([
f"### {step['name']}\n{json.dumps(step['output'], ensure_ascii=False)[:200]}"
for step in results["steps"]
])
prompt = f"""生成一份专业的数据分析报告。
分析目标：{goal}
各步骤结果摘要：
{steps_summary}
报告格式要求：
## 执行摘要
（2-3句话概括核心发现）
## 分析背景
## 数据概况
## 关键发现
## 行动建议
## 附录"""
return self.llm.generate(prompt)
10.3 案例三：内容生成工厂
class ContentGenerationWorkflow:
"""
内容生成工厂工作流
支持多平台、多格式的内容批量生成与优化
"""
def __init__(self, llm_client):
self.llm = llm_client
self.platform_specs = {
"wechat_article": {
"max_length": 20000,
"style": "深度、有见解",
"format": "markdown"
},
"xiaohongshu": {
"max_length": 1000,
"style": "活泼、亲切、口语化",
"format": "短文+emoji"
},
"twitter": {
"max_length": 280,
"style": "简洁、有冲击力",
"format": "短文本"
},
"linkedin": {
"max_length": 3000,
"style": "专业、有深度",
"format": "专业文章"
}
}
def generate_content_suite(self, topic: str,
platforms: list[str] = None) -> dict:
"""为多个平台生成内容套件"""
platforms = platforms or ["wechat_article", "xiaohongshu"]
results = {"topic": topic, "contents": {}}
# 步骤1: 主题研究
print(" 📚 主题研究...")
research = self._research_topic(topic)
results["research"] = research
# 步骤2: 生成核心内容
print(" ✍️ 生成核心内容...")
core_content = self._generate_core_content(topic, research)
results["core_content"] = core_content
# 步骤3: 适配各平台
for platform in platforms:
print(f" 🔄 适配平台: {platform}")
spec = self.platform_specs.get(platform, {})
adapted = self._adapt_for_platform(
core_content, platform, spec
)
results["contents"][platform] = adapted
# 步骤4: 质量评审
print(" 🔍 质量评审...")
for platform, content in results["contents"].items():
review = self._review_content(content, platform)
results["contents"][platform] = {
"content": content,
"review": review
}
return results
def _research_topic(self, topic: str) -> dict:
"""主题研究"""
prompt = f"""对主题「{topic}」进行深度研究，提供：
1. 核心知识点（5-8个）
2. 目标受众画像
3. 当前热门角度
4. 竞品内容分析要点
5. 差异化切入建议
请用结构化的JSON格式返回。"""
result = self.llm.generate(prompt)
return {"research": result}
def _generate_core_content(self, topic: str, research: dict) -> str:
"""生成核心内容"""
prompt = f"""基于以下研究，撰写关于「{topic}」的核心内容。
研究摘要：{research['research'][:500]}
要求：
- 结构清晰，逻辑严密
- 包含实际案例或数据支撑
- 有独到见解，避免泛泛而谈
- 长度约3000字"""
return self.llm.generate(prompt)
def _adapt_for_platform(self, core_content: str,
platform: str, spec: dict) -> str:
"""将核心内容适配到特定平台"""
prompt = f"""将以下核心内容适配为{platform}平台的格式。
核心内容摘要：
{core_content[:1000]}
平台要求：
- 最大长度：{spec.get('max_length', '不限')}字
- 风格：{spec.get('style', '通用')}
- 格式：{spec.get('format', '文本')}
请直接输出适配后的内容。"""
return self.llm.generate(prompt)
def _review_content(self, content: str, platform: str) -> dict:
"""内容质量评审"""
prompt = f"""评审以下{platform}平台的内容质量。
内容：
{content[:500]}
评审维度（每项1-10分）：
1. 吸引力（标题/开头是否抓人）
2. 信息量（是否有足够价值）
3. 可读性（排版、节奏、表达）
4. 平台适配度（是否符合平台调性）
5. 传播性（是否容易引发分享）
返回JSON格式的评分和改进建议。"""
result = self.llm.generate(prompt)
return result
总结
本教程系统讲解了AI Agent工作流自动化的完整技术栈：
- 设计模式：顺序链、路由分发、并行扇出/汇聚、状态机、反思循环、人机协作六种核心模式
- 编排框架：LangChain的LCEL表达式语言和LangGraph状态图，支持持久化和断点续执行
- 调度触发：Cron定时、事件驱动、消息队列三种触发机制
- 控制流：条件分支引擎和循环控制，支持复杂的业务逻辑
- 容错机制：分层错误处理、指数退避重试、断路器模式
- 人工审批：完整的审批引擎，支持多级审批和超时处理
- 系统集成：统一集成框架和MCP协议，连接CRM、邮件、数据库等外部系统
- 可观测性：结构化日志、分布式追踪、指标收集
- 企业架构：分层架构设计、Agent注册中心、工作流持久化
- 实战案例：智能客服、数据分析、内容生成三个完整案例
掌握这些技术，你就能构建出可靠、可扩展、可观测的企业级Agent工作流系统。
