Agentic Workflow 智能工作流设计完全教程
从单 Agent 到多 Agent 协作，系统掌握智能工作流的设计模式、实现框架与企业级部署。
目录
- Agentic Workflow 概述与核心概念
- 工作流设计模式
- 多 Agent 协作架构设计
- 任务分解与规划策略
- 工具调用与外部集成
- 状态管理与上下文传递
- 错误处理与容错机制
- 人机协作（Human-in-the-Loop）
- 主流框架对比
- 企业级工作流部署与监控
1. Agentic Workflow 概述与核心概念
1.1 什么是 Agentic Workflow
Agentic Workflow（智能体工作流）是一种将大语言模型（LLM）作为核心推理引擎，通过自主决策、工具调用和多步推理来完成复杂任务的系统架构。与传统的确定性工作流不同，Agentic Workflow 具备动态决策能力——Agent 可以根据中间结果调整执行路径，而非遵循固定的流程图。
与传统工作流的本质区别：
传统工作流（确定性）：
输入 → 步骤A → 步骤B → 步骤C → 输出
（每一步都是预定义的，不会改变）
Agentic Workflow（非确定性）：
输入 → 思考 → 决策 → [可能调用工具] → 观察 → 再思考 → ... → 输出
（Agent 根据中间结果自主决定下一步行动）
1.2 核心概念
Agent（智能体）
Agent 是工作流的基本执行单元，它具备以下能力：
- 感知（Perception）：接收用户输入、环境状态和工具返回结果
- 推理（Reasoning）：使用 LLM 分析问题并制定行动计划
- 行动（Action）：调用工具、生成文本或触发其他 Agent
- 记忆（Memory）：维护短期（上下文窗口）和长期（外部存储）记忆
Tool（工具）
工具是 Agent 与外部世界交互的接口：
# 工具的标准定义格式
tools = [
{
"type": "function",
"function": {
"name": "search_web",
"description": "搜索互联网获取最新信息",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "搜索关键词"
},
"num_results": {
"type": "integer",
"description": "返回结果数量",
"default": 5
}
},
"required": ["query"]
}
}
}
]
Planning（规划）
规划是 Agent 将复杂任务分解为可执行步骤的能力：
class Planner:
"""任务规划器"""
def __init__(self, llm):
self.llm = llm
def create_plan(self, task: str, context: dict = None) -> list[dict]:
"""将任务分解为执行步骤"""
prompt = f"""你是一个任务规划专家。请将以下任务分解为清晰的执行步骤。
任务：{task}
上下文：{context or '无'}
请以JSON格式输出步骤列表，每个步骤包含：
- step_id: 步骤编号
- action: 具体行动描述
- tool: 需要使用的工具（如果有的话）
- depends_on: 依赖的前置步骤编号列表
- expected_output: 预期输出描述
输出："""
response = self.llm.generate(prompt)
return self.parse_plan(response)
def revise_plan(self, plan: list[dict], feedback: str) -> list[dict]:
"""根据反馈修改计划"""
prompt = f"""当前执行计划遇到问题，请根据反馈修改计划。
当前计划：{json.dumps(plan, ensure_ascii=False)}
反馈/错误：{feedback}
请输出修改后的计划："""
response = self.llm.generate(prompt)
return self.parse_plan(response)
1.3 Agentic AI 的演进路径
Level 1: 简单问答
用户 → LLM → 回答
（无工具、无记忆、无规划）
Level 2: 工具增强
用户 → LLM → [工具调用] → LLM → 回答
（有工具，但单步执行）
Level 3: ReAct 循环
用户 → [思考→行动→观察]×N → 回答
（多步推理，动态决策）
Level 4: 多 Agent 协作
用户 → 协调Agent → [Agent A, Agent B, Agent C] → 协调Agent → 回答
（分工协作，专业互补）
Level 5: 自主工作流
用户 → [目标设定→规划→执行→反思→调整]×N → 结果
（自主规划、自我纠错、持续优化）
2. 工作流设计模式
2.1 顺序执行模式（Sequential）
最简单的模式，步骤按顺序依次执行：
class SequentialWorkflow:
"""顺序执行工作流"""
def __init__(self, agents: list):
self.agents = agents
def execute(self, initial_input: str) -> str:
result = initial_input
for agent in self.agents:
result = agent.run(result)
print(f"[{agent.name}] 完成，输出长度: {len(result)}")
return result
# 示例：内容创作流水线
workflow = SequentialWorkflow([
ResearchAgent(), # 步骤1：资料收集
OutlineAgent(), # 步骤2：大纲生成
WritingAgent(), # 步骤3：内容撰写
ReviewAgent(), # 步骤4：质量审核
FormattingAgent(), # 步骤5：格式排版
])
result = workflow.execute("写一篇关于量子计算的技术博客")
适用场景： 翻译流水线、数据处理管道、文档生成流程。
2.2 并行执行模式（Parallel）
多个 Agent 同时执行独立任务，最后汇总结果：
import asyncio
from concurrent.futures import ThreadPoolExecutor
class ParallelWorkflow:
"""并行执行工作流"""
def __init__(self, agents: list, aggregator):
self.agents = agents
self.aggregator = aggregator
async def execute(self, input_data: str) -> str:
# 并行执行所有 Agent
tasks = [
asyncio.create_task(agent.arun(input_data))
for agent in self.agents
]
results = await asyncio.gather(*tasks)
# 汇总结果
return self.aggregator.merge(results)
# 示例：多维度市场分析
workflow = ParallelWorkflow(
agents=[
CompetitorAnalysisAgent(), # 竞品分析
MarketTrendAgent(), # 市场趋势
CustomerInsightAgent(), # 客户洞察
TechnicalFeasibilityAgent(), # 技术可行性
],
aggregator=ReportAggregator()
)
report = asyncio.run(workflow.execute("分析智能手表市场机会"))
适用场景： 多源信息聚合、并行测试、多角度分析。
2.3 条件分支模式（Conditional）
根据中间结果动态选择执行路径：
class ConditionalWorkflow:
"""条件分支工作流"""
def __init__(self):
self.routes = {}
self.default_route = None
def add_route(self, condition, agent):
"""添加条件路由"""
self.routes[condition] = agent
def set_default(self, agent):
self.default_route = agent
def execute(self, input_data: str) -> str:
# 路由决策
route_key = self.classify_input(input_data)
if route_key in self.routes:
agent = self.routes[route_key]
print(f"路由到: {agent.name}")
return agent.run(input_data)
elif self.default_route:
return self.default_route.run(input_data)
else:
raise ValueError(f"无匹配路由: {route_key}")
def classify_input(self, text: str) -> str:
"""使用 LLM 进行意图分类"""
prompt = f"""将以下用户请求分类为一个类别：
用户请求：{text}
可选类别：
- code_review: 代码审查相关
- bug_fix: Bug修复相关
- feature: 新功能开发
- documentation: 文档撰写
- general: 其他
类别："""
return llm.generate(prompt).strip()
# 示例：智能客服路由
workflow = ConditionalWorkflow()
workflow.add_route("technical", TechnicalSupportAgent())
workflow.add_route("billing", BillingAgent())
workflow.add_route("sales", SalesAgent())
workflow.set_default(GeneralAgent())
response = workflow.execute("我的订单为什么还没发货？")
2.4 循环模式（Loop）
Agent 反复执行直到满足退出条件：
class LoopWorkflow:
"""循环工作流 - 带退出条件的迭代执行"""
def __init__(self, agent, max_iterations=10, quality_threshold=0.8):
self.agent = agent
self.max_iterations = max_iterations
self.quality_threshold = quality_threshold
def execute(self, input_data: str) -> str:
result = input_data
history = []
for i in range(self.max_iterations):
# 执行当前迭代
result = self.agent.run(result, iteration=i, history=history)
history.append(result)
# 评估质量
quality = self.evaluate_quality(result, input_data)
print(f"迭代 {i+1}: 质量评分 = {quality:.2f}")
if quality >= self.quality_threshold:
print(f"达到质量阈值，退出循环")
return result
print(f"达到最大迭代次数 {self.max_iterations}")
return result
def evaluate_quality(self, result: str, original: str) -> float:
"""使用 LLM 评估输出质量"""
prompt = f"""评估以下输出的质量（0-1分）：
原始需求：{original}
当前输出：{result}
评分标准：
- 完整性：是否涵盖了所有要点
- 准确性：内容是否正确
- 清晰度：表达是否清晰易懂
- 实用性：是否可以直接使用
请只输出一个0-1之间的小数："""
try:
return float(llm.generate(prompt).strip())
except:
return 0.5
# 示例：迭代优化写作
workflow = LoopWorkflow(
agent=WritingRefinerAgent(),
max_iterations=5,
quality_threshold=0.9
)
final_article = workflow.execute("写一篇关于AI Agent的技术文章")
2.5 层级委托模式（Hierarchical）
上级 Agent 将任务委托给下级 Agent：
class HierarchicalWorkflow:
"""层级委托工作流"""
def __init__(self):
self.manager = ManagerAgent()
self.workers = {}
def register_worker(self, name: str, agent):
self.workers[name] = agent
def execute(self, task: str) -> str:
# 管理者分解任务
subtasks = self.manager.decompose(task, list(self.workers.keys()))
results = {}
for subtask in subtasks:
worker_name = subtask["assigned_to"]
worker = self.workers[worker_name]
# 如果子任务还需要进一步分解
if subtask.get("complexity", "low") == "high":
sub_result = self._delegate(worker, subtask["description"])
else:
sub_result = worker.run(subtask["description"])
results[subtask["id"]] = sub_result
# 管理者整合结果
return self.manager.integrate(task, results)
# 示例：软件开发团队
workflow = HierarchicalWorkflow()
workflow.register_worker("architect", ArchitectAgent())
workflow.register_worker("frontend", FrontendAgent())
workflow.register_worker("backend", BackendAgent())
workflow.register_worker("tester", QAAgent())
result = workflow.execute("开发一个用户登录系统")
2.6 设计模式选型指南
任务特征 推荐模式 示例
─────────────────────────────────────────────────────
步骤固定、线性依赖 顺序执行 文档翻译流水线
多个独立子任务 并行执行 多源数据采集
需要根据输入动态路由 条件分支 智能客服分流
需要迭代优化直到达标 循环模式 代码生成与调试
任务复杂需要分工 层级委托 软件项目开发
混合特征 嵌套组合 企业级自动化流程
3. 多 Agent 协作架构设计
3.1 协作模式
模式一：辩论式协作
多个 Agent 从不同角度分析问题，通过"辩论"达成共识：
class DebateWorkflow:
"""辩论式多 Agent 协作"""
def __init__(self, agents: list, judge):
self.agents = agents # 参与辩论的 Agent
self.judge = judge # 裁判 Agent
def execute(self, topic: str, rounds: int = 3) -> str:
debate_history = []
for round_num in range(rounds):
round_responses = []
for agent in self.agents:
# 每个 Agent 考虑之前所有发言
prompt = f"""讨论主题：{topic}
之前的讨论：
{self._format_history(debate_history)}
请从你的专业角度发表观点。可以支持、反对或补充之前的发言。
你的角色：{agent.role}"""
response = agent.run(prompt)
round_responses.append({
"agent": agent.name,
"round": round_num + 1,
"response": response
})
debate_history.extend(round_responses)
# 裁判总结
summary_prompt = f"""讨论主题：{topic}
以下是各位专家的讨论记录：
{self._format_history(debate_history)}
请综合各方观点，给出最终结论和建议："""
return self.judge.run(summary_prompt)
# 示例：技术方案评审
debate = DebateWorkflow(
agents=[
SecurityExpertAgent(), # 安全专家
PerformanceExpertAgent(), # 性能专家
CostExpertAgent(), # 成本专家
],
judge=ChiefArchitectAgent()
)
decision = debate.execute("选择数据库方案：PostgreSQL vs MongoDB vs TiDB")
模式二：审查式协作
一个 Agent 生成，另一个 Agent 审查并提出改进：
class ReviewWorkflow:
"""审查式协作 - 生成者 + 审查者"""
def __init__(self, generator, reviewer, max_revisions=3):
self.generator = generator
self.reviewer = reviewer
self.max_revisions = max_revisions
def execute(self, task: str) -> str:
# 生成初稿
draft = self.generator.run(task)
for revision in range(self.max_revisions):
# 审查
review = self.reviewer.run(f"""请审查以下内容：
原始任务：{task}
当前内容：
{draft}
请指出问题并给出具体修改建议：""")
# 检查是否需要修改
if self._is_approved(review):
print(f"审查通过（第 {revision + 1} 轮）")
return draft
# 根据审查意见修改
draft = self.generator.run(f"""请根据审查意见修改内容：
原始任务：{task}
当前内容：
{draft}
审查意见：
{review}
请输出修改后的完整内容：""")
return draft
def _is_approved(self, review: str) -> bool:
"""判断审查是否通过"""
approval_keywords = ["通过", "合格", "优秀", "LGTM", "approved"]
return any(kw in review for kw in approval_keywords)
模式三：专家委员会
多个专家 Agent 各自独立完成任务，投票选出最佳方案：
class CommitteeWorkflow:
"""专家委员会模式"""
def __init__(self, experts: list, coordinator):
self.experts = experts
self.coordinator = coordinator
def execute(self, task: str) -> dict:
# 各专家独立完成任务
proposals = {}
for expert in self.experts:
proposal = expert.run(task)
proposals[expert.name] = proposal
# 协调者评估并选择最佳方案
evaluation_prompt = f"""任务：{task}
以下是各位专家的方案：
{self._format_proposals(proposals)}
请评估每个方案的优缺点，并选择最佳方案或组合各方案的长处，
给出最终推荐。"""
final = self.coordinator.run(evaluation_prompt)
return {
"proposals": proposals,
"final_recommendation": final
}
3.2 Agent 角色定义
from dataclasses import dataclass
from typing import Optional
@dataclass
class AgentRole:
"""Agent 角色定义"""
name: str
role: str
expertise: list[str]
system_prompt: str
tools: list[str]
constraints: Optional[list[str]] = None
# 定义一个软件开发团队
team_roles = {
"pm": AgentRole(
name="产品经理",
role="需求分析与项目管理",
expertise=["需求分析", "用户故事", "优先级排序"],
system_prompt="""你是一个经验丰富的产品经理。
你的职责是理解用户需求，撰写清晰的用户故事和验收标准。
始终从用户体验角度思考问题。""",
tools=["文档编辑", "任务管理"],
constraints=["不做技术实现决策"]
),
"architect": AgentRole(
name="架构师",
role="系统设计与技术选型",
expertise=["系统设计", "架构模式", "技术选型"],
system_prompt="""你是一个资深软件架构师。
你负责设计系统架构，选择合适的技术栈，确保系统的可扩展性和可维护性。
你的决策需要考虑性能、安全、成本等多个维度。""",
tools=["代码分析", "架构图生成"],
constraints=["不直接写业务代码"]
),
"developer": AgentRole(
name="开发工程师",
role="代码实现",
expertise=["编码", "调试", "单元测试"],
system_prompt="""你是一个高效的开发工程师。
你负责根据架构设计实现功能代码，编写单元测试，确保代码质量。
遵循 SOLID 原则和项目编码规范。""",
tools=["代码编辑", "终端执行", "测试框架"],
constraints=["重大设计变更需架构师确认"]
),
"reviewer": AgentRole(
name="代码审查员",
role="代码质量把关",
expertise=["代码审查", "最佳实践", "安全审计"],
system_prompt="""你是一个严格的代码审查员。
你审查代码的正确性、可读性、性能和安全性。
给出具体的、可操作的改进建议。""",
tools=["代码分析", "静态检查"],
constraints=["不直接修改代码，只提建议"]
)
}
3.3 完整的多 Agent 系统实现
import json
from typing import Any
class MultiAgentSystem:
"""多 Agent 协作系统"""
def __init__(self):
self.agents = {}
self.message_bus = MessageBus()
self.state = SharedState()
def register_agent(self, name: str, agent):
"""注册 Agent"""
self.agents[name] = agent
self.message_bus.subscribe(name, agent.receive_message)
def execute(self, task: str) -> str:
"""执行任务"""
# 1. 任务分解
plan = self.agents["coordinator"].plan(task)
# 2. 分配并执行
results = {}
for step in plan["steps"]:
agent_name = step["agent"]
agent = self.agents[agent_name]
# 传递上下文
context = {
"task": task,
"step": step,
"previous_results": results,
"shared_state": self.state.get_all()
}
result = agent.execute(step["description"], context)
results[step["id"]] = result
# 更新共享状态
self.state.update(step["id"], result)
# 广播进展
self.message_bus.broadcast({
"type": "step_complete",
"step_id": step["id"],
"agent": agent_name,
"summary": result[:200] + "..." if len(result) > 200 else result
})
# 3. 整合结果
final = self.agents["coordinator"].integrate(task, results)
return final
class MessageBus:
"""Agent 间通信的消息总线"""
def __init__(self):
self.subscribers = {}
def subscribe(self, agent_name: str, callback):
self.subscribers[agent_name] = callback
def send(self, target: str, message: dict):
if target in self.subscribers:
self.subscribers[target](message)
def broadcast(self, message: dict):
for name, callback in self.subscribers.items():
callback(message)
class SharedState:
"""Agent 间共享状态"""
def __init__(self):
self._state = {}
def update(self, key: str, value: Any):
self._state[key] = value
def get(self, key: str, default=None):
return self._state.get(key, default)
def get_all(self) -> dict:
return dict(self._state)
4. 任务分解与规划策略
4.1 任务分解方法
方法一：递归分解
class RecursiveDecomposer:
"""递归任务分解器"""
def __init__(self, llm, max_depth=3, min_granularity="single_tool_call"):
self.llm = llm
self.max_depth = max_depth
self.min_granularity = min_granularity
def decompose(self, task: str, depth: int = 0) -> dict:
if depth >= self.max_depth:
return {"task": task, "type": "atomic", "children": []}
prompt = f"""分析以下任务，判断是否可以分解为更小的子任务。
任务：{task}
当前深度：{depth}
最大深度：{self.max_depth}
如果可以分解，请输出JSON格式：
{{"decomposable": true, "subtasks": ["子任务1", "子任务2", ...]}}
如果已经是原子任务（单个工具调用或简单推理），输出：
{{"decomposable": false}}"""
response = json.loads(self.llm.generate(prompt))
if not response.get("decomposable", False):
return {"task": task, "type": "atomic", "children": []}
children = []
for subtask in response["subtasks"]:
child = self.decompose(subtask, depth + 1)
children.append(child)
return {"task": task, "type": "composite", "children": children}
方法二：基于依赖图的分解
class DependencyGraphDecomposer:
"""基于依赖图的任务分解"""
def decompose(self, task: str) -> dict:
# LLM 分析任务依赖关系
prompt = f"""将以下任务分解为子任务，并分析它们之间的依赖关系。
任务：{task}
请输出JSON格式：
{{
"tasks": [
{{"id": "t1", "description": "子任务描述", "estimated_time": "短/中/长"}},
...
],
"dependencies": [
{{"from": "t1", "to": "t2", "type": "data|control"}},
...
]
}}"""
plan = json.loads(llm.generate(prompt))
# 拓扑排序确定执行顺序
execution_order = self.topological_sort(plan["tasks"], plan["dependencies"])
# 识别可并行执行的任务
parallel_groups = self.find_parallel_groups(plan["tasks"], plan["dependencies"])
return {
"plan": plan,
"execution_order": execution_order,
"parallel_groups": parallel_groups
}
def topological_sort(self, tasks, dependencies):
"""拓扑排序"""
from collections import deque
graph = {t["id"]: [] for t in tasks}
in_degree = {t["id"]: 0 for t in tasks}
for dep in dependencies:
graph[dep["from"]].append(dep["to"])
in_degree[dep["to"]] += 1
queue = deque([t for t, d in in_degree.items() if d == 0])
order = []
while queue:
node = queue.popleft()
order.append(node)
for neighbor in graph[node]:
in_degree[neighbor] -= 1
if in_degree[neighbor] == 0:
queue.append(neighbor)
return order
def find_parallel_groups(self, tasks, dependencies):
"""识别可并行执行的任务组"""
# 无依赖关系且无共同后续依赖的任务可以并行
groups = []
executed = set()
for task_id in self.topological_sort(tasks, dependencies):
deps = {d["from"] for d in dependencies if d["to"] == task_id}
if deps.issubset(executed):
# 可以立即执行
if groups and not any(
d["to"] in [t for t in groups[-1]]
for d in dependencies if d["from"] == task_id
):
groups[-1].append(task_id)
else:
groups.append([task_id])
executed.add(task_id)
return groups
4.2 规划策略
ReAct（Reasoning + Acting）
class ReActAgent:
"""ReAct 模式 Agent"""
def __init__(self, llm, tools: dict, max_steps=10):
self.llm = llm
self.tools = tools
self.max_steps = max_steps
def run(self, task: str) -> str:
prompt = f"""你是一个能够思考和行动的AI助手。
你可以使用以下工具：{list(self.tools.keys())}
请按以下格式回答：
Thought: 我需要思考...
Action: tool_name(args)
Observation: [工具返回结果]
... (重复 Thought/Action/Observation)
Thought: 我现在可以给出最终答案了
Final Answer: 最终答案
任务：{task}"""
for step in range(self.max_steps):
response = self.llm.generate(prompt)
# 解析响应
if "Final Answer:" in response:
return self.extract_final_answer(response)
if "Action:" in response:
tool_name, args = self.parse_action(response)
observation = self.execute_tool(tool_name, args)
prompt += f"\n{response}\nObservation: {observation}\n"
else:
prompt += f"\n{response}\n"
return "达到最大步骤数限制"
def execute_tool(self, tool_name: str, args: dict) -> str:
if tool_name in self.tools:
try:
return str(self.tools[tool_name](**args))
except Exception as e:
return f"工具执行错误: {str(e)}"
return f"未知工具: {tool_name}"
Plan-and-Execute
class PlanAndExecuteAgent:
"""先规划后执行的 Agent"""
def __init__(self, planner_llm, executor_llm, tools):
self.planner = planner_llm
self.executor = executor_llm
self.tools = tools
def run(self, task: str) -> str:
# 阶段1：规划
plan = self.create_plan(task)
print(f"计划：{len(plan['steps'])} 个步骤")
# 阶段2：逐步执行
results = []
for i, step in enumerate(plan["steps"]):
print(f"执行步骤 {i+1}: {step['description']}")
result = self.execute_step(step, results)
results.append({"step": step, "result": result})
# 检查是否需要重新规划
if self.should_replan(step, result, plan):
print("检测到异常，重新规划...")
plan = self.replan(task, results)
# 阶段3：总结
return self.summarize(task, results)
def should_replan(self, step, result, plan) -> bool:
"""判断是否需要重新规划"""
failure_indicators = ["错误", "失败", "无法", "error", "failed"]
return any(indicator in result.lower() for indicator in failure_indicators)
5. 工具调用与外部集成
5.1 工具定义框架
from abc import ABC, abstractmethod
from typing import Any, Callable
from pydantic import BaseModel, Field
class ToolParameter(BaseModel):
name: str
type: str
description: str
required: bool = True
default: Any = None
class ToolDefinition(BaseModel):
name: str
description: str
parameters: list[ToolParameter]
returns: str
examples: list[dict] = []
class BaseTool(ABC):
"""工具基类"""
@abstractmethod
def definition(self) -> ToolDefinition:
"""返回工具定义"""
pass
@abstractmethod
def execute(self, **kwargs) -> str:
"""执行工具"""
pass
def to_function_schema(self) -> dict:
"""转换为 OpenAI 函数调用格式"""
defn = self.definition()
properties = {}
required = []
for param in defn.parameters:
properties[param.name] = {
"type": param.type,
"description": param.description
}
if param.required:
required.append(param.name)
return {
"type": "function",
"function": {
"name": defn.name,
"description": defn.description,
"parameters": {
"type": "object",
"properties": properties,
"required": required
}
}
}
# 具体工具实现示例
class WebSearchTool(BaseTool):
"""网络搜索工具"""
def definition(self) -> ToolDefinition:
return ToolDefinition(
name="web_search",
description="搜索互联网获取最新信息",
parameters=[
ToolParameter(
name="query",
type="string",
description="搜索关键词"
),
ToolParameter(
name="num_results",
type="integer",
description="返回结果数量",
required=False,
default=5
)
],
returns="搜索结果列表，包含标题、URL和摘要"
)
def execute(self, query: str, num_results: int = 5, **kwargs) -> str:
# 实际搜索实现
import requests
# ... 搜索逻辑
return json.dumps(results, ensure_ascii=False)
class DatabaseQueryTool(BaseTool):
"""数据库查询工具"""
def __init__(self, connection_string: str):
self.conn = connection_string
def definition(self) -> ToolDefinition:
return ToolDefinition(
name="query_database",
description="执行SQL查询获取数据",
parameters=[
ToolParameter(
name="sql",
type="string",
description="SQL查询语句（只支持SELECT）"
),
ToolParameter(
name="database",
type="string",
description="数据库名称",
required=False,
default="default"
)
],
returns="查询结果的JSON格式"
)
def execute(self, sql: str, database: str = "default", **kwargs) -> str:
# 安全检查
if not sql.strip().upper().startswith("SELECT"):
return json.dumps({"error": "只允许SELECT查询"})
# 执行查询
# ... 数据库操作
return json.dumps(results, ensure_ascii=False, default=str)
5.2 工具调用编排
class ToolOrchestrator:
"""工具调用编排器"""
def __init__(self, tools: list[BaseTool]):
self.tools = {t.definition().name: t for t in tools}
self.schemas = [t.to_function_schema() for t in tools]
self.call_history = []
def plan_tool_calls(self, task: str, context: dict = None) -> list[dict]:
"""使用 LLM 规划工具调用序列"""
prompt = f"""根据任务需求，规划需要调用的工具序列。
任务：{task}
可用工具：{list(self.tools.keys())}
上下文：{json.dumps(context or {}, ensure_ascii=False)}
请输出工具调用计划（JSON数组）：
[
{{"tool": "tool_name", "args": {{}}, "purpose": "调用目的"}},
...
]
注意：
- 考虑工具之间的数据依赖关系
- 如果后续工具需要前面工具的输出，用 $step_N 引用
- 尽量减少不必要的工具调用"""
plan = json.loads(llm.generate(prompt))
return plan
def execute_plan(self, plan: list[dict]) -> dict:
"""执行工具调用计划"""
results = {}
for i, step in enumerate(plan):
tool_name = step["tool"]
args = step["args"]
# 解析引用（$step_N → 之前步骤的结果）
resolved_args = self.resolve_references(args, results)
# 执行工具
if tool_name in self.tools:
try:
result = self.tools[tool_name].execute(**resolved_args)
results[f"step_{i}"] = {
"status": "success",
"result": result,
"tool": tool_name
}
except Exception as e:
results[f"step_{i}"] = {
"status": "error",
"error": str(e),
"tool": tool_name
}
else:
results[f"step_{i}"] = {
"status": "error",
"error": f"未知工具: {tool_name}"
}
self.call_history.append({
"step": i,
"tool": tool_name,
"args": resolved_args,
"result": results[f"step_{i}"]
})
return results
def resolve_references(self, args: dict, results: dict) -> dict:
"""解析参数中的引用"""
resolved = {}
for key, value in args.items():
if isinstance(value, str) and value.startswith("$step_"):
step_key = value[1:] # 去掉 $
if step_key in results and results[step_key]["status"] == "success":
resolved[key] = results[step_key]["result"]
else:
resolved[key] = value # 保留原始引用
else:
resolved[key] = value
return resolved
5.3 常用工具集成
# 常用工具集合
class ToolKit:
"""预置工具集合"""
@staticmethod
def get_default_tools() -> list[BaseTool]:
return [
WebSearchTool(),
DatabaseQueryTool(connection_string="..."),
FileReadTool(),
FileWriteTool(),
CodeExecutionTool(),
EmailSenderTool(),
CalendarTool(),
APICallerTool(),
]
class CodeExecutionTool(BaseTool):
"""安全的代码执行工具"""
def definition(self) -> ToolDefinition:
return ToolDefinition(
name="execute_code",
description="在安全沙箱中执行Python代码",
parameters=[
ToolParameter(
name="code",
type="string",
description="要执行的Python代码"
),
ToolParameter(
name="timeout",
type="integer",
description="超时时间（秒）",
required=False,
default=30
)
],
returns="代码执行的输出结果"
)
def execute(self, code: str, timeout: int = 30, **kwargs) -> str:
import subprocess
import tempfile
# 安全检查
dangerous_patterns = [
"import os", "subprocess", "exec(", "eval(",
"open('/etc", "open('/proc", "__import__"
]
for pattern in dangerous_patterns:
if pattern in code:
return f"安全拒绝：代码包含危险操作 '{pattern}'"
# 在沙箱中执行
with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
f.write(code)
f.flush()
try:
result = subprocess.run(
["python3", f.name],
capture_output=True,
text=True,
timeout=timeout
)
return result.stdout + result.stderr
except subprocess.TimeoutExpired:
return f"执行超时（{timeout}秒）"
6. 状态管理与上下文传递
6.1 工作流状态设计
from datetime import datetime
from enum import Enum
from typing import Any, Optional
class WorkflowStatus(Enum):
PENDING = "pending"
RUNNING = "running"
PAUSED = "paused"
COMPLETED = "completed"
FAILED = "failed"
CANCELLED = "cancelled"
class WorkflowState:
"""工作流状态管理"""
def __init__(self, workflow_id: str):
self.workflow_id = workflow_id
self.status = WorkflowStatus.PENDING
self.created_at = datetime.now()
self.updated_at = datetime.now()
# 执行状态
self.current_step = 0
self.total_steps = 0
self.step_results = {}
self.errors = []
# 上下文数据
self.context = {}
self.variables = {}
self.checkpoints = []
def update_step(self, step_id: str, result: Any, status: str = "completed"):
"""更新步骤状态"""
self.step_results[step_id] = {
"result": result,
"status": status,
"timestamp": datetime.now().isoformat()
}
self.current_step += 1
self.updated_at = datetime.now()
def set_variable(self, key: str, value: Any):
"""设置工作流变量"""
self.variables[key] = value
self.updated_at = datetime.now()
def get_variable(self, key: str, default=None) -> Any:
"""获取工作流变量"""
return self.variables.get(key, default)
def create_checkpoint(self):
"""创建检查点（用于恢复）"""
checkpoint = {
"step": self.current_step,
"variables": dict(self.variables),
"step_results": dict(self.step_results),
"timestamp": datetime.now().isoformat()
}
self.checkpoints.append(checkpoint)
def restore_checkpoint(self, index: int = -1):
"""从检查点恢复"""
if not self.checkpoints:
raise ValueError("没有可用的检查点")
checkpoint = self.checkpoints[index]
self.current_step = checkpoint["step"]
self.variables = dict(checkpoint["variables"])
self.step_results = dict(checkpoint["step_results"])
self.status = WorkflowStatus.RUNNING
def to_dict(self) -> dict:
"""序列化为字典（用于持久化）"""
return {
"workflow_id": self.workflow_id,
"status": self.status.value,
"current_step": self.current_step,
"total_steps": self.total_steps,
"step_results": self.step_results,
"variables": self.variables,
"errors": self.errors,
"created_at": self.created_at.isoformat(),
"updated_at": self.updated_at.isoformat()
}
@classmethod
def from_dict(cls, data: dict) -> "WorkflowState":
"""从字典反序列化"""
state = cls(data["workflow_id"])
state.status = WorkflowStatus(data["status"])
state.current_step = data["current_step"]
state.total_steps = data["total_steps"]
state.step_results = data["step_results"]
state.variables = data["variables"]
state.errors = data["errors"]
return state
6.2 上下文窗口管理
class ContextManager:
"""上下文管理器 - 控制传递给 LLM 的上下文"""
def __init__(self, max_tokens: int = 4096):
self.max_tokens = max_tokens
self.messages = []
self.system_prompt = ""
self.summaries = [] # 历史摘要
def add_message(self, role: str, content: str):
"""添加消息"""
self.messages.append({
"role": role,
"content": content,
"timestamp": datetime.now().isoformat()
})
# 检查是否需要压缩
if self._estimate_tokens() > self.max_tokens * 0.8:
self._compress()
def get_messages(self, include_summary: bool = True) -> list[dict]:
"""获取当前上下文消息"""
messages = []
# 系统提示词
if self.system_prompt:
messages.append({"role": "system", "content": self.system_prompt})
# 历史摘要
if include_summary and self.summaries:
summary_text = "\n".join(self.summaries[-3:]) # 最近3条摘要
messages.append({
"role": "system",
"content": f"历史对话摘要：\n{summary_text}"
})
# 当前消息
messages.extend(self.messages)
return messages
def _compress(self):
"""压缩历史消息"""
if len(self.messages) < 6:
return
# 保留最近的消息
keep_recent = 4
old_messages = self.messages[:-keep_recent]
# 生成摘要
summary_prompt = "请用3-5句话总结以下对话的要点：\n"
for msg in old_messages:
summary_prompt += f"[{msg['role']}]: {msg['content'][:200]}\n"
summary = llm.generate(summary_prompt)
self.summaries.append(summary)
# 只保留最近的消息
self.messages = self.messages[-keep_recent:]
def _estimate_tokens(self) -> int:
"""粗略估算 token 数"""
total_chars = sum(len(m["content"]) for m in self.messages)
return total_chars // 2 # 粗略估算：2字符≈1token
6.3 持久化存储
import json
import sqlite3
from pathlib import Path
class WorkflowStore:
"""工作流状态持久化存储"""
def __init__(self, db_path: str = "workflows.db"):
self.db_path = db_path
self._init_db()
def _init_db(self):
conn = sqlite3.connect(self.db_path)
conn.execute("""
CREATE TABLE IF NOT EXISTS workflows (
workflow_id TEXT PRIMARY KEY,
state_json TEXT NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
conn.execute("""
CREATE TABLE IF NOT EXISTS workflow_logs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
workflow_id TEXT NOT NULL,
step_id TEXT,
event_type TEXT,
event_data TEXT,
timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (workflow_id) REFERENCES workflows(workflow_id)
)
""")
conn.commit()
conn.close()
def save_state(self, state: WorkflowState):
"""保存工作流状态"""
conn = sqlite3.connect(self.db_path)
conn.execute(
"""INSERT OR REPLACE INTO workflows (workflow_id, state_json, updated_at)
VALUES (?, ?, CURRENT_TIMESTAMP)""",
(state.workflow_id, json.dumps(state.to_dict(), ensure_ascii=False))
)
conn.commit()
conn.close()
def load_state(self, workflow_id: str) -> Optional[WorkflowState]:
"""加载工作流状态"""
conn = sqlite3.connect(self.db_path)
cursor = conn.execute(
"SELECT state_json FROM workflows WHERE workflow_id = ?",
(workflow_id,)
)
row = cursor.fetchone()
conn.close()
if row:
return WorkflowState.from_dict(json.loads(row[0]))
return None
def log_event(self, workflow_id: str, step_id: str, event_type: str, data: dict):
"""记录工作流事件"""
conn = sqlite3.connect(self.db_path)
conn.execute(
"""INSERT INTO workflow_logs (workflow_id, step_id, event_type, event_data)
VALUES (?, ?, ?, ?)""",
(workflow_id, step_id, event_type, json.dumps(data, ensure_ascii=False))
)
conn.commit()
conn.close()
7. 错误处理与容错机制
7.1 错误分类与处理策略
from enum import Enum
from typing import Callable
class ErrorSeverity(Enum):
LOW = "low" # 可忽略，继续执行
MEDIUM = "medium" # 需要重试
HIGH = "high" # 需要回退到上一个检查点
CRITICAL = "critical" # 需要终止工作流
class ErrorHandler:
"""统一错误处理器"""
def __init__(self):
self.strategies = {}
self.retry_config = {
"max_retries": 3,
"backoff_factor": 2,
"initial_delay": 1
}
def register_strategy(self, error_type: type, strategy: Callable):
"""注册错误处理策略"""
self.strategies[error_type] = strategy
def handle(self, error: Exception, context: dict) -> dict:
"""处理错误"""
error_type = type(error)
# 查找匹配的处理策略
if error_type in self.strategies:
return self.strategies[error_type](error, context)
# 默认处理策略
severity = self.classify_error(error)
if severity == ErrorSeverity.LOW:
return {"action": "continue", "message": str(error)}
elif severity == ErrorSeverity.MEDIUM:
return {"action": "retry", "message": str(error)}
elif severity == ErrorSeverity.HIGH:
return {"action": "rollback", "message": str(error)}
else:
return {"action": "abort", "message": str(error)}
def classify_error(self, error: Exception) -> ErrorSeverity:
"""错误严重性分类"""
if isinstance(error, (TimeoutError, ConnectionError)):
return ErrorSeverity.MEDIUM
elif isinstance(error, (ValueError, KeyError)):
return ErrorSeverity.LOW
elif isinstance(error, (PermissionError, MemoryError)):
return ErrorSeverity.CRITICAL
return ErrorSeverity.MEDIUM
class RetryManager:
"""重试管理器"""
def __init__(self, max_retries=3, backoff_factor=2):
self.max_retries = max_retries
self.backoff_factor = backoff_factor
def execute_with_retry(self, func: Callable, *args, **kwargs):
"""带重试的执行"""
import time
last_error = None
delay = 1
for attempt in range(self.max_retries + 1):
try:
return func(*args, **kwargs)
except Exception as e:
last_error = e
if attempt < self.max_retries:
print(f"尝试 {attempt + 1} 失败: {e}, {delay}秒后重试...")
time.sleep(delay)
delay *= self.backoff_factor
raise last_error
7.2 工作流级容错
class ResilientWorkflow:
"""具备容错能力的工作流"""
def __init__(self, store: WorkflowStore):
self.store = store
self.error_handler = ErrorHandler()
self.retry_manager = RetryManager()
def execute(self, state: WorkflowState, steps: list) -> WorkflowState:
"""容错执行工作流"""
state.status = WorkflowStatus.RUNNING
for i, step in enumerate(steps[state.current_step:], start=state.current_step):
state.current_step = i
# 创建检查点
if i % 5 == 0: # 每5步创建一次检查点
state.create_checkpoint()
self.store.save_state(state)
try:
# 带重试的步骤执行
result = self.retry_manager.execute_with_retry(
step.execute, state.context
)
state.update_step(step.id, result)
except Exception as e:
# 错误处理
action = self.error_handler.handle(e, {
"step": step,
"state": state,
"attempt": i
})
if action["action"] == "continue":
state.update_step(step.id, None, status="skipped")
continue
elif action["action"] == "retry":
# 已在重试管理器中处理
pass
elif action["action"] == "rollback":
state.restore_checkpoint()
i = state.current_step - 1 # 回退后重新开始
continue
else: # abort
state.status = WorkflowStatus.FAILED
state.errors.append({
"step": i,
"error": str(e),
"action": "abort"
})
self.store.save_state(state)
raise
state.status = WorkflowStatus.COMPLETED
self.store.save_state(state)
return state
7.3 LLM 调用容错
class LLMFallbackChain:
"""LLM 降级链 - 主模型失败时自动切换备选"""
def __init__(self):
self.models = []
def add_model(self, name: str, client, model: str, priority: int = 0):
"""添加模型到降级链"""
self.models.append({
"name": name,
"client": client,
"model": model,
"priority": priority
})
self.models.sort(key=lambda x: x["priority"], reverse=True)
def generate(self, prompt: str, **kwargs) -> str:
"""带降级的 LLM 调用"""
errors = []
for model_info in self.models:
try:
response = model_info["client"].chat.completions.create(
model=model_info["model"],
messages=[{"role": "user", "content": prompt}],
**kwargs
)
return response.choices[0].message.content
except Exception as e:
errors.append(f"{model_info['name']}: {str(e)}")
print(f"模型 {model_info['name']} 失败，尝试下一个...")
continue
raise RuntimeError(f"所有模型均失败:\n" + "\n".join(errors))
# 使用示例
llm_chain = LLMFallbackChain()
llm_chain.add_model("primary", openai_client, "gpt-4o", priority=10)
llm_chain.add_model("secondary", deepseek_client, "deepseek-chat", priority=5)
llm_chain.add_model("fallback", local_client, "qwen2.5-7b", priority=1)
8. 人机协作（Human-in-the-Loop）
8.1 HITL 设计模式
from abc import ABC, abstractmethod
from enum import Enum
from typing import Optional
class ApprovalStatus(Enum):
APPROVED = "approved"
REJECTED = "rejected"
MODIFIED = "modified"
PENDING = "pending"
class HumanCheckpoint:
"""人工检查点"""
def __init__(self, checkpoint_id: str, prompt: str, options: list[str] = None):
self.checkpoint_id = checkpoint_id
self.prompt = prompt
self.options = options or ["approve", "reject", "modify"]
self.status = ApprovalStatus.PENDING
self.feedback = None
self.modified_result = None
def to_dict(self) -> dict:
return {
"checkpoint_id": self.checkpoint_id,
"prompt": self.prompt,
"options": self.options,
"status": self.status.value,
"feedback": self.feedback
}
class HumanInTheLoopWorkflow:
"""支持人机协作的工作流"""
def __init__(self):
self.checkpoints = {}
self.pending_approvals = []
self.approval_callbacks = {}
def register_checkpoint(self, step_id: str, checkpoint: HumanCheckpoint,
on_approve: Callable = None, on_reject: Callable = None):
"""注册需要人工审批的检查点"""
self.checkpoints[step_id] = checkpoint
if on_approve:
self.approval_callbacks[f"{step_id}_approve"] = on_approve
if on_reject:
self.approval_callbacks[f"{step_id}_reject"] = on_reject
def request_approval(self, step_id: str, context: dict) -> ApprovalStatus:
"""请求人工审批"""
if step_id not in self.checkpoints:
return ApprovalStatus.APPROVED # 无检查点则自动通过
checkpoint = self.checkpoints[step_id]
# 展示给用户
print(f"\n{'='*60}")
print(f"⏳ 需要人工审批")
print(f"步骤: {step_id}")
print(f"说明: {checkpoint.prompt}")
print(f"上下文: {json.dumps(context, ensure_ascii=False, indent=2)[:500]}")
print(f"选项: {', '.join(checkpoint.options)}")
print(f"{'='*60}")
# 等待用户输入
user_input = input("请输入决策 (approve/reject/modify): ").strip().lower()
if user_input == "approve":
checkpoint.status = ApprovalStatus.APPROVED
if f"{step_id}_approve" in self.approval_callbacks:
self.approval_callbacks[f"{step_id}_approve"]()
elif user_input == "reject":
checkpoint.status = ApprovalStatus.REJECTED
checkpoint.feedback = input("请输入拒绝原因: ")
if f"{step_id}_reject" in self.approval_callbacks:
self.approval_callbacks[f"{step_id}_reject"]()
elif user_input == "modify":
checkpoint.status = ApprovalStatus.MODIFIED
checkpoint.modified_result = input("请输入修改后的内容: ")
return checkpoint.status
# 使用示例
hitl = HumanInTheLoopWorkflow()
# 在关键步骤注册审批点
hitl.register_checkpoint(
step_id="send_email",
checkpoint=HumanCheckpoint(
checkpoint_id="email_approval",
prompt="即将发送以下邮件，请确认",
options=["approve", "reject", "modify"]
),
on_approve=lambda: print("邮件已发送"),
on_reject=lambda: print("邮件已取消")
)
8.2 异步审批系统
import asyncio
from dataclasses import dataclass, field
from datetime import datetime
@dataclass
class ApprovalRequest:
request_id: str
workflow_id: str
step_id: str
description: str
data: dict
status: str = "pending"
created_at: datetime = field(default_factory=datetime.now)
resolved_at: Optional[datetime] = None
resolver: Optional[str] = None
decision: Optional[str] = None
comments: Optional[str] = None
class AsyncApprovalSystem:
"""异步审批系统"""
def __init__(self):
self.pending_requests: dict[str, ApprovalRequest] = {}
self.approval_events: dict[str, asyncio.Event] = {}
async def request_approval(self, request: ApprovalRequest) -> ApprovalRequest:
"""发起审批请求并等待"""
self.pending_requests[request.request_id] = request
self.approval_events[request.request_id] = asyncio.Event()
# 通知审批者（可通过邮件、Slack、Web UI等）
await self.notify_approvers(request)
# 等待审批结果
try:
await asyncio.wait_for(
self.approval_events[request.request_id].wait(),
timeout=3600 # 1小时超时
)
except asyncio.TimeoutError:
request.status = "timeout"
return request
def resolve_approval(self, request_id: str, decision: str,
resolver: str, comments: str = None):
"""审批者做出决定"""
if request_id in self.pending_requests:
request = self.pending_requests[request_id]
request.status = decision
request.decision = decision
request.resolver = resolver
request.comments = comments
request.resolved_at = datetime.now()
# 通知等待中的工作流
if request_id in self.approval_events:
self.approval_events[request_id].set()
async def notify_approvers(self, request: ApprovalRequest):
"""通知审批者"""
# 集成通知系统
notification = {
"title": f"需要审批: {request.description}",
"workflow_id": request.workflow_id,
"step": request.step_id,
"data_preview": str(request.data)[:200],
"action_url": f"/approval/{request.request_id}"
}
# 发送到 Slack/邮件/Webhook 等
print(f"审批通知已发送: {notification['title']}")
8.3 渐进式自动化
class ProgressiveAutomation:
"""渐进式自动化 - 随信任度提升逐步减少人工干预"""
def __init__(self):
self.trust_scores = {} # 步骤 → 信任度
self.success_history = {} # 步骤 → 成功记录
self.approval_threshold = 0.9 # 信任度阈值
def should_require_approval(self, step_id: str, context: dict) -> bool:
"""判断是否需要人工审批"""
trust = self.trust_scores.get(step_id, 0.5)
# 高风险操作始终需要审批
high_risk_operations = ["delete", "send_email", "deploy", "payment"]
if any(risk in step_id.lower() for risk in high_risk_operations):
return True
# 根据信任度决定
return trust < self.approval_threshold
def update_trust(self, step_id: str, success: bool):
"""更新信任度"""
if step_id not in self.success_history:
self.success_history[step_id] = []
self.success_history[step_id].append(success)
# 计算近期成功率
recent = self.success_history[step_id][-20:] # 最近20次
success_rate = sum(recent) / len(recent)
# 平滑更新信任度
current_trust = self.trust_scores.get(step_id, 0.5)
self.trust_scores[step_id] = current_trust * 0.7 + success_rate * 0.3
print(f"[{step_id}] 信任度更新: {current_trust:.2f} → {self.trust_scores[step_id]:.2f}")
9. 主流框架对比
9.1 LangGraph
LangGraph 是 LangChain 团队开发的图状态机框架，适合构建复杂的、有状态的 Agent 工作流。
# LangGraph 核心概念示例
from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated
import operator
class AgentState(TypedDict):
"""工作流状态定义"""
messages: Annotated[list, operator.add]
current_agent: str
task: str
results: dict
iteration: int
def researcher(state: AgentState) -> AgentState:
"""研究Agent"""
task = state["task"]
# 执行研究...
research_result = f"关于'{task}'的研究结果..."
return {
"messages": [f"研究员: {research_result}"],
"results": {"research": research_result}
}
def writer(state: AgentState) -> AgentState:
"""写作Agent"""
research = state["results"].get("research", "")
article = f"基于研究撰写的文章..."
return {
"messages": [f"作者: {article}"],
"results": {**state["results"], "article": article}
}
def reviewer(state: AgentState) -> AgentState:
"""审阅Agent"""
article = state["results"].get("article", "")
review = "审阅通过，质量良好"
return {
"messages": [f"审阅者: {review}"],
"results": {**state["results"], "review": review}
}
def should_continue(state: AgentState) -> str:
"""条件路由"""
if state["iteration"] >= 3:
return "end"
review = state["results"].get("review", "")
if "通过" in review:
return "end"
return "revise"
# 构建工作流图
graph = StateGraph(AgentState)
# 添加节点
graph.add_node("researcher", researcher)
graph.add_node("writer", writer)
graph.add_node("reviewer", reviewer)
# 添加边
graph.add_edge(START, "researcher")
graph.add_edge("researcher", "writer")
graph.add_edge("writer", "reviewer")
graph.add_conditional_edges(
"reviewer",
should_continue,
{
"end": END,
"revise": "writer" # 审阅不通过则重新写作
}
)
# 编译并运行
workflow = graph.compile()
result = workflow.invoke({
"messages": [],
"current_agent": "researcher",
"task": "写一篇关于AI Agent的技术博客",
"results": {},
"iteration": 0
})
LangGraph 特点：
- ✅ 图结构清晰，状态流转可视化
- ✅ 内置检查点和恢复机制
- ✅ 与 LangChain 生态深度集成
- ❌ 学习曲线较陡
- ❌ 对简单场景显得过于复杂
9.2 CrewAI
CrewAI 采用"团队协作"隐喻，让多个 Agent 扮演不同角色共同完成任务。
# CrewAI 核心概念示例
from crewai import Agent, Task, Crew, Process
# 定义 Agent（角色）
researcher = Agent(
role="资深研究分析师",
goal="收集和分析关于{topic}的最新、最准确的信息",
backstory="""你是一位经验丰富的研究分析师，
擅长从多个信息源获取数据并进行深入分析。
你的分析总是基于事实和数据。""",
verbose=True,
allow_delegation=False,
tools=[search_tool, browse_tool]
)
writer = Agent(
role="技术内容创作者",
goal="将复杂的技术概念转化为通俗易懂的优质内容",
backstory="""你是一位出色的技术写作者，
擅长将深奥的技术知识用生动的语言表达出来。
你的文章总是兼具深度和可读性。""",
verbose=True,
allow_delegation=False,
tools=[file_write_tool]
)
editor = Agent(
role="资深编辑",
goal="确保内容质量达到发布标准",
backstory="""你是一位严格但公正的编辑，
对内容质量有极高的要求。
你关注准确性、逻辑性和可读性。""",
verbose=True,
allow_delegation=True
)
# 定义任务
research_task = Task(
description="""研究{topic}的最新发展，包括：
1. 核心技术原理
2. 最新研究进展
3. 实际应用案例
4. 未来发展趋势""",
expected_output="详细的研究报告，包含数据来源",
agent=researcher
)
writing_task = Task(
description="""基于研究报告，撰写一篇技术博客文章：
- 标题吸引人
- 结构清晰（引言、正文、结论）
- 包含代码示例
- 适合中级开发者阅读""",
expected_output="2000-3000字的技术博客文章",
agent=writer
)
review_task = Task(
description="""审阅文章并提出修改建议：
- 事实准确性
- 逻辑连贯性
- 技术细节正确性
- 语言表达质量""",
expected_output="审阅报告和具体修改建议",
agent=editor
)
# 组建团队
crew = Crew(
agents=[researcher, writer, editor],
tasks=[research_task, writing_task, review_task],
process=Process.sequential, # 顺序执行
verbose=True
)
# 执行任务
result = crew.kickoff(inputs={"topic": "Agentic AI"})
print(result)
CrewAI 特点：
- ✅ 角色隐喻直观，易上手
- ✅ 内置任务委派和协作机制
- ✅ 适合模拟团队工作场景
- ❌ 自定义控制能力有限
- ❌ 复杂工作流表达能力不如 LangGraph
9.3 AutoGen
微软开源的多 Agent 对话框架，强调 Agent 间的自然对话协作。
# AutoGen 核心概念示例
from autogen import AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager
# 配置 LLM
llm_config = {
"model": "gpt-4o",
"temperature": 0.7,
"api_key": "your-api-key"
}
# 创建 Agent
planner = AssistantAgent(
name="Planner",
system_message="""你是一个项目规划专家。
你的职责是分析需求、制定计划、分配任务。
当你认为计划已经完善时，说 'PLAN_COMPLETE'。""",
llm_config=llm_config
)
coder = AssistantAgent(
name="Coder",
system_message="""你是一个高级Python开发者。
你根据计划编写高质量的代码。
你只写代码，不做其他事情。
代码块请用 ```python ... ``` 包裹。""",
llm_config=llm_config
)
reviewer = AssistantAgent(
name="Reviewer",
system_message="""你是一个代码审查专家。
你审查代码的质量、安全性和最佳实践。
如果代码符合要求，说 'CODE_APPROVED'。
否则，给出具体的修改建议。""",
llm_config=llm_config
)
# 用户代理（代表人类参与对话）
user_proxy = UserProxyAgent(
name="User",
human_input_mode="TERMINATE", # 最终结果时询问用户
max_consecutive_auto_reply=10,
code_execution_config={
"work_dir": "workspace",
"use_docker": True # 安全沙箱执行
}
)
# 创建群聊
group_chat = GroupChat(
agents=[user_proxy, planner, coder, reviewer],
messages=[],
max_round=20,
speaker_selection_method="auto" # 自动选择下一个发言者
)
manager = GroupChatManager(
groupchat=group_chat,
llm_config=llm_config
)
# 启动对话
user_proxy.initiate_chat(
manager,
message="请帮我开发一个命令行待办事项管理工具，支持增删改查和优先级排序。"
)
AutoGen 特点：
- ✅ 对话式协作，自然灵活
- ✅ 内置代码执行沙箱
- ✅ 支持人类随时介入对话
- ❌ 对话式架构难以精确控制流程
- ❌ 复杂工作流需要大量提示词工程
9.4 框架选型指南
需求场景 推荐框架 理由
──────────────────────────────────────────────────────────
复杂有状态工作流 LangGraph 图结构，状态管理强大
团队协作模拟 CrewAI 角色隐喻，开箱即用
探索性研究/头脑风暴 AutoGen 对话式，灵活自由
需要精确流程控制 LangGraph 显式状态机
快速原型/简单场景 CrewAI 上手最快
代码生成与执行 AutoGen 内置沙箱
企业级生产部署 LangGraph 检查点、监控最完善
10. 企业级工作流部署与监控
10.1 部署架构
# docker-compose.yml - 工作流服务部署
version: '3.8'
services:
# 工作流引擎
workflow-engine:
build: ./engine
ports:
- "8000:8000"
environment:
- DATABASE_URL=postgresql://user:pass@db:5432/workflows
- REDIS_URL=redis://redis:6379
- LLM_API_KEY=${LLM_API_KEY}
depends_on:
- db
- redis
deploy:
replicas: 3
resources:
limits:
memory: 4G
cpus: '2'
# 任务队列
task-worker:
build: ./worker
environment:
- BROKER_URL=redis://redis:6379
- DATABASE_URL=postgresql://user:pass@db:5432/workflows
depends_on:
- redis
- db
deploy:
replicas: 5
# 数据库
db:
image: postgres:16
volumes:
- pgdata:/var/lib/postgresql/data
environment:
POSTGRES_DB: workflows
POSTGRES_USER: user
POSTGRES_PASSWORD: pass
# 缓存与消息队列
redis:
image: redis:7-alpine
volumes:
- redisdata:/data
# 监控
prometheus:
image: prom/prometheus
ports:
- "9090:9090"
volumes:
- ./prometheus.yml:/etc/prometheus/prometheus.yml
grafana:
image: grafana/grafana
ports:
- "3000:3000"
depends_on:
- prometheus
volumes:
pgdata:
redisdata:
10.2 监控指标
from prometheus_client import Counter, Histogram, Gauge, start_http_server
import time
# 定义监控指标
workflow_total = Counter(
'workflow_total',
'工作流执行总数',
['workflow_type', 'status']
)
workflow_duration = Histogram(
'workflow_duration_seconds',
'工作流执行耗时',
['workflow_type'],
buckets=[1, 5, 10, 30, 60, 120, 300, 600]
)
active_workflows = Gauge(
'active_workflows',
'当前活跃工作流数量'
)
agent_calls = Counter(
'agent_calls_total',
'Agent 调用次数',
['agent_name', 'tool_name', 'status']
)
llm_tokens = Counter(
'llm_tokens_total',
'LLM Token 使用量',
['model', 'type'] # type: input/output
)
llm_latency = Histogram(
'llm_latency_seconds',
'LLM 调用延迟',
['model']
)
class MetricsCollector:
"""指标收集器"""
def __init__(self):
self.start_http_server()
def start_http_server(self):
"""启动指标暴露端口"""
start_http_server(9100)
def record_workflow_start(self, workflow_type: str):
active_workflows.inc()
def record_workflow_end(self, workflow_type: str, status: str, duration: float):
active_workflows.dec()
workflow_total.labels(workflow_type=workflow_type, status=status).inc()
workflow_duration.labels(workflow_type=workflow_type).observe(duration)
def record_agent_call(self, agent_name: str, tool_name: str,
status: str, duration: float):
agent_calls.labels(
agent_name=agent_name,
tool_name=tool_name,
status=status
).inc()
def record_llm_usage(self, model: str, input_tokens: int, output_tokens: int,
latency: float):
llm_tokens.labels(model=model, type="input").inc(input_tokens)
llm_tokens.labels(model=model, type="output").inc(output_tokens)
llm_latency.labels(model=model).observe(latency)
10.3 可观测性
import logging
import json
from datetime import datetime
from contextlib import contextmanager
class WorkflowLogger:
"""结构化工作流日志"""
def __init__(self, workflow_id: str):
self.workflow_id = workflow_id
self.logger = logging.getLogger(f"workflow.{workflow_id}")
self.logger.setLevel(logging.INFO)
# JSON 格式化器
handler = logging.StreamHandler()
handler.setFormatter(JsonFormatter())
self.logger.addHandler(handler)
@contextmanager
def trace_step(self, step_id: str, agent_name: str):
"""步骤执行追踪"""
start_time = time.time()
span = {
"workflow_id": self.workflow_id,
"step_id": step_id,
"agent": agent_name,
"start_time": datetime.now().isoformat()
}
self.logger.info(f"步骤开始: {step_id}", extra={"span": span, "event": "step_start"})
try:
yield span
span["status"] = "success"
except Exception as e:
span["status"] = "error"
span["error"] = str(e)
self.logger.error(f"步骤失败: {step_id} - {e}", extra={"span": span, "event": "step_error"})
raise
finally:
span["duration_ms"] = (time.time() - start_time) * 1000
self.logger.info(f"步骤完成: {step_id}", extra={"span": span, "event": "step_end"})
def log_decision(self, decision_point: str, options: list, chosen: str, reasoning: str):
"""记录决策点"""
self.logger.info("决策点", extra={
"event": "decision",
"workflow_id": self.workflow_id,
"decision_point": decision_point,
"options": options,
"chosen": chosen,
"reasoning": reasoning
})
class JsonFormatter(logging.Formatter):
"""JSON 日志格式化器"""
def format(self, record):
log_data = {
"timestamp": datetime.now().isoformat(),
"level": record.levelname,
"message": record.getMessage(),
"logger": record.name
}
if hasattr(record, 'span'):
log_data["span"] = record.span
if hasattr(record, 'event'):
log_data["event"] = record.event
return json.dumps(log_data, ensure_ascii=False)
10.4 成本控制
class CostController:
"""成本控制器"""
def __init__(self, budget_config: dict):
self.budget = budget_config
self.usage = {
"tokens": 0,
"api_calls": 0,
"compute_seconds": 0
}
def check_budget(self, estimated_cost: float) -> bool:
"""检查是否超出预算"""
daily_budget = self.budget.get("daily_limit_usd", 100)
current_spend = self.get_current_spend()
if current_spend + estimated_cost > daily_budget:
return False
return True
def get_optimization_suggestions(self) -> list[str]:
"""获取成本优化建议"""
suggestions = []
# 分析 token 使用
if self.usage["tokens"] > 1000000:
suggestions.append(
"Token 使用量过高，考虑：\n"
"1. 使用更短的提示词模板\n"
"2. 限制输出长度（max_tokens）\n"
"3. 使用更便宜的模型处理简单任务"
)
# 分析模型选择
suggestions.append(
"成本优化建议：\n"
"- 简单分类/提取任务 → 使用 GPT-4o-mini\n"
"- 代码生成 → 使用 DeepSeek Coder\n"
"- 复杂推理 → 使用 GPT-4o 或 Claude"
)
return suggestions
def estimate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
"""估算调用成本"""
pricing = {
"gpt-4o": {"input": 2.5, "output": 10.0}, # $/1M tokens
"gpt-4o-mini": {"input": 0.15, "output": 0.6},
"deepseek-chat": {"input": 0.14, "output": 0.28},
"claude-3-5-sonnet": {"input": 3.0, "output": 15.0},
}
if model not in pricing:
return 0.0
rate = pricing[model]
cost = (input_tokens * rate["input"] + output_tokens * rate["output"]) / 1_000_000
return cost
10.5 生产部署清单
## 生产环境部署清单
### 安全性
- [ ] API Key 使用环境变量或密钥管理服务，不硬编码
- [ ] 用户输入进行严格的提示词注入防护
- [ ] 工具调用设置权限白名单
- [ ] 敏感数据脱敏后再传给 LLM
- [ ] 代码执行在沙箱环境中进行
### 可靠性
- [ ] LLM 调用设置超时和重试机制
- [ ] 关键步骤设置检查点，支持断点恢复
- [ ] 异常情况有降级方案（如使用更小的模型）
- [ ] 工作流状态持久化到数据库
- [ ] 实现幂等性（重复执行不产生副作用）
### 性能
- [ ] 非关键路径使用异步执行
- [ ] LLM 响应流式输出，降低感知延迟
- [ ] 相似请求结果缓存
- [ ] 合理设置上下文窗口大小
- [ ] 监控并优化 token 使用量
### 可观测性
- [ ] 结构化日志，包含 workflow_id、step_id
- [ ] 关键指标暴露给 Prometheus
- [ ] Grafana 面板：成功率、延迟、成本
- [ ] 异常告警（失败率 > 5%、延迟 > 阈值）
- [ ] 定期生成成本报告
### 运维
- [ ] 工作流版本管理，支持回滚
- [ ] 灰度发布新版本工作流
- [ ] 文档化的故障排查手册
- [ ] 定期备份工作流定义和执行历史
总结
Agentic Workflow 是构建下一代 AI 应用的核心架构模式。本教程覆盖了从基础概念到企业级部署的完整知识体系：
- 设计模式：掌握顺序、并行、条件、循环、层级五种基本模式，能够组合应对各种复杂场景
- 多 Agent 协作：理解辩论式、审查式、委员会式等协作模式，合理设计 Agent 角色分工
- 任务规划：运用 ReAct、Plan-and-Execute 等策略，实现智能任务分解与动态调整
- 工具集成：构建标准化的工具接口，实现安全可控的外部系统集成
- 状态管理：设计健壮的状态持久化和上下文管理机制
- 容错机制：实现重试、降级、检查点恢复等多层次容错
- 人机协作：在关键节点引入人工审批，实现渐进式自动化
- 框架选型：根据场景选择 LangGraph、CrewAI 或 AutoGen
- 企业部署：构建完整的监控、日志、成本控制体系
从简单的工作流开始，逐步迭代优化，最终构建出可靠、高效、可观测的智能工作流系统。
下一步学习建议：
- 使用 LangGraph 构建一个简单的 ReAct Agent，体验状态流转
- 用 CrewAI 搭建一个 3 人协作团队，完成一个实际任务
- 为你的工作流添加人工审批节点和错误处理逻辑
- 搭建 Prometheus + Grafana 监控面板，观察工作流运行指标
本教程最后更新：2025年6月
