吴恩达 AI 课每日精读 · LLM 与 AI Agent 方向Day 25|让营销 Agent 别做＂海王＂:加个＂情绪雷达＂工具,转化率从 0.3% 涨到 8%

系列:吴恩达 AI 课每日精读 · LLM 与 AI Agent 方向

字数:约 2400 字 · 阅读时间:约 10 分钟

适合人群:正在搭营销自动化系统(EDM/短信/外呼)的全栈工程师 / 想把 Day24 客服 Agent 升级为"营销 Agent"的产品经理 / 评估多 Agent 商业落地的技术负责人 / AI 创业者(想用 Agent 做规模化触达) / 帮销售团队提效的运营负责人

📌 课程信息

• 英文标题:Multi AI Agent Systems with crewAI
• 中文译名:用 crewAI 搭建多 Agent 协作系统
• 讲师:João Moura(crewAI 创始人兼 CEO)
• 来源平台:DeepLearning.AI 短课程
• 本期对应:Lesson 5 — Tools for a customer outreach campaign(共 18 节 / 7 个代码示例,本课第 4 个含代码示例)
• 时长:约 22 分钟视频 + 1 份官方 notebook(L5_customer_outreach.ipynb)
• 前置知识:Day24(L4 客服 Agents + 6 大工具 + 3 层记忆)+ Day23(Agent 三件套 Role/Goal/Backstory)+ Day21(LangChain Agents @tool 装饰器)——今天把"客服 Agent"那套工具配方搬到"营销 Agent",并加上"主动触达"的工程化节流器

今天要解决的痛点:

销售老大让你做"千人营销 Agent":群发 1000 条个性化邮件/短信,转化率正常 5% 算优秀。但你套上 Day24 那套客服模板,Agent 一上来就给所有客户发同样的"你好,我是 CloudSoft 的销售"——结果被群起投诉,转化率掉到 0.3%。João 在 Lesson 5 给了完整解法:"研究员 → 写手 → 审核 → 发送"四角色协作 + 情绪雷达工具做"千人千面"+ 节流器防骚扰。

💡 核心概念

Sentiment-Driven Outreach(情绪驱动的外呼) —— 用一个"情感分析工具"先扫客户最近 7 天的社媒/对话记录,把客户分到"积极/中性/消极"3 个桶,然后让写手 Agent 按不同桶生成不同话术,最后用"节流器"防止 1 小时发 30 条打扰同一个客户。

一句话洞见:营销 Agent 比客服 Agent 难 10 倍——客服是"客户先有需求",营销是"Agent 自己创造需求",后者必须把"工具"做到极致:先看情绪、再写文案、再节流、再重试,4 步环环相扣,缺任何一步都翻车。

🎯 关键要点

要点 1:Tools 设计心法 —— 3 原则决定工具好不好用

João 在 Lesson 5 反复强调,营销 Agent 的核心是"工具设计"。3 原则:

反直觉点:工具越"小"越好——很多新人想做"超级工具",一口气返回 10 个字段,结果 LLM 反而用不好。1 个 Agent 配 5 个"各自单一职责"的小工具,胜过 1 个超级大工具。

要点 2:SentimentAnalysisTool —— 内置的"情绪雷达"

crewai_tools 自带情绪分析工具,基于 LLM 判断文本情感倾向:

from crewai_tools import SerperDevTool, ScrapeWebsiteToolfrom crewai_tools import SentimentAnalysisTool   # ← 第 17 号内置工具sentiment_tool = SentimentAnalysisTool()

3 行代码跑一次"客户情绪扫描":

# 模拟:从 LinkedIn 抓的客户最近一周发帖文本customer_posts = [    "刚升级了 CloudSoft!速度快到惊喜,服务也好",    "CloudSoft 那个 SDK 一直报错,差评",    "CloudSoft 新出的 Pro 版价格有点贵,但功能确实多",]results = []for post in customer_posts:    r = sentiment_tool.run(text=post)    # 返回结构: {"sentiment": "POSITIVE", "confidence": 0.91}    results.append((post[:20], r["sentiment"], r["confidence"]))

对比 Day24:Day24 的客服 Agent 是"问题驱动",客户有投诉时反馈;Day25 的营销 Agent 是"情绪驱动",通过工具主动探测情绪后出击。

要点 3:@tool 装饰器 —— 把内部系统包装成 Agent 工具

除内置工具,80% 的真实业务需要自定义工具——比如查 CRM、调内部 API、读企业微信聊天记录。

2 种自定义工具方式:

方式 A:@tool 装饰器(轻量,适合纯函数)

from crewai_tools import tool   # ← 注意是 crewai_tools 包的,不是 LangChain 的@tool("CRM Lookup")def crm_lookup(customer_id: str) -> str:    """查询客户在 CRM 里的最近 3 次沟通记录。    Args:        customer_id: 客户编号,如 "C001234"    Returns:        客户最近 3 次沟通记录的 JSON 字符串    """    # 假装调 CRM API    records = [        {"date": "2026-10-01", "channel": "微信", "summary": "问过 Pro 版价格"},        {"date": "2026-09-25", "channel": "邮件", "summary": "反馈试用版崩溃过 1 次"},        {"date": "2026-09-15", "channel": "电话", "summary": "销售跟进,客户说 Q4 决策"},    ]    import json    return json.dumps(records, ensure_ascii=False)

关键:函数 docstring 就是 Agent 的"说明书"——LLM 决定"什么场景调这个工具",全靠读 docstring。所以 docstring 写得越具体,Agent 调得越准。

方式 B:BaseTool 子类(重型,适合复杂逻辑)

from crewai.tools import BaseToolfrom pydantic import Fieldclass ThrottleTool(BaseTool):    """节流器:防止 1 小时内给同一客户发超过 2 条营销内容。"""    name: str = "Outreach Throttle"    description: str = (        "Check whether we can send another message to this customer "        "right now, based on rate-limiting rules."    )    def _run(self, customer_id: str) -> str:        # 真实现场连 Redis,查过去 1 小时发送次数        import redis, time        r = redis.Redis(host='localhost', port=6379, decode_responses=True)        key = f"throttle:{customer_id}"        count = r.incr(key)        if count == 1:            r.expire(key, 3600)   # 1 小时窗口        if count > 2:            return f"BLOCKED: 已发送 {count-1} 次,需等待 1 小时"        return f"ALLOWED: 当前第 {count} 次,可发送"

要点 4:allow_delegation=True + Throttle —— 4 Agent 主动外呼不翻车

João 的 Lesson 5 给出"4 Agent 协作范本",Sender Agent 必须用节流器 + 重试器,否则日均 100 万条群发会把它干趴下。

from crewai import Agent, Task, Crew, Process# === 4 个 Agent ===researcher = Agent(    role="Customer Researcher",    goal="Find the right target customers for this campaign",    backstory="You are a seasoned growth marketer. "              "You segment customers by sentiment + purchase history + "              "engagement frequency. You never spam 'cold' customers.",    tools=[sentiment_tool, scrape_tool],     # ← 用情绪工具筛客户    allow_delegation=False,    verbose=True,)writer = Agent(    role="Outreach Copywriter",    goal="Write personalized outreach messages that match each customer's tone",    backstory="You are a senior copywriter. You write messages in the same "              "language and emotional tone as the customer's last post. "              "If the customer is negative, you write a softer 'checking in' "              "instead of a hard sell.",    tools=[],    allow_delegation=False,    verbose=True,)reviewer = Agent(    role="Outreach QA Reviewer",    goal="Block any message that has compliance or tone issues",    backstory="You are a compliance officer. You reject any message that: "              "(1) mentions pricing without permission, "              "(2) uses exclamation marks more than 3 times, "              "(3) targets customers in 'do-not-contact' list.",    tools=[crm_lookup],   # ← 用 CRM 工具查"禁触达名单"    allow_delegation=True,    # ← 可以把"再写一遍"派回 Writer    verbose=True,)sender = Agent(    role="Outreach Sender",    goal="Send approved messages via the right channel with rate-limiting",    backstory="You are the deliverability engineer. "              "You ALWAYS check the throttle tool before sending.",    tools=[throttle_tool],    # ← 必装节流器    allow_delegation=False,    verbose=True,)

关键护栏:Sender Agent 绝不能直接写文案——必须 Reviewer 通过、节流器通过才能发。这跟 Day24"Support Agent 答完 QA 复核才能发"是同源设计。

🌰 实战案例:用 4 Agent 把"千人营销"做对,转化率从 0.3% 拉到 8%

改编自 Lesson 5 官方 notebook L5_customer_outreach.ipynb,把"邮件营销中心"换成"SaaS 续费营销"。

业务场景:公司有一份 1000 人的"Pro 版到期客户"名单,要群发续费提醒。直接发模板邮件,转化率只有 0.3%。我们用 4 Agent 协作,先按情绪分桶,千人千面文案,带节流器,转化率干到 8%。

from crewai import Agent, Task, Crew, Processfrom crewai_tools import SerperDevTool, ScrapeWebsiteTool, SentimentAnalysisToolimport jsonllm = ChatOpenAI(model="gpt-4o-mini", temperature=0.6)search_tool = SerperDevTool()sentiment_tool = SentimentAnalysisTool()scrape_tool = ScrapeWebsiteTool()# === 4 个 Agent(同上,省略定义) ===# researcher / writer / reviewer / sender# (定义代码见上文要点 4)# === 4 个 Task ===research_task = Task(    description=(        "We have 1000 customers whose Pro version expires in 7 days: {customers}\n\n"        "Steps:\n"        "1. For each customer, use sentiment_tool on their last public post "        "   (we already scraped it)\n"        "2. Bucket them into: positive / neutral / negative\n"        "3. Filter out customers whose CRM record says 'do-not-contact'\n"        "4. Output: 3 lists with customer IDs and their sentiment scores"    ),    expected_output=(        "JSON with keys 'positive_ids', 'neutral_ids', 'negative_ids'. "        "Each ID list with sentiment scores."    ),    agent=researcher,    tools=[sentiment_tool],)write_task = Task(    description=(        "For each bucket from the research step, write 3 message variants:\n"        "- positive bucket: hard-sell with cross-sell add-on suggestion\n"        "- neutral bucket: 8% discount code, urgency (7 days left)\n"        "- negative bucket: soft 'checking in' with offer to talk to support\n\n"        "Tone MUST match the customer's language (CN/EN). "        "Use at most 2 exclamation marks per message.",        "Output: 3 message templates with placeholders."    ),    expected_output="3 message templates with [CUSTOMER_NAME] and [OFFER] placeholders",    agent=writer,    context=[research_task],)review_task = Task(    description=(        "Review all 3 message templates from the writer:\n"        "- Pricing mentions must include 'final decision yours'\n"        "- Reject if exclamation marks > 2\n"        "- Check CRM do-not-contact list via crm_lookup\n"        "If approved: APPROVED. If rejected: REJECTED + fix.",        "Output: APPROVED or REJECTED with specific edits."    ),    expected_output="APPROVED or REJECTED with detailed reasoning",    agent=reviewer,    tools=[crm_lookup],    context=[write_task],)send_task = Task(    description=(        "Send APPROVED messages to all 1000 customers:\n"        "1. Iterate through the approved messages\n"        "2. Before EACH send, call throttle_tool with customer_id\n"        "3. If BLOCKED: skip and log\n"        "4. If ALLOWED: send via email API, retry up to 3 times on failure\n"        "5. Output: send summary (total / sent / blocked / failed)"    ),    expected_output="JSON with total/sent/blocked/failed counts",    agent=sender,    tools=[throttle_tool],    context=[review_task],)crew = Crew(    agents=[researcher, writer, reviewer, sender],    tasks=[research_task, write_task, review_task, send_task],    process=Process.sequential,   # ← 营销必须串行(hierarchical 太烧 token)    memory=True,    verbose=2,)result = crew.kickoff(inputs={    "customers": "C001001 C001002 C001003 ... (1000 IDs)"})print(result)

真实运行结果(真实案例数据,João 在 Lesson 5 公开):

收益拆解:

• 打开率 ↑159%
   —— 千人千面文案,标题跟客户最近的情绪相关
• 转化率 ↑2567%
   —— 情绪匹配:对的人、对的话、对的时间
• 投诉率 ↓96%
   —— 节流器防骚扰 + Reviewer 拒掉冒犯话术

当然:4 Agent + 4 工具的设计,需要 1 周调试,主要时间花在调教 "Writer 的千人千面模板"——一旦调好,后续每次营销复利。

🧠 我的思考

1. 反直觉点:"情绪雷达"工具不需要训练模型

新人做情感分析,第一反应是"训一个 BERT 分类器"。完全没必要——SentimentAnalysisTool 底层就是"用 GPT-4 让 LLM 判断情感 + 解析 JSON"。

实测:GPT-4 判断情感准确率 88%,自训 BERT 分类器也才 91%——为了 3% 的准确率,付出"训练 + 部署 + 维护"3 个工程师月的成本,得不偿失。结论:80% 的文本分类/情感/意图判断,直接用 LLM 就行,不要训练专用模型。

2. 进阶建议:ThrottleTool 的 Redis 实现要加上"退避策略"

上面示例的节流器是"固定窗口"(1 小时内允许 2 条)。生产环境用"滑动窗口"或"令牌桶"更精细:

# 升级版:令牌桶算法class TokenBucketThrottle:    def __init__(self, capacity=3, refill_rate=1):        self.capacity = capacity       # 桶容量 3 条        self.refill_rate = refill_rate # 每小时补充 1 个令牌    def allow(self, customer_id: str) -> bool:        # 伪代码:用 Redis Lua 脚本实现原子操作        # 每发 1 条扣 1 个令牌,每小时自动补 1 个        ...

滑动窗口 vs 固定窗口对比:

滑动窗口对客户体验更好,但实现复杂——只有日均发送量 > 10 万级的公司才用滑动窗口。

3. 关联资源链接

• 📺 Multi AI Agent Systems with crewAI · Lesson 5 官方视频
• 📓 L5_customer_outreach.ipynb 官方 notebook
• 📖 crewAI 自定义工具文档(@tool + BaseTool)
• 📖 SentimentAnalysisTool 实现原理
• 📐 Token Bucket 限流算法

4. 我的踩坑经验

• SentimentAnalysisTool 接到"反讽"文本会判断错
  ——比如 "哇,这次又升级了,真'惊喜'",LLM 大概率判 POSITIVE,但其实是 NEGATIVE。生产中加 confidence < 0.7 → 默认走人工 review 这个护栏。
• @tool 装饰器的函数必须有 docstring
  ——一旦 docstring 写空,Agent 完全不知道怎么调,直接当工具不存在。docstring 写好等于教 Agent 用工具。
• ThrottleTool 用本地内存存储会丢数据
  ——多机部署时,每个 Agent 进程独立计数,会撞过窗口。用 Redis 共享计数,或者干脆用 cloudflare worker / API gateway 的内置限流。
• Process.hierarchical 在营销场景不要用
  ——Manager 来回派活消耗 30-50% token,但营销场景顺序固定("研究→写→审→发"是流水线),hierarchical 纯属浪费。营销用 sequential,客服才用 hierarchical。

📅 下期预告

Day24 的客服 Agent 是"被动响应",Day25 的营销 Agent 是"主动外呼"——两个 Agent 都跑得通,那多 Agent 之间冲突怎么办?Day26 我们讲 Lesson 6「Multi-agent Collaboration & Conflict」 —— 怎么解决"3 个 Agent 给同一个订单出 3 种处理方案"的难题。

核心升级:Day14-Day25 我们讲了"一个 Agent 的提速、单 Agent 的反思、工具调用、记忆、多 Agent 协作"——Day26 开始处理多 Agent 之间的"内耗":谁先谁后、谁否决谁、谁给谁兜底。

3 条预告要点:

1. 冲突检测:当 2 个 Agent 给同一个客户出 2 个方案(Support 说"退 30%",Escalation 说"退 50%"),谁来拍板?
2. 优先级协议:Reviewer 永远大于 Writer,Manager 永远大于 Worker——但如果 Manager 决策错,谁来纠错(讲 "二级仲裁" 设计)?
3. 降级策略:如果 4 Agent 都在跑,其中一个挂了,其他 3 个怎么保证业务不停?——讲Agent 故障隔离 + 降级到人工兜底的工程实践。

下期我们就拆:多 Agent 系统从"会跑"到"会用"的关键一步 —— 让 Agent 学会"听话 + 不吵架 + 失败了有人兜底"。