在 AI 大模型时代，单一智能体已难以应对复杂任务，多智能体协作框架成为新趋势。CrewAI 作为开源多智能体编排工具，能模拟真实团队协作流程，让不同角色智能体各司其职。本文将结合完整源码，带大家从零实现一个自动化图书生成系统，涵盖从大纲设计到章节撰写的全流程。

一、CrewAI 核心基础：你需要了解的关键概念

在开始实战前，先明确 CrewAI 的核心组件与优势，这是搭建系统的基础。

1.1 核心组件解析

CrewAI 通过五大核心组件实现多智能体协作，各组件职责清晰且相互关联：

• Agent（智能体）：具备特定角色的 AI 实体，拥有role（角色）、goal（目标）、backstory（背景故事）三大关键属性，可搭配工具完成专业任务。
• Task（任务）：智能体需完成的具体工作单元，包含description（任务描述）、expected_output（预期输出）、agent（负责智能体）等属性，支持任务依赖配置。
• Crew（团队）：智能体与任务的组织容器，负责协调整体流程，可设置process（执行模式，如顺序 / 并行）、verbose（日志开关）等参数。
• Flow（流程）：提供精细化工作流控制，支持@start（流程起点）、@listen（事件监听）、@router（条件路由），实现状态管理与复杂逻辑。
• Tool（工具）：智能体的能力扩展，如网络搜索、PDF 解析、向量数据库查询等，可通过装饰器快速封装自定义工具。

1.2 CrewAI 核心优势

相比传统单模型应用，CrewAI 的优势体现在协作与实用性上：

• 角色驱动设计：每个智能体有明确分工，如 “研究员”“作家”“编辑”，贴合真实工作场景。
• 智能协作能力：智能体间可自主委派任务、共享信息，无需人工干预流程衔接。
• 灵活工具集成：支持 DDGS 搜索、SerperDevTool、自定义 PDF 解析等工具，轻松扩展功能。
• 生产级特性：内置内存管理、知识库、错误处理机制，可直接用于实际项目。

二、环境搭建：从安装到项目初始化

搭建图书生成系统需先配置开发环境，确保依赖包与项目结构正确。

2.1 系统与依赖要求

• 基础环境：Python 3.10+、pip/uv 包管理器
• 核心依赖：通过 pip 安装 CrewAI 及相关工具

# 安装CrewAI核心库
pip install crewai
# 安装CrewAI工具集
pip install crewai-tools
# 安装网络搜索工具
pip install ddgs
# 安装PDF解析与向量存储依赖
pip install markitdown chonkie qdrant-client sentence-transformers

2.2 项目结构设计

采用 “模块化 + 配置化” 结构，便于维护与扩展，完整目录如下：

write_a_book/
├── src/
│   ├── write_a_book/
│   │   ├── crews/  # 智能体团队目录
│   │   │   ├── outline_book_crew/  # 大纲生成团队
│   │   │   │   ├── config/
│   │   │   │   │   ├── agents.yaml  # 大纲团队智能体配置
│   │   │   │   │   └── tasks.yaml   # 大纲团队任务配置
│   │   │   │   └── outline_book_crew.py  # 大纲团队逻辑
│   │   │   └── write_book_chapter_crew/  # 章节撰写团队
│   │   │       ├── config/
│   │   │       │   ├── agents.yaml  # 章节团队智能体配置
│   │   │       │   └── tasks.yaml   # 章节团队任务配置
│   │   │       └── write_book_chapter_crew.py  # 章节团队逻辑
│   │   ├── tools/  # 工具目录
│   │   │   ├── ddgs_tools.py  # DDGS搜索工具
│   │   │   └── custom_tool.py  # 自定义工具模板
│   │   ├── types.py  # 数据模型定义
│   │   └── main.py  # 主流程入口
└── tests/  # 测试目录

三、核心模块开发：从智能体到工具封装

分模块实现大纲生成、章节撰写、工具集成三大核心功能，每个模块遵循 “配置 + 逻辑” 分离原则。

3.1 数据模型定义（types.py）

使用 Pydantic 定义结构化数据模型，确保数据一致性，核心模型包括：

from typing import List
from pydantic import BaseModel

# 章节大纲模型：存储章节标题与描述
class ChapterOutline(BaseModel):
    title: str
    description: str

# 图书大纲模型：包含多个章节大纲
class BookOutline(BaseModel):
    chapters: List[ChapterOutline]

# 章节内容模型：存储章节标题与完整内容
class Chapter(BaseModel):
    title: str
    content: str

3.2 工具封装：实现网络搜索能力

以 DDGS（DuckDuckGo 搜索）为例，封装自定义工具，供智能体调用：

# ddgs_tools.py
from crewai.tools import tool
from ddgs import DDGS

@tool("DDGS Text Search")  # 工具名称，智能体将通过此名称识别
def ddgs_text_search(query: str, max_results: int = 5) -> str:
    """
    使用DDGS进行文本搜索，获取相关信息
    Args:
        query: 搜索关键词
        max_results: 最大返回结果数，默认5条
    Returns:
        格式化的搜索结果字符串
    """
    ddgs = DDGS()
    try:
        results = ddgs.text(
            query=query,
            max_results=max_results,
            backend="auto",
            region='zh-cn',
            safesearch='moderate'
        )
        # 格式化输出结果
        output = []
        for i, result in enumerate(results, 1):
            output.append(f"\n结果 {i}:")
            output.append(f"标题: {result.get('title', 'N/A')}")
            output.append(f"链接: {result.get('href', 'N/A')}")
            output.append(f"摘要: {result.get('body', 'N/A')[:200]}...")
            output.append("-" * 50)
        return "\n".join(output)
    except Exception as e:
        return f"搜索失败: {str(e)}"

3.3 大纲生成团队开发

大纲生成团队负责根据主题生成书籍结构，包含 “研究员” 和 “大纲设计师” 两个智能体。

3.3.1 智能体配置（agents.yaml）

# 研究智能体：收集主题相关信息
research_agent:
  role: >
    研究代理
  goal: >
    收集关于{topic}的全面信息，用于创建有组织且结构良好的书籍大纲。
    以下是作者对书籍期望目标的一些额外信息:\n\n {goal}
  backstory: >
    你是一位经验丰富的研究员，以收集最佳资源和理解任何主题的关键要素而闻名。
    你的目标是收集所有相关信息，以便书籍大纲准确且信息丰富。

# 大纲生成智能体：将研究结果转化为结构化大纲
outline_agent:
  role: >
    图书概述代理
  goal: >
    基于研究，生成关于以下主题的书籍大纲: {topic}
    生成的大纲应包括按顺序排列的所有章节，并为每个章节提供标题和描述。
    以下是作者对书籍期望目标的一些额外信息:\n\n {goal}
  backstory: >
    你是一位熟练的组织者，擅长将零散的信息转化为结构化格式。
    你的目标是创建清晰、简洁的章节大纲，涵盖所有关键主题和子主题。

3.3.2 任务配置（tasks.yaml）

# 主题研究任务：由研究智能体执行
research_topic:
  description: >
    研究{topic}提供的主题，收集创建书籍大纲所需的重要信息。
    确保专注于高质量、可靠的来源。
    以下是作者对书籍期望目标的一些额外信息:\n\n {goal}
  expected_output: >
    一组关于{topic}的关键点和重要信息，用于创建大纲。
  agent: research_agent  # 指定执行智能体

# 大纲生成任务：由大纲智能体执行
generate_outline:
  description: >
    根据研究结果，按顺序创建包含章节的书籍大纲。
    确保每个章节有标题和简要描述，突出涵盖的主题和子主题。
    每个章节字数不超过3000字，且无重复章节或主题。
    以下是作者对书籍期望目标的一些额外信息:\n\n {goal}
  expected_output: >
    包含章节标题和描述的书籍大纲，说明每个章节将包含的内容。
  agent: outline_agent  # 指定执行智能体

3.3.3 团队逻辑实现（outline_book_crew.py）

from crewai import Agent, Crew, Process, Task, LLM
from crewai.project import CrewBase, agent, crew, task
from crewai_tools import SerperDevTool
from write_a_book.src.write_a_book.types import BookOutline

@CrewBase  # 标记为Crew团队类，自动收集智能体与任务
class OutlineCrew:
    # 配置文件路径
    agents_config = "config/agents.yaml"
    tasks_config = "config/tasks.yaml"
    
    # 初始化LLM（使用DeepSeek模型）
    llm = LLM(
        model="deepseek/deepseek-chat",
        api_key="sk-23c834bcca994706a105abb54adf212e",  # 替换为你的API密钥
        base_url="https://api.deepseek.com/v1",
        model_kwargs={"llm_provider": "deepseek"}
    )
    
    @agent  # 标记为智能体，自动注册到团队
    def research_agent(self) -> Agent:
        """创建研究智能体，具备网络搜索能力"""
        search_tool = SerperDevTool()  # 加载搜索工具
        return Agent(
            config=self.agents_config["research_agent"],
            tools=[search_tool],  # 为智能体分配工具
            llm=self.llm,
            verbose=True  # 启用日志，便于调试
        )
    
    @agent
    def outline_agent(self) -> Agent:
        """创建大纲生成智能体"""
        return Agent(
            config=self.agents_config["outline_agent"],
            llm=self.llm,
            verbose=True
        )
    
    @task  # 标记为任务，自动注册到团队
    def research_topic(self) -> Task:
        """创建主题研究任务"""
        return Task(
            config=self.tasks_config["research_topic"]
        )
    
    @task
    def generate_outline(self) -> Task:
        """创建大纲生成任务，指定输出格式为BookOutline"""
        return Task(
            config=self.tasks_config["generate_outline"],
            output_pydantic=BookOutline  # 强制输出结构化数据
        )
    
    @crew  # 标记为团队，整合智能体与任务
    def crew(self) -> Crew:
        """创建大纲生成团队，设置执行模式为顺序执行"""
        return Crew(
            agents=self.agents,  # 自动获取@agent标记的智能体
            tasks=self.tasks,    # 自动获取@task标记的任务
            process=Process.sequential,  # 任务按顺序执行
            verbose=True
        )

3.4 章节撰写团队开发

章节撰写团队基于大纲生成具体内容，包含 “研究员” 和 “作家” 两个智能体，逻辑与大纲团队类似。

3.4.1 智能体配置（agents.yaml）

# 研究智能体：收集章节相关信息
researcher_agent:
  role: >
    Research Agent
  goal: >
    收集关于{topic}和{chapter_title}的全面信息，用于增强章节内容。
    以下是作者对书籍和章节期望目标的一些额外信息:\n\n {goal}
    以下是章节大纲的描述:\n\n {chapter_description}
  backstory: >
    你是一位经验丰富的研究员，擅长寻找任何给定主题中最相关和最新的信息。
    你的工作是提供有见地的数据，以支持和丰富章节的写作过程。

# 作家智能体：撰写章节内容
writer_agent:
  role: >
    Chapter Writer
  goal: >
    根据提供的章节标题、目标和大纲，撰写结构良好的书籍章节。
    章节应以Markdown格式编写，并包含约3000字。
  backstory: >
    你是一位出色的作家，以创作引人入胜、研究充分且信息丰富的内容而闻名。
    你擅长将复杂的想法转化为可读性强且组织良好的章节。

3.4.2 团队逻辑实现（write_book_chapter_crew.py）

from crewai import Agent, Crew, Process, Task, LLM
from crewai.project import CrewBase, agent, crew, task
from write_a_book.src.write_a_book.tools.ddgs_tools import ddgs_text_search
from write_a_book.src.write_a_book.types import Chapter

@CrewBase
class WriteBookChapterCrew:
    agents_config = "config/agents.yaml"
    tasks_config = "config/tasks.yaml"
    
    # 初始化LLM
    llm = LLM(
        model="deepseek/deepseek-chat",
        api_key="sk-23c834bcca994706a105abb54adf212e",
        base_url="https://api.deepseek.com/v1",
        model_kwargs={"llm_provider": "deepseek"}
    )
    
    @agent
    def researcher_agent(self) -> Agent:
        """创建研究智能体，使用DDGS搜索工具"""
        return Agent(
            config=self.agents_config["researcher_agent"],
            tools=[ddgs_text_search],  # 分配自定义搜索工具
            llm=self.llm,
            verbose=True
        )
    
    @agent
    def writer_agent(self) -> Agent:
        """创建作家智能体"""
        return Agent(
            config=self.agents_config["writer_agent"],
            llm=self.llm,
            verbose=True
        )
    
    @task
    def research_chapter(self) -> Task:
        """创建章节研究任务"""
        return Task(
            config=self.tasks_config["research_chapter"]
        )
    
    @task
    def write_chapter(self) -> Task:
        """创建章节撰写任务，输出格式为Chapter"""
        return Task(
            config=self.tasks_config["write_chapter"],
            output_pydantic=Chapter
        )
    
    @crew
    def crew(self) -> Crew:
        """创建章节撰写团队，顺序执行任务"""
        return Crew(
            agents=self.agents,
            tasks=self.tasks,
            process=Process.sequential,
            verbose=True
        )

四、工作流整合：实现端到端图书生成

通过 CrewAI 的 Flow 功能，串联大纲生成、章节撰写、内容保存三大步骤，实现自动化流程。

4.1 工作流逻辑（main.py）

import asyncio
from typing import List
from crewai.flow.flow import Flow, listen, start
from pydantic import BaseModel, Field
from crews.write_book_chapter_crew.write_book_chapter_crew import WriteBookChapterCrew
from write_a_book.src.write_a_book.types import Chapter, ChapterOutline
from crews.outline_book_crew.outline_book_crew import OutlineCrew

# 定义书籍生成流程的状态模型
class BookState(BaseModel):
    id: str = "book_flow_instance"  # Flow框架要求的唯一ID
    title: str = "2024年9月人工智能的现状"  # 书籍标题
    book: List[Chapter] = Field(default_factory=lambda: list, description="书籍章节列表")
    book_outline: List[ChapterOutline] = Field(default_factory=lambda: list, description="书籍大纲列表")
    topic: str = "探索截至2025年1月各行业中人工智能的最新趋势"  # 书籍主题
    goal: str = """
        本书的目标是全面概述2025年10月人工智能的现状。
        它将深入探讨影响各行业的最新趋势，分析重大进展，
        并讨论未来可能的发展方向。本书旨在向读者介绍前沿的人工智能技术，
        并为他们迎接该领域的未来创新做好准备。
    """

# 定义书籍生成工作流
class BookFlow(Flow[BookState]):
    initial_state = BookState  # 初始状态
    
    @start()  # 标记为流程起点
    def generate_book_outline(self):
        """第一步：生成书籍大纲"""
        print("启动大纲生成团队")
        # 调用大纲生成团队
        output = OutlineCrew().crew().kickoff(
            inputs={"topic": self.state.topic, "goal": self.state.goal}
        )
        chapters = output["chapters"]
        self.state.book_outline = chapters  # 保存大纲到状态
        return chapters
    
    @listen(generate_book_outline)  # 监听大纲生成完成事件
    async def write_chapters(self):
        """第二步：异步撰写所有章节"""
        print("开始撰写章节")
        tasks = []
        
        # 定义单个章节的异步撰写函数
        async def write_single_chapter(chapter_outline):
            output = WriteBookChapterCrew().crew().kickoff(
                inputs={
                    "goal": self.state.goal,
                    "topic": self.state.topic,
                    "chapter_title": chapter_outline.title,
                    "chapter_description": chapter_outline.description,
                    "book_outline": [c.model_dump_json() for c in self.state.book_outline]
                }
            )
            return Chapter(title=output["title"], content=output["content"])
        
        # 为每个章节创建异步任务
        for outline in self.state.book_outline:
            print(f"撰写章节：{outline.title}")
            tasks.append(asyncio.create_task(write_single_chapter(outline)))
        
        # 等待所有章节完成
        chapters = await asyncio.gather(*tasks)
        self.state.book.extend(chapters)  # 保存章节到状态
    
    @listen(write_chapters)  # 监听章节撰写完成事件
    async def join_and_save_chapter(self):
        """第三步：整合章节并保存为Markdown文件"""
        print("整合并保存书籍")
        # 拼接章节内容
        book_content = ""
        for chapter in self.state.book:
            book_content += f"# {chapter.title}\n\n"
            book_content += f"{chapter.content}\n\n"
        
        # 保存为文件
        filename = f"./{self.state.title.replace(' ', '_')}.md"
        with open(filename, "w", encoding="utf-8") as f:
            f.write(book_content)
        print(f"书籍已保存至：{filename}")
        return book_content

# 启动工作流
def kickoff():
    book_flow = BookFlow()
    book_flow.kickoff()

if __name__ == "__main__":
    kickoff()

4.2 运行与测试

执行主程序，即可自动完成图书生成：

python src/write_a_book/main.py

运行过程中，终端会输出各智能体的执行日志，最终在项目根目录生成 Markdown 格式的书籍文件（如2024年9月人工智能的现状.md）。

五、最佳实践与优化建议

为提升系统稳定性与效率，结合 CrewAI 官方推荐实践，给出以下优化方向：

5.1 智能体设计优化

• 明确角色定位：避免 “万能助手” 式智能体，如 “SEO 优化专家” 比 “助手” 更具体。
• 精简工具分配：仅为智能体分配必要工具，如 “作家” 无需搜索工具，减少资源浪费。
• 启用记忆功能：对需要上下文的任务（如客服），设置memory=True，让智能体记住历史对话。

5.2 性能与成本控制

• 启用缓存：对重复查询任务，设置cache=True，避免重复调用 LLM，降低 API 成本。
• 分级使用 LLM：简单任务（如数据收集）用 GPT-3.5，复杂任务（如战略分析）用 GPT-4。
• 并行执行任务：对独立章节的撰写，通过async_execution=True实现并行，提升效率。

5.3 错误处理

添加异常捕获与重试逻辑，确保流程稳定：

try:
    # 执行团队任务
    result = crew.kickoff(inputs={"topic": "人工智能"})
except Exception as e:
    print(f"执行失败：{str(e)}")
    # 重试逻辑
    time.sleep(3)
    result = crew.kickoff(inputs={"topic": "人工智能"})

六、总结与扩展

本文通过完整案例，展示了如何用 CrewAI 搭建多智能体协作的图书生成系统。该系统不仅能自动生成书籍，还可扩展到其他场景：

• 内容创作：如博客、报告、剧本的自动化生成。
• 市场分析：整合数据收集、竞品分析、报告生成智能体。
• 客户支持：通过 RAG 工具对接知识库，实现智能客服。

后续可进一步集成 Streamlit 搭建 Web 界面，让非技术用户也能通过可视化操作生成图书；或接入向量数据库，增强智能体的知识检索能力。