纲要
- 项目介绍与核心概念
CrewAI多智能体框架的特点- 营销策略生成器的目标
- 项目结构总览
- 智能体与任务配置
- 四个智能体的角色与目标
- 五个任务的流水线与输出
- 数据模型定义
- 营销创意模型
- 策略与文案模型
- 运行模式选择:顺序执行 vs. 层级管理
- 完整可运行代码
- 依赖与入口脚本
- YAML 配置文件
- 执行流程与日志解读
- 效果展示与改进方向
- 总结
项目介绍
CrewAI是一个基于角色的多智能体编排框架,通过 YAML 配置即可快速构建协作型 AI 应用。它抽象了智能体(Agent)、任务(Task)、流程(Process)等组件,让开发者无需深入底层 Prompt 拼接就能搭建出工业级的多智能体系统。
本文将以一个营销策略生成器为例,展示如何利用CrewAI让多个智能体协作完成:市场分析、策略制定、创意生成和文案输出。整个项目采用配置驱动的方式,新手也能在几分钟内跑通。
项目结构
项目使用poetry管理依赖,核心代码与配置文件分离,结构清晰:
marketing-post/ ├── pyproject.toml ├── README.md ├── .env └── src ├── main.py └── marketing_post_crew ├── __init__.py ├── crew.py ├── config │ ├── agents.yaml │ └── tasks.yaml └── models.py.env:存放OPENAI_API_KEY等密钥main.py:程序入口,调用Crew并传递输入crew.py:定义MarketingPostCrew类,组装智能体和任务config/agents.yaml、config/tasks.yaml:智能体与任务的配置文件models.py:Pydantic 数据模型,控制结构化输出
智能体与任务配置
智能体定义(agents.yaml)
在本案例中定义了四个智能体,分别扮演不同角色:
- 首席市场分析师(market_analyst):深入分析产品与竞争对手,输出洞察报告。
- 首席营销策略师(marketing_strategist):综合分析师结果,制定完整营销策略。
- 创意内容创作者(creative_content_creator):根据策略生成高影响力的广告文案与活动创意。
- 首席创意总监(chief_creative_director):审核、优化团队产出,保证质量与目标一致性。
示例配置片段:
chief_market_analyst:role:首席市场分析师goal:对产品和竞争对手进行出色分析,提供深入见解以指导营销策略backstory:>作为一家顶级数字营销公司的首席市场分析师,你专长于剖析在线商业环境, 能从海量信息中提炼出影响竞争格局的关键洞察。chief_marketing_strategist:role:首席营销策略师goal:综合产品分析中的精彩见解,制定卓越的营销策略backstory:>你是一家领先数字营销机构的首席策略师,以打造推动成功的高度定制化策略而闻名。creative_content_creator:role:创意内容创作者goal:开发引人入胜的社交媒体内容,重点创作高影响力的广告文案backstory:>作为顶级营销机构的创意内容创作者,你擅长用叙事与受众共鸣, 产出能在嘈杂信息流中脱颖而出的作品。chief_creative_director:role:首席创意总监goal:监督团队工作,确保产出最佳且与产品目标一致,必要时提出澄清或追加任务backstory:>你是一位资深创意总监,对视觉与文字有极高审美标准, 并能精准把控品牌调性与市场目标之间的平衡。任务定义(tasks.yaml)
共设计五个任务,形成从研究到最终文案的完整流水线:
research_task:在指定行业背景下,对客户及竞争对手进行彻底研究,输出包含人口偏好、市场定位和受众参与度的完整报告。project_understanding_task:理解项目细节与目标受众,收集额外信息,输出详细摘要与受众画像。marketing_strategy_task:制定详细营销策略,涵盖目标、关键信息、建议战术、渠道和 KPI。campaign_ideation_task:基于策略生成五个营销活动创意,每个附带简短描述与预期效果。copywriting_task:为每个创意活动撰写最终文案,包括标题和正文。
每个任务都通过expected_output规定了输出格式,配合结构化数据模型,确保结果可直接解析。
数据模型设计
为了让智能体的输出结构化且可控,我们定义了三个 Pydantic 模型:
frompydanticimportBaseModel,FieldfromtypingimportList,OptionalclassCampaignIdea(BaseModel):name:str=Field(...,description="活动名称")description:str=Field(...,description="活动简短描述")target_audience:str=Field(...,description="目标受众")channel:str=Field(...,description="营销渠道(如抖音、公众号、小红书)")expected_impact:str=Field(...,description="预期影响")classMarketingStrategy(BaseModel):strategy_name:str=Field(...,description="策略名称")tactics:List[str]=Field(...,description="战术列表")channels:List[str]=Field(...,description="营销渠道")kpis:List[str]=Field(...,description="关键绩效指标")target_audience:str=Field(...,description="目标受众描述")key_message:str=Field(...,description="关键信息")classAdCopy(BaseModel):title:str=Field(...,description="文案标题")body:str=Field(...,description="文案正文")campaign_name:Optional[str]=Field(None,description="所属活动名称")这些模型会绑定到对应任务的output_pydantic字段,CrewAI内部会引导大模型输出符合 Schema 的 JSON 结果。
运行模式选择
CrewAI支持两种任务执行流程:
| 模式 | Process 参数 | 说明 |
|---|---|---|
| 顺序执行 | Process.sequential | 任务一个接一个线性执行,上一个任务的输出作为下一个的输入 |
| 层级管理 | Process.hierarchical | 指定一个 Manager LLM 或 Manager Agent 进行任务委派、审核与调度,类似“主管-成员”结构 |
层级模式更适合需要全局把控与质量审查的场景。本案例使用层级模式,并设置一个manager_llm负责分配任务给各个专家角色。
完整可运行代码
下面提供一个可以直接运行的 Python 项目,只需配置好 API Key 并安装依赖即可执行。
环境准备
新建项目目录,创建pyproject.toml(或用requirements.txt):
[tool.poetry] name = "marketing-post-crew" version = "0.1.0" description = "Multi-agent marketing strategy generator with CrewAI" authors = ["Your Name <you@example.com>"] [tool.poetry.dependencies] python = "^3.10" crewai = ">=0.28.0" python-dotenv = "^1.0.0" [build-system] requires = ["poetry-core"] build-backend = "poetry.core.masonry.api"运行poetry install安装依赖。然后创建.env文件:
OPENAI_API_KEY=your-api-key-here主入口脚本(src/main.py)
importosfromdotenvimportload_dotenvfrommarketing_post_crew.crewimportMarketingPostCrew load_dotenv()defrun():inputs={"product_name":"智能会议记录仪","industry":"企业服务 / SaaS","target_audience":"中小企业主、远程办公团队、会议频繁的项目经理","competitors":"Notion AI, Otter.ai, Fireflies.ai","unique_selling_point":"本地化多语言会议纪要,支持定制化模板与私有部署"}crew=MarketingPostCrew().crew()result=crew.kickoff(inputs=inputs)print("\n======== 最终输出 ========\n")print(result)if__name__=="__main__":run()Crew 组装(src/marketing_post_crew/crew.py)
fromcrewaiimportAgent,Task,Crew,Processfromcrewai.projectimportCrewBase,agent,task,crewfrommarketing_post_crew.modelsimportCampaignIdea,MarketingStrategy,AdCopyimportos@CrewBaseclassMarketingPostCrew:agents_config=os.path.join(os.path.dirname(__file__),'config','agents.yaml')tasks_config=os.path.join(os.path.dirname(__file__),'config','tasks.yaml')# -------------------- Agents --------------------@agentdefchief_market_analyst(self)->Agent:returnAgent(config=self.agents_config['chief_market_analyst'],verbose=True,allow_delegation=False)@agentdefchief_marketing_strategist(self)->Agent:returnAgent(config=self.agents_config['chief_marketing_strategist'],verbose=True,allow_delegation=False)@agentdefcreative_content_creator(self)->Agent:returnAgent(config=self.agents_config['creative_content_creator'],verbose=True,allow_delegation=False)@agentdefchief_creative_director(self)->Agent:returnAgent(config=self.agents_config['chief_creative_director'],verbose=True,allow_delegation=True)# -------------------- Tasks --------------------@taskdefresearch_task(self)->Task:returnTask(config=self.tasks_config['research_task'],output_file='output/research_report.md')@taskdefproject_understanding_task(self)->Task:returnTask(config=self.tasks_config['project_understanding_task'],output_file='output/project_summary.md')@taskdefmarketing_strategy_task(self)->Task:returnTask(config=self.tasks_config['marketing_strategy_task'],output_pydantic=MarketingStrategy,output_file='output/strategy.json')@taskdefcampaign_ideation_task(self)->Task:returnTask(config=self.tasks_config['campaign_ideation_task'],output_pydantic=CampaignIdea,output_file='output/campaign_ideas.json')@taskdefcopywriting_task(self)->Task:returnTask(config=self.tasks_config['copywriting_task'],output_pydantic=AdCopy,output_file='output/ad_copies.json')@crewdefcrew(self)->Crew:returnCrew(agents=[self.chief_market_analyst(),self.chief_marketing_strategist(),self.creative_content_creator(),self.chief_creative_director()],tasks=[self.research_task(),self.project_understanding_task(),self.marketing_strategy_task(),self.campaign_ideation_task(),self.copywriting_task()],process=Process.hierarchical,manager_llm=self.chief_market_analyst().llm,# 使用默认配置的 LLM 作为管理者verbose=True)任务配置文件(src/marketing_post_crew/config/tasks.yaml)
research_task:description:>在 {industry} 背景下,对产品 {product_name} 及其竞争对手 {competitors} 进行全面研究。 分析市场趋势、用户偏好、竞争对手的优势与弱点。expected_output:>一份包含客户分析、竞争对手对比、市场定位建议、人口统计偏好和受众参与度的完整报告。project_understanding_task:description:>审查项目材料,深刻理解产品 {product_name} 的独特卖点 ({unique_selling_point}) 及目标受众 ({target_audience})。 收集任何缺失的信息,确保所有细节明确。expected_output:>项目详细摘要与目标受众画像,包括需求、痛点、获取信息的渠道偏好。marketing_strategy_task:description:>基于研究结果和受众画像,为 {product_name} 制定一套完整的营销策略。 需包含目标、关键信息、建议战术、营销渠道和 KPI 指标。expected_output:>结构化的营销策略文档,符合 MarketingStrategy 数据模型。campaign_ideation_task:description:>根据营销策略,发散思维生成 5 个具体营销活动创意。 每个创意应包含名称、简短描述、目标受众、渠道和预期效果。expected_output:>5 个 CampaignIdea 对象的列表。copywriting_task:description:>为每个营销活动撰写最终投放文案,包括吸引人的标题和正文。expected_output:>AdCopy 对象列表,每条对应一个活动。运行结果示例
执行poetry run python src/main.py后,控制台会输出各智能体的思考过程和最终生成的 JSON 文件。日志清楚地展示了 Manager LLM 如何将任务委派给不同专家,以及各智能体之间的数据流转。
最终output/目录下会生成:
research_report.md:市场研究报告project_summary.md:项目摘要与受众画像strategy.json:符合MarketingStrategy模型的策略文档campaign_ideas.json:五个活动创意ad_copies.json:对应的广告文案
整个流程在几分钟内完成,产出的营销材料可直接供人类团队评审、修改后使用。
总结
本文基于CrewAI构建了一个完整的多智能体营销策略生成器,从市场分析到最终文案输出全由多个 AI 智能体协同完成。通过 YAML 配置与 Pydantic 数据模型,开发者可以快速定义角色、任务和输出格式,极大地降低了多智能体应用的门槛。层级管理模式让任务分配与质量控制更加自动化,适合真实商业场景。
本文覆盖了原文中提到的项目结构、智能体配置、任务流程、运行模式以及最终的输出展示,并补充了代码示例。