【学习笔记】AI Agent 设计模式入门教程

"请阅读这份市场报告，分析趋势，然后给营销团队写一封邮件。"

步骤1："请总结以下市场报告的主要发现：[报告内容]"
步骤2："基于以上摘要，识别前三个新兴趋势并提取支持数据：[步骤1输出]"
步骤3："向营销团队起草一封邮件，概述以下趋势及其数据：[步骤2输出]"

步骤1：从URL或文档中提取文本内容
步骤2：清理和总结文本
步骤3：提取特定实体（姓名、日期、位置）
步骤4：搜索内部知识库获取相关信息
步骤5：生成包含摘要、实体和搜索结果的最终报告

步骤1：根据用户兴趣生成5个主题想法
步骤2：用户选择后，生成详细大纲
步骤3：逐节撰写内容（每节都基于前文的上下文）
步骤4：审查和完善完整草稿的连贯性、语气和语法

步骤1：处理用户话语，识别意图和关键实体
步骤2：更新对话状态
步骤3：基于当前状态生成响应
→ 循环执行，累积对话历史

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# 初始化模型
llm = ChatOpenAI(temperature=0)

# 步骤1：提取信息
prompt_extract = ChatPromptTemplate.from_template(
    "从以下文本中提取技术规格：\n\n{text_input}"
)

# 步骤2：转换为 JSON
prompt_transform = ChatPromptTemplate.from_template(
    "将以下规格转换为 JSON 对象，使用 'cpu'、'memory' 和 'storage' 作为键：\n\n{specifications}"
)

# 构建链：提取 → LLM → 字符串解析
extraction_chain = prompt_extract | llm | StrOutputParser()

# 完整链：将提取结果传入转换提示词
full_chain = (
    {"specifications": extraction_chain}
    | prompt_transform
    | llm
    | StrOutputParser()
)

# 执行
result = full_chain.invoke({"text_input": "这台服务器配备了 Intel i7 处理器，32GB 内存和 1TB SSD 存储。"})
print(result)

"分析以下用户查询，仅输出类别：'订单状态'、'产品信息'、'技术支持' 或 '其他'。"

用户输入："我昨天买的手机什么时候到？"
LLM 输出："订单状态"

用户查询 → 向量化 → 与各类别向量比较 → 选择最相似的类别

if "订单" in query or "快递" in query:
    route_to = "order_system"
elif "价格" in query or "功能" in query:
    route_to = "product_catalog"
elif "怎么" in query or "故障" in query:
    route_to = "tech_support"
else:
    route_to = "human_agent"

标注数据 → 训练分类器 → 实时预测路由

用户："我的订单到哪了？"
    → 意图识别：订单查询
    → 路由到：订单查询 Agent
    → 调用订单数据库 API
    → 返回："您的订单 #12345 预计明天送达"

用户："这款手机支持5G吗？"
    → 意图识别：产品咨询
    → 路由到：产品信息 Agent
    → 查询产品数据库
    → 返回："是的，支持5G网络..."

用户："手机开不了机怎么办？"
    → 意图识别：技术支持
    → 路由到：故障排除 Agent
    → 返回故障排查步骤

科学问题 → 科学专家 Agent
历史问题 → 历史专家 Agent
编程问题 → 代码专家 Agent
医学问题 → 医学专家 Agent（需额外安全审核）

串行方式：
  步骤1：搜索新闻（2秒）
  步骤2：查股票数据（2秒）
  步骤3：看社交媒体（2秒）
  步骤4：查公司数据库（2秒）
  总耗时：8秒

并行方式：
  同时执行：搜索新闻 + 查股票 + 看社交媒体 + 查数据库（各2秒）
  合并结果（1秒）
  总耗时：3秒

# 并行任务
任务1：搜索新闻文章
任务2：获取股票数据
任务3：监控社交媒体提及
任务4：查询公司数据库

# 所有任务完成后
合并：综合分析生成报告

# 并行任务
任务1：查询航班价格
任务2：查询酒店可用性
任务3：查询当地活动
任务4：查询餐厅推荐

# 所有任务完成后
合并：生成完整旅行方案

# 并行任务
任务1：情感分析
任务2：关键词提取
任务3：反馈分类
任务4：紧急问题识别

# 所有任务完成后
合并：生成多维度分析报告

from langchain_core.runnables import RunnableParallel, RunnablePassthrough
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)

# 定义并行的多个处理链
chain1 = ChatPromptTemplate.from_template("总结以下文本的主要观点：{input}") | llm
chain2 = ChatPromptTemplate.from_template("提取以下文本的关键词：{input}") | llm
chain3 = ChatPromptTemplate.from_template("分析以下文本的情感倾向：{input}") | llm

# 并行执行
parallel_chain = RunnableParallel(
    summary=chain1,
    keywords=chain2,
    sentiment=chain3
)

# 输入相同的内容，三个链同时处理
result = parallel_chain.invoke({"input": "这家餐厅的食物非常美味，服务也很周到！"})
print(result["summary"])   # 摘要
print(result["keywords"])  # 关键词
print(result["sentiment"]) # 情感分析

┌─────────┐    ┌─────────┐    ┌─────────┐    ┌─────────┐
│  执行   │ → │  评估   │ → │  反思   │ → │  优化   │
│ 生成初稿│    │ 检查质量│    │ 确定改进│    │ 生成改进稿│
└─────────┘    └─────────┘    └─────────┘    └─────────┘
      ↑                                              │
      └────────────── 循环迭代 ←─────────────────────┘

# 第一轮：生成草稿
draft = llm.invoke("写一篇关于人工智能的文章")

# 第二轮：自我审查
review = llm.invoke(f"请检查以下文章的问题（语法、逻辑、事实错误）：\n{draft}")

# 第三轮：基于审查修改
final = llm.invoke(f"根据以下审查意见修改文章：\n原文：{draft}\n审查意见：{review}")

# 执行者 Agent：专注于生成内容
writer = Agent(
    role="技术作家",
    goal="撰写高质量的技术文章"
)

# 评审者 Agent：专注于发现问题
reviewer = Agent(
    role="资深编辑",
    goal="检查文章的事实准确性、逻辑性和可读性"
)

# 工作流程
draft = writer.write("AI Agent 设计模式")
feedback = reviewer.review(draft)
final = writer.revise(draft, feedback)

步骤1：编写初始代码
步骤2：运行测试或静态分析
步骤3：识别错误或低效之处
步骤4：基于发现修改代码
步骤5：重复直到通过所有测试

步骤1：生成草稿
步骤2：评审流畅性、语气和清晰度
步骤3：基于评审重写
步骤4：重复直到满足质量标准

步骤1：提出一个解题步骤
步骤2：评估是否更接近解决方案
步骤3：必要时回溯或选择不同步骤
步骤4：重复直到问题解决

用户提问
    ↓
LLM 分析：是否需要工具？
    ↓
是 → 生成工具调用请求（JSON格式）
    ↓
框架执行工具调用
    ↓
获取工具返回结果
    ↓
LLM 基于结果生成最终回答
    ↓
返回给用户

用户："伦敦的天气如何？"

LLM 思考：用户问天气，我需要调用天气工具。

工具调用：
{
  "tool": "weather_api",
  "parameters": {
    "location": "伦敦"
  }
}

工具返回：
{
  "temperature": 15,
  "condition": "多云",
  "humidity": 72
}

LLM 生成回答："伦敦当前多云，温度15°C，湿度72%。"

用户："AAPL 的当前价格是多少？如果我以150美元买入100股，计算潜在收益。"

步骤1：调用股票 API 获取 AAPL 当前价格（$178.15）
步骤2：调用计算器工具计算收益：(178.15 - 150) × 100 = $2,815
步骤3：格式化输出完整分析报告

用户："关闭客厅的灯，把空调调到26度。"

步骤1：调用智能家居 API 关闭客厅灯
步骤2：调用智能家居 API 设置空调温度
步骤3：确认执行结果

from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate

# 定义工具
@tool
def get_weather(location: str) -> str:
    """获取指定位置的当前天气。"""
    # 实际实现会调用天气 API
    weather_data = {
        "北京": "晴朗，25°C",
        "上海": "小雨，22°C",
        "伦敦": "多云，15°C"
    }
    return weather_data.get(location, "未知位置")

@tool
def calculator(expression: str) -> str:
    """计算数学表达式。"""
    try:
        result = eval(expression)
        return str(result)
    except:
        return "计算错误"

# 创建 Agent
llm = ChatOpenAI(model="gpt-4o")
tools = [get_weather, calculator]

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个有用的助手，可以使用工具来帮助用户。"),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

agent = create_tool_calling_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools)

# 执行
result = agent_executor.invoke({"input": "北京天气怎么样？然后计算 15 * 23 + 8"})
print(result["output"])

用户："研究司美格鲁肽对全球医疗保健系统的经济影响"

Agent 规划：
1. 搜索司美格鲁肽的基本信息和用途
2. 查找全球医疗保健支出数据
3. 分析司美格鲁肽在不同国家的定价
4. 研究其对糖尿病治疗成本的影响
5. 评估对医疗系统预算的长期影响
6. 综合所有信息生成结构化报告

执行过程中：
- 发现某国数据缺失 → 调整计划，寻找替代数据源
- 发现新的相关政策 → 增加分析维度

用户："帮我开发一个待办事项应用"

Agent 规划：
1. 需求分析：确定功能列表（添加、删除、标记完成、筛选）
2. 技术选型：选择技术栈（React + Node.js + MongoDB）
3. 数据库设计：设计数据模型
4. 后端开发：实现 API 接口
5. 前端开发：实现用户界面
6. 集成测试：验证功能完整性
7. 部署：发布到服务器

研究员 Agent → 作家 Agent → 编辑 Agent → 发布 Agent

          ┌→ 市场分析 Agent ─┐
用户请求 → ┼→ 技术分析 Agent ─┼→ 汇总 Agent → 最终报告
          └→ 竞品分析 Agent ─┘

正方 Agent："应该投资A项目"
反方 Agent："不应该投资A项目"
评判 Agent："综合双方观点，结论是..."

        项目经理 Agent
       /      |      \
开发Agent  测试Agent  文档Agent

医疗诊断团队：
- 症状分析 Agent
- 检查报告解读 Agent
- 诊断建议 Agent
- 治疗方案 Agent

写作者 Agent 生成文章
    ↓
审查者 Agent 检查（事实、安全、合规）
    ↓
写作者 Agent 根据反馈修改
    ↓
循环直到通过审查

# 定义角色
requirement_analyst = Agent(role="需求分析师", goal="理解并澄清用户需求")
coder = Agent(role="代码工程师", goal="编写高质量代码")
tester = Agent(role="测试工程师", goal="发现并报告问题")
document_writer = Agent(role="技术文档工程师", goal="编写使用文档")

# 协作流程
requirements = requirement_analyst.analyze("用户想要一个在线商城")
code = coder.implement(requirements)
test_results = tester.test(code)
if test_results.has_issues:
    code = coder.fix(code, test_results.issues)
documentation = document_writer.write(code)

market_researcher = Agent(role="市场研究员", goal="分析目标受众")
copywriter = Agent(role="文案撰写", goal="创作吸引人的文案")
designer = Agent(role="设计师", goal="生成视觉素材")
social_media_manager = Agent(role="社媒运营", goal="制定发布计划")

# 并行启动研究
audience = market_researcher.research("新产品发布")

# 基于研究结果并行创作
copy = copywriter.write(audience)
design = designer.create(audience)
schedule = social_media_manager.plan(audience)

# 汇总为完整营销方案
campaign = merge(copy, design, schedule)

# 短期记忆：对话历史
conversation_history = [
    {"role": "user", "content": "你好"},
    {"role": "assistant", "content": "你好！有什么可以帮助你的？"},
    {"role": "user", "content": "我想订披萨"},
    # ...
]

# 长期记忆：用户画像（存储在数据库）
user_profile = {
    "user_id": "12345",
    "preferences": {
        "food": ["川菜", "意大利菜"],
        "language": "中文",
        "notification": True
    },
    "history_facts": {
        "last_order": "2024-01-15 麻辣香锅",
        "favorite_restaurant": "川味轩"
    }
}

# 工作记忆：任务执行状态
task_state = {
    "current_step": 3,
    "intermediate_results": {
        "step1": "完成用户认证",
        "step2": "获取用户位置"
    },
    "pending_actions": ["搜索附近餐厅", "展示推荐列表"]
}

用户提问
    ↓
查询向量化 → 在向量数据库中搜索相似内容
    ↓
获取相关知识片段（Top-K）
    ↓
将知识片段 + 用户问题 组合成提示词
    ↓
LLM 基于检索到的知识生成回答
    ↓
返回答案（可附带引用来源）

用户："公司的年假政策是什么？"

RAG 流程：
1. 将问题转换为向量
2. 在员工手册向量库中检索相关内容
3. 获取 "年假政策" 章节的内容
4. LLM 基于检索内容回答："根据员工手册第3章，入职满1年可享受10天年假..."

用户："怎么重置密码？"

RAG 流程：
1. 检索帮助文档中关于"密码重置"的内容
2. 获取操作步骤
3. 生成清晰的指导回答

┌─────────────┐         MCP 协议          ┌─────────────┐
│   Agent     │ ◄──────────────────────► │    Tool     │
│  (Client)   │   标准化请求/响应格式      │  (Server)   │
└─────────────┘                          └─────────────┘

Agent 工作流程
    ↓
计划审批环节 ──→ 人类确认计划是否可行
    ↓
工具使用授权 ──→ 人类确认是否允许调用敏感工具
    ↓
歧义消解节点 ──→ 人类澄清模糊的需求
    ↓
最终输出审核 ──→ 人类审核最终成果
    ↓
交付给用户

┌─────────────────────────────────────────┐
│           输入层防护                     │
│  - 检测恶意输入（注入攻击、提示词攻击）    │
│  - 过滤敏感信息（身份证号、密码）          │
└─────────────────────────────────────────┘
                    ↓
┌─────────────────────────────────────────┐
│           处理层防护                     │
│  - 限制工具调用权限（最小权限原则）        │
│  - 监控异常行为模式                       │
│  - 设置资源使用上限（防止无限循环）        │
└─────────────────────────────────────────┘
                    ↓
┌─────────────────────────────────────────┐
│           输出层防护                     │
│  - 内容安全过滤（暴力、仇恨、色情）        │
│  - 事实准确性检查                        │
│  - 敏感信息脱敏                          │
└─────────────────────────────────────────┘

任务：预订从北京到上海的航班
评估：
- ✅ 是否成功预订？
- ✅ 航班时间是否正确？
- ✅ 价格是否在预算内？
- ✅ 乘客信息是否准确？

评估：
- 是否选择了合适的工具？
- 是否遵循了正确的步骤？
- 是否在合理时间内完成？
- 是否有不必要的重复操作？

评估：
- 回答是否有帮助？（1-5分）
- 回答是否准确？（1-5分）
- 交互是否自然？（1-5分）
- 用户是否愿意再次使用？

优点：
- 最成熟的生态，文档丰富
- 大量预置组件和集成
- 社区活跃，问题容易找到答案

缺点：
- 版本迭代快，有时有破坏性变更
- 抽象层次多，学习曲线较陡

适合：大多数 Agent 项目，特别是需要快速原型开发

优点：
- 基于图的工作流定义，灵活强大
- 内置状态管理
- 支持循环和条件分支

缺点：
- 比 LangChain 更复杂
- 需要理解图的概念

适合：复杂的多步骤工作流、需要状态持久化的场景

优点：
- 与 Gemini 模型深度集成
- 强大的多智能体编排能力
- Google 官方支持

缺点：
- 与 Google 生态绑定较深
- 社区相对较小

适合：使用 Google Cloud、Gemini 的项目

优点：
- 角色扮演概念直观易懂
- 多智能体协作非常简单
- 代码简洁，上手快

缺点：
- 相对较新，生态不如 LangChain
- 复杂定制可能受限

适合：多智能体协作项目、快速原型

初学者入门 → CrewAI（简单直观）
通用项目 → LangChain（生态完善）
复杂工作流 → LangGraph（图结构强大）
Google 生态 → Google ADK（深度集成）
编程辅助 → AutoGen（Microsoft 出品）
企业应用 → Semantic Kernel（企业级特性）

🚀 基础模式
   ├── 提示词链 → 分解复杂任务
   ├── 路由     → 分发到不同处理路径
   └── 并行化   → 同时执行独立任务

🧠 进阶模式
   ├── 反思     → 自我评估和改进
   ├── 工具使用 → 连接外部世界
   └── 规划     → 动态分解目标

🤝 高级模式
   ├── 多智能体协作 → 团队化工作
   ├── 记忆管理     → 记住上下文
   └── 知识检索     → 访问外部知识

🛡️ 实践模式
   ├── 人机协同     → 人类监督
   ├── 安全防护     → 行为约束
   └── 评估监控     → 质量保证

🔗 协议标准
   ├── MCP → Agent 与工具的标准
   └── A2A → Agent 与 Agent 的标准