【赚钱必读】别只聊天!这 5 个 AI Agent 模式能让你副业效率翻倍
单纯的 Prompt 已经不够了。掌握从'计划执行'到'多智能体协作'的 5 种核心模式,让 AI 真正成为你的赚钱合伙人。
你将获得什么
- 5 种 AI Agent 设计模式的完整实现方案
- 每种模式的 Python 代码示例(基于 OpenAI SDK)
- 真实副业场景应用案例,拿来即用
- 模式选型对比表,帮你快速决策
如果你还在把 ChatGPT 当搜索引擎用,那你只发挥了 AI 10% 的能力。真正的生产力飞跃,来自让 AI 像员工一样自主工作——这就是 AI Agent。
本文将带你掌握 5 种经过验证的 Agent 设计模式,每种都配有可运行的代码和副业落地方案。
前置条件
- 基本的 Python 知识(能读懂函数和类即可)
- OpenAI 或 DeepSeek API Key(代码示例基于 OpenAI SDK,DeepSeek 兼容同一接口)
- 已安装
openai、chromadb等依赖库 - 预计阅读时间:25 分钟
pip install openai chromadb requests
模式一:Plan-and-Execute(计划执行模式)
核心原理
Plan-and-Execute 是最基础也最强大的 Agent 模式。它的核心思想是先规划,再执行:
- Planner(规划器):接收用户的高层指令,将其拆解为有序的子任务列表
- Executor(执行器):按顺序逐一执行每个子任务,将上一步的输出作为下一步的输入
- Replanner(重规划器):在执行过程中根据实际结果动态调整计划
这种模式模拟了人类处理复杂任务的方式——你不会一口气写完一篇文章,而是先列提纲、再逐段撰写、最后修改润色。
实现代码
import json
from openai import OpenAI
client = OpenAI() # 或使用 DeepSeek: OpenAI(api_key="your-key", base_url="https://api.deepseek.com")
def create_plan(user_task: str) -> list[str]:
"""让 LLM 将复杂任务拆解为子任务列表"""
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{
"role": "system",
"content": (
"你是一个任务规划专家。将用户的任务拆解为 3-7 个具体的、可执行的子步骤。"
"以 JSON 数组格式返回,每个元素是一个步骤描述字符串。"
)
},
{"role": "user", "content": f"请拆解这个任务:{user_task}"}
],
response_format={"type": "json_object"}
)
result = json.loads(response.choices[0].message.content)
return result.get("steps", [])
def execute_step(step: str, context: str) -> str:
"""执行单个子任务,传入之前步骤的上下文"""
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{
"role": "system",
"content": "你是一个执行专家。根据上下文完成指定任务,输出详细结果。"
},
{
"role": "user",
"content": f"当前上下文:\n{context}\n\n请执行这个步骤:{step}"
}
]
)
return response.choices[0].message.content
def plan_and_execute(user_task: str) -> str:
"""Plan-and-Execute 主流程"""
# 阶段一:规划
print(f"📋 正在规划任务:{user_task}")
steps = create_plan(user_task)
print(f"📝 生成了 {len(steps)} 个子任务")
# 阶段二:逐步执行
context = f"原始任务:{user_task}\n"
for i, step in enumerate(steps):
print(f"⚙️ 执行步骤 {i+1}/{len(steps)}:{step}")
result = execute_step(step, context)
context += f"\n--- 步骤 {i+1} 结果 ---\n{result}\n"
# 阶段三:汇总输出
final = execute_step("根据以上所有步骤的结果,生成最终的完整输出", context)
return final
# 使用示例
output = plan_and_execute("写一篇关于 Python 异步编程的 SEO 博客文章")
print(output)
副业应用场景
SEO 内容自动生成流水线:
一条指令触发完整的内容生产链路:
- 自动调研关键词和竞品文章
- 生成 SEO 优化的标题和大纲
- 逐章节撰写初稿
- 插入内链、优化 meta description
- 格式化为 Markdown 并准备发布
✅ 预期效果:一条指令”写一篇关于XXX的文章”,Agent 自动完成从调研到发布的全部步骤。原本需要 3-4 小时的工作,缩短到 10 分钟。
模式二:Self-Reflection(自我反思模式)
核心原理
Self-Reflection 模式让 Agent 具备自我审查和迭代改进的能力。流程如下:
- Generator(生成器):产出初始内容
- Critic(评审器):按照预设标准评审内容,给出具体改进建议
- Refiner(改进器):根据评审意见修改内容
- 重复步骤 2-3,直到达到质量标准或达到最大迭代次数
关键设计要点是:评审器需要有明确的评分标准,否则容易陷入无意义的循环修改。
实现代码
from openai import OpenAI
client = OpenAI()
def generate_draft(task: str) -> str:
"""生成初始内容"""
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": "你是一个专业的文案撰写人。"},
{"role": "user", "content": task}
]
)
return response.choices[0].message.content
def critique(content: str, criteria: list[str]) -> dict:
"""评审内容,返回评分和改进建议"""
criteria_text = "\n".join(f"- {c}" for c in criteria)
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{
"role": "system",
"content": (
"你是一个严格的内容审核专家。根据给定标准评审内容。"
"以 JSON 格式返回:{\"score\": 1-10, \"passed\": true/false, "
"\"issues\": [\"问题1\", \"问题2\"], \"suggestions\": [\"建议1\", \"建议2\"]}"
)
},
{
"role": "user",
"content": f"评审标准:\n{criteria_text}\n\n待评审内容:\n{content}"
}
],
response_format={"type": "json_object"}
)
import json
return json.loads(response.choices[0].message.content)
def refine(content: str, suggestions: list[str]) -> str:
"""根据建议改进内容"""
suggestions_text = "\n".join(f"- {s}" for s in suggestions)
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{
"role": "system",
"content": "你是一个文案优化专家。根据改进建议修改内容,保持原有风格。"
},
{
"role": "user",
"content": f"改进建议:\n{suggestions_text}\n\n原始内容:\n{content}"
}
]
)
return response.choices[0].message.content
def self_reflection_loop(
task: str,
criteria: list[str],
max_iterations: int = 3,
pass_score: int = 8
) -> str:
"""Self-Reflection 主循环"""
# 生成初稿
content = generate_draft(task)
print(f"📝 初稿已生成({len(content)} 字)")
for i in range(max_iterations):
# 评审
review = critique(content, criteria)
score = review.get("score", 0)
print(f"🔍 第 {i+1} 轮评审:得分 {score}/10")
if review.get("passed") or score >= pass_score:
print(f"✅ 内容通过评审!最终得分:{score}/10")
return content
# 改进
print(f"🔄 发现 {len(review.get('issues', []))} 个问题,正在改进...")
content = refine(content, review.get("suggestions", []))
print(f"⚠️ 达到最大迭代次数 {max_iterations},返回当前最佳版本")
return content
# 使用示例
result = self_reflection_loop(
task="写一段 200 字的产品描述:智能台灯,支持语音控制和自动调光",
criteria=[
"文案要有吸引力,避免平铺直叙",
"包含至少 3 个产品卖点",
"适合电商详情页使用",
"字数在 150-250 字之间",
"包含行动号召(CTA)"
]
)
print(result)
副业应用场景
电商文案批量优化:
从”能用”到”好用”的质量飞跃:
- 初稿生成 → AI 按转化率标准评审 → 自动修改 → 再评审 → 终稿输出
- 批量处理商品详情页文案,每条自动迭代 2-3 轮
- 评审标准可自定义:SEO 友好度、情感调性、卖点覆盖率等
✅ 预期效果:文案质量相比单次生成提升 40%+,特别是在专业度和说服力方面有明显改善。
模式三:Tool Use(工具调用模式)
核心原理
Tool Use 是让 Agent 突破纯文本限制的关键模式。通过 Function Calling,Agent 可以:
- 感知环境:调用搜索引擎获取实时信息、读取数据库
- 执行操作:发送邮件、操作文件系统、调用第三方 API
- 计算处理:运行代码、执行数学计算、处理数据
工作流程:LLM 分析用户意图 → 决定调用哪个工具 → 构造参数 → 执行工具 → 将结果返回给 LLM → 生成最终回答。
核心优势在于 Agent 能自主决定何时使用什么工具,而非依赖硬编码的 if-else 逻辑。
实现代码
import json
import requests
from openai import OpenAI
client = OpenAI()
# 定义可用工具
tools = [
{
"type": "function",
"function": {
"name": "web_search",
"description": "搜索互联网获取最新信息",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "搜索关键词"
}
},
"required": ["query"]
}
}
},
{
"type": "function",
"function": {
"name": "calculate",
"description": "执行数学计算",
"parameters": {
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "数学表达式,例如 '100 * 1.15'"
}
},
"required": ["expression"]
}
}
},
{
"type": "function",
"function": {
"name": "send_notification",
"description": "发送通知消息",
"parameters": {
"type": "object",
"properties": {
"channel": {
"type": "string",
"enum": ["email", "webhook", "sms"],
"description": "通知渠道"
},
"message": {
"type": "string",
"description": "通知内容"
}
},
"required": ["channel", "message"]
}
}
}
]
# 工具执行函数
def execute_tool(name: str, args: dict) -> str:
if name == "web_search":
# 实际项目中可接入 SerpAPI、Tavily 等
return f"搜索结果:找到关于 '{args['query']}' 的 10 条相关结果..."
elif name == "calculate":
try:
result = eval(args["expression"]) # 生产环境应使用安全的表达式解析器
return f"计算结果:{result}"
except Exception as e:
return f"计算错误:{str(e)}"
elif name == "send_notification":
# 实际项目中调用邮件/Webhook API
return f"通知已发送至 {args['channel']}:{args['message']}"
return "未知工具"
def tool_use_agent(user_message: str, max_rounds: int = 5) -> str:
"""Tool Use Agent 主循环"""
messages = [
{
"role": "system",
"content": (
"你是一个强大的助手,可以使用工具完成任务。"
"分析用户需求,选择合适的工具,必要时多次调用不同工具。"
)
},
{"role": "user", "content": user_message}
]
for _ in range(max_rounds):
response = client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=tools,
tool_choice="auto"
)
msg = response.choices[0].message
# 如果没有工具调用,说明 Agent 已完成任务
if not msg.tool_calls:
return msg.content
# 执行所有工具调用
messages.append(msg)
for tool_call in msg.tool_calls:
func_name = tool_call.function.name
func_args = json.loads(tool_call.function.arguments)
print(f"🔧 调用工具:{func_name}({func_args})")
result = execute_tool(func_name, func_args)
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": result
})
return "达到最大调用轮次,任务可能未完全完成。"
# 使用示例
answer = tool_use_agent("帮我查一下 2026 年最火的 AI 框架,然后计算如果我每月投入 2000 元学习,一年总投入是多少")
print(answer)
副业应用场景
竞品监控 Agent:
自动化竞品情报收集与分析:
- 定时搜索竞品最新动态(产品更新、价格变化、营销活动)
- 调用数据分析工具对比关键指标
- 自动生成竞品分析报告
- 通过邮件或 Webhook 推送通知
✅ 预期效果:自动完成需要 3-4 个工具协同的复杂任务,将每天 2 小时的手动监控工作变为全自动。
模式四:Multi-Agent Collaboration(多智能体协作)
核心原理
Multi-Agent 模式是最接近真实团队协作的设计。核心要素:
- 专业化分工:每个 Agent 有独立的 System Prompt 和工具集,专注于一个领域
- 路由机制:一个 Router Agent 负责分析请求,将任务分配给合适的专业 Agent
- Handoff(交接):Agent 之间通过结构化消息传递上下文,实现无缝切换
- 共享记忆:所有 Agent 可以访问共同的对话历史和知识库
关键设计原则:每个 Agent 的职责边界要清晰,避免职责重叠导致的冲突。
实现代码
import json
from openai import OpenAI
client = OpenAI()
class Agent:
"""基础 Agent 类"""
def __init__(self, name: str, system_prompt: str, model: str = "gpt-4o"):
self.name = name
self.system_prompt = system_prompt
self.model = model
def respond(self, messages: list[dict]) -> str:
full_messages = [
{"role": "system", "content": self.system_prompt}
] + messages
response = client.chat.completions.create(
model=self.model,
messages=full_messages
)
return response.choices[0].message.content
# 创建专业 Agent 团队
router_agent = Agent(
name="路由 Agent",
system_prompt=(
"你是一个客服路由专家。分析用户消息,判断应该转给哪个专业 Agent。"
"返回 JSON:{\"route_to\": \"tech_support\" | \"sales\" | \"billing\", "
"\"reason\": \"转接原因\", \"summary\": \"用户问题摘要\"}"
)
)
tech_agent = Agent(
name="技术支持 Agent",
system_prompt=(
"你是一个资深技术支持工程师。专门处理产品使用问题、Bug 报告和技术咨询。"
"回答要专业但易懂,必要时提供分步操作指南。"
)
)
sales_agent = Agent(
name="销售 Agent",
system_prompt=(
"你是一个经验丰富的销售顾问。专门处理产品咨询、价格方案和购买引导。"
"回答要热情专业,适时推荐合适的方案,但不要过度推销。"
)
)
billing_agent = Agent(
name="账单 Agent",
system_prompt=(
"你是一个账务处理专家。专门处理退款、发票、付款和账户余额相关问题。"
"回答要准确严谨,涉及金额时要再三确认。"
)
)
# Agent 路由映射
agent_map = {
"tech_support": tech_agent,
"sales": sales_agent,
"billing": billing_agent,
}
def multi_agent_chat(user_message: str) -> str:
"""多智能体协作主流程"""
messages = [{"role": "user", "content": user_message}]
# 第一步:路由决策
route_response = router_agent.respond(messages)
print(f"🔀 路由决策:{route_response}")
try:
route_data = json.loads(route_response)
target = route_data.get("route_to", "tech_support")
summary = route_data.get("summary", user_message)
except json.JSONDecodeError:
target = "tech_support"
summary = user_message
# 第二步:转接到专业 Agent
agent = agent_map.get(target, tech_agent)
print(f"📞 转接至:{agent.name}")
# 带上路由摘要,为专业 Agent 提供上下文
context_messages = [
{
"role": "user",
"content": f"[路由摘要:{summary}]\n\n用户原始消息:{user_message}"
}
]
response = agent.respond(context_messages)
return f"[{agent.name}]\n{response}"
# 使用示例
print(multi_agent_chat("我购买的专业版套餐,想咨询一下怎么部署到服务器上"))
print("---")
print(multi_agent_chat("我上个月多扣了 200 块钱,请帮我查一下"))
副业应用场景
自动化客服系统:
构建一个能处理各类客户问题的智能客服:
- 路由 Agent:分析用户意图,自动分发到对应专业 Agent
- 技术 Agent:处理产品使用、故障排查
- 销售 Agent:处理购买咨询、方案推荐
- 账务 Agent:处理退款、发票等财务问题
每个 Agent 都有自己的专业知识库和对话风格,协作处理复杂的跨领域问题。
✅ 预期效果:处理复杂客户请求时自动在不同专业 Agent 间切换,客户满意度提升,人工客服工作量减少 60%+。
模式五:RAG + Agent(知识增强模式)
核心原理
RAG(Retrieval-Augmented Generation)解决了 LLM 的两大痛点:
- 知识过时:LLM 训练数据有截止日期,无法回答最新问题
- 幻觉问题:LLM 会”一本正经地胡说八道”
RAG + Agent 的工作流程:
- 文档处理:将私有文档切分为小块(Chunks),生成向量嵌入(Embeddings)
- 向量存储:将嵌入存入向量数据库(如 ChromaDB、Pinecone)
- 检索增强:用户提问时,先从向量库中检索最相关的文档片段
- 增强生成:将检索到的内容注入 Prompt,LLM 基于真实文档生成回答
实现代码
import os
import json
import chromadb
from openai import OpenAI
client = OpenAI()
chroma_client = chromadb.Client()
def get_embedding(text: str) -> list[float]:
"""获取文本的向量嵌入"""
response = client.embeddings.create(
model="text-embedding-3-small",
input=text
)
return response.data[0].embedding
class KnowledgeBase:
"""基于 ChromaDB 的知识库"""
def __init__(self, collection_name: str):
self.collection = chroma_client.get_or_create_collection(
name=collection_name,
metadata={"hnsw:space": "cosine"}
)
def add_documents(self, documents: list[dict]):
"""批量添加文档到知识库
documents: [{"id": "doc1", "content": "...", "metadata": {...}}, ...]
"""
ids = [doc["id"] for doc in documents]
contents = [doc["content"] for doc in documents]
metadatas = [doc.get("metadata", {}) for doc in documents]
embeddings = [get_embedding(content) for content in contents]
self.collection.add(
ids=ids,
documents=contents,
embeddings=embeddings,
metadatas=metadatas
)
print(f"✅ 已添加 {len(documents)} 个文档到知识库")
def search(self, query: str, top_k: int = 3) -> list[dict]:
"""检索最相关的文档片段"""
query_embedding = get_embedding(query)
results = self.collection.query(
query_embeddings=[query_embedding],
n_results=top_k
)
retrieved = []
for i in range(len(results["ids"][0])):
retrieved.append({
"id": results["ids"][0][i],
"content": results["documents"][0][i],
"distance": results["distances"][0][i] if results.get("distances") else None
})
return retrieved
def rag_agent(query: str, kb: KnowledgeBase) -> str:
"""RAG Agent 主流程"""
# 第一步:检索相关文档
print(f"🔍 正在检索:{query}")
docs = kb.search(query, top_k=3)
# 构建上下文
context_parts = []
for i, doc in enumerate(docs):
context_parts.append(f"[文档 {i+1}]\n{doc['content']}")
context = "\n\n".join(context_parts)
print(f"📚 找到 {len(docs)} 个相关文档片段")
# 第二步:增强生成
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{
"role": "system",
"content": (
"你是一个专业的知识库问答助手。根据提供的参考文档回答用户问题。"
"要求:\n"
"1. 只基于参考文档中的信息回答,不要编造\n"
"2. 如果文档中没有相关信息,明确告知用户\n"
"3. 引用具体的文档来源\n"
"4. 回答要简洁准确"
)
},
{
"role": "user",
"content": f"参考文档:\n{context}\n\n用户问题:{query}"
}
]
)
return response.choices[0].message.content
# 使用示例:构建产品知识库
kb = KnowledgeBase("product_docs")
# 添加文档(实际项目中从文件批量导入)
kb.add_documents([
{
"id": "pricing_basic",
"content": "基础版套餐:每月 99 元,包含 5 个项目,10GB 存储空间,邮件支持。",
"metadata": {"category": "pricing"}
},
{
"id": "pricing_pro",
"content": "专业版套餐:每月 299 元,包含无限项目,100GB 存储空间,优先技术支持,API 访问权限。",
"metadata": {"category": "pricing"}
},
{
"id": "api_auth",
"content": "API 认证方式:在请求头中添加 Authorization: Bearer YOUR_API_KEY。API Key 可在控制台-设置-API管理中生成。",
"metadata": {"category": "api"}
},
{
"id": "deploy_guide",
"content": "部署指南:支持 Docker 部署和 Kubernetes 部署。Docker 命令:docker run -d -p 8080:8080 myapp:latest。需要先配置环境变量 DATABASE_URL 和 REDIS_URL。",
"metadata": {"category": "deployment"}
}
])
# 提问
answer = rag_agent("专业版多少钱?有什么功能?", kb)
print(answer)
副业应用场景
私域知识库问答机器人:
为你的课程、社群或产品构建专属问答系统:
- 将你的教程文档、FAQ、产品手册导入知识库
- 用户提问时自动检索最相关的内容并生成精准回答
- 可接入微信公众号、网站聊天窗口等渠道
- 随时添加新文档,知识库持续更新
✅ 预期效果:基于你的私有文档回答问题,准确率 90%+,大幅减少重复解答的人力成本。
模式对比与选型指南
| 模式 | 实现难度 | 适用场景 | 推荐工具/框架 | API 调用成本 |
|---|---|---|---|---|
| Plan-and-Execute | ⭐⭐ 中等 | 多步骤复杂任务、内容生产 | LangGraph, CrewAI | 中(多次调用) |
| Self-Reflection | ⭐ 简单 | 质量敏感型输出、文案优化 | 原生 OpenAI SDK | 低(2-4 轮迭代) |
| Tool Use | ⭐⭐ 中等 | 需要外部数据/操作的任务 | OpenAI Function Calling | 中 |
| Multi-Agent | ⭐⭐⭐ 较高 | 复杂业务流程、客服系统 | AutoGen, CrewAI | 高(多 Agent 并行) |
| RAG + Agent | ⭐⭐ 中等 | 私有知识问答、文档检索 | LlamaIndex, LangChain | 中(含 Embedding) |
选型建议:
- 刚入门:从 Self-Reflection 开始,最容易实现,效果立竿见影
- 内容创业者:Plan-and-Execute + Self-Reflection 组合,自动产出高质量内容
- 技术接单:Tool Use + Multi-Agent,可以构建完整的自动化解决方案
- 知识付费:RAG + Agent 是标配,让你的知识库活起来
常见问题
Q1:这些模式可以组合使用吗?
当然可以,而且推荐组合使用。例如:Plan-and-Execute 的每个步骤中可以使用 Tool Use 调用外部工具;Multi-Agent 中的每个 Agent 都可以内置 Self-Reflection 来保证输出质量;RAG 可以作为任何 Agent 的底层能力来增强知识。
Q2:用 OpenAI 还是 DeepSeek?
代码示例使用 OpenAI SDK,但 DeepSeek 完全兼容同一套接口,只需修改 base_url 和 api_key:
client = OpenAI(
api_key="your-deepseek-key",
base_url="https://api.deepseek.com"
)
DeepSeek 的优势是价格更低(约 OpenAI 的 1/10),在中文场景下表现也很优秀。建议开发阶段用 DeepSeek 降低成本,上线后根据效果选择。
Q3:API 调用成本怎么控制?
几个实用技巧:
- 缓存策略:相同的查询结果做本地缓存,避免重复调用
- 模型分级:简单任务用
gpt-4o-mini或 DeepSeek,复杂任务才用gpt-4o - 控制迭代次数:Self-Reflection 设置合理的
max_iterations(通常 2-3 轮足够) - 批量处理:攒够一批再调用,利用 Batch API 获得 50% 折扣
Q4:如何处理 Agent 的错误和异常?
生产环境中务必加入:
- 重试机制:API 调用失败时自动重试(建议指数退避)
- 超时控制:设置合理的超时时间,避免无限等待
- Fallback 策略:主 Agent 失败时切换到备用方案
- 日志记录:记录每次调用的输入输出,便于调试
Q5:这些模式需要什么样的硬件?
不需要本地 GPU。所有模式都是调用云端 API,一台普通笔记本电脑就能运行。唯一需要注意的是 RAG 模式中如果文档量很大(超过 10 万条),向量数据库可能需要更多内存,建议使用云端向量数据库(如 Pinecone)。
下一步扩展
掌握了这 5 种基础模式之后,你可以继续探索:
- Agent 可观测性:接入 LangSmith 或 Langfuse,可视化追踪每一步调用,快速定位问题
- Guardrails(安全护栏):为 Agent 添加输入输出过滤,防止生成不当内容或执行危险操作
- Human-in-the-Loop:在关键决策点引入人工审核,兼顾效率和可控性
- Memory(长期记忆):为 Agent 添加持久化记忆,让它记住用户偏好和历史对话
- 评估体系:建立 Agent 性能评估框架,用真实数据衡量效果
每种模式都不是终点,而是你构建 AI 自动化系统的积木。把它们灵活组合,你就能打造出真正能替你赚钱的 AI 合伙人。
现在就选一个模式,动手实现你的第一个 Agent 吧。