简单之美 | 使用 CrewAI 开发 AI Agent 应用

CrewAI 是一个构建 AI Agent 应用的框架，可以非常方便地开发 AI Agent 应用，支持多个 Agent 之间协作以实现特定的任务。有关 CrewAI 中关于 Crew、Agent、Task 等概念，可以参考官网文档了解，下面是官网给出的一个 CrewAI 框架的设计概览，如下图所示：
CrewAI Framework Overview
在使用 CrewAI 开发 AI Agent 应用时，为了方便快速测试流程，可以采用直观的编码方式来体验或验证基本流程，也可以采用工程的方式构建更加复杂的 AI Agent 应用。下面，我们基于 CrewAI 分别通过这两种方式实现一个简单的 AI Agent 应用。

直接编码开发构建

下面，我们通过一个简单的例子，通过直观编码的来实现基本的 AI Agent，通过一个 Agent 和一个 Task 完成报表的查询和可视化输出。为了方便，可以使用 JupyterLab 开发和调试程序。

1.准备工作

在本地使用 Ollama 来运行大模型，从 Ollama 官网下载并安装 Ollama，为方便调试先安装 DeepSeek R1 7B 的模型，大约 5G 大小：

ollama run deepseek-r1:7b

在命令行中可以输入自然语言进行推理，说明安装成功。
然后，在本地安装 CrewAI 的工具包（这里我使用了一个 Anaconda 虚拟环境）：

conda create -n crewai python=3.12
conda activate CrewAI
pip install 'crewai[tools]'

下面可以开发我们的程序，如果需要可以打开 CrewAI 的 Debug 功能：

from langchain_openai import ChatOpenAI
from crewai import LLM

# Turn debug for LLM
import litellm
litellm._logging._turn_on_debug()

2.初始化大模型 LLM

可以使用本地的 LLM，实现代码如下所示：

# Initialize local LLM
llm = ChatOpenAI(
    api_key = "NA",
    model = "ollama/deepseek-r1:7b",
    base_url = "http://localhost:11434/v1")
print("Local LLM: ", llm)

也可以使用调用 API 的方式，代码如下所示：

# Connect remote LLM API
llm = LLM(
    api_key = "sk-xxxxxxx",
    model = "deepseek/Pro/deepseek-ai/DeepSeek-R1",
    base_url = "https://api.xxxx.com/v1",
    temperature=0.3)
print("Remote LLM: ", llm)

使用大模型服务提供商提供的 API，一般需要指定 API Key、模型名称、Base URL，根据自己的情况进行选择。这里需要说明的是，使用 CrewAI 时模型名称 model 的值要求满足格式“LLM 提供商名称/模型名称”，否则运行代码时无法通过验证，具体地 LLM 提供商可以查看 LiteLLM 网站 https://docs.litellm.ai/docs/providers/ 支持哪些 LLM Provider。

3.实现 CrewAI Agent、Task

创建一个 CrewAI 应用，需要实现 Agent、Task，还有构成它们的 Tools。我们实现的简单功能是，通过向 AI Agent 提问，查询指定的接口获取统计结果数据，并实现简单的可视化。这里，抽象出一个 Tool、一个 Task、一个 Agent，对应的实现代码如下所示：

# Create every agent & tool & task
from crewai import Agent, Task, Crew
from crewai.tools import BaseTool

# Define available tools
class ReportQueryTool(BaseTool):
    name: str = "报表查询工具"
    description: str = "根据输入日期范围，查询统计结果数据"

    def _run(self, start_date: str, end_date: str) -> str:
        """
        根据日期范围查询，结果是按天、城市统计的活跃用户数报表结果数据。
        参数:
            - start_date: 开始日期，格式为 yyyyMMdd
            - end_date: 截止日期，格式为 yyyyMMdd
        返回:
            - 查询结果以 JSON 格式字符串返回，结果包含按天、城市两个维度的活跃用户数统计结果
        """
        return '[{"dt":"20240101","city":"BJ","active_user_cnt":12765},{"dt":"20240101","city":"SH","active_user_cnt":823222},{"dt":"20240102","city":"BJ","active_user_cnt": 53025},{"dt":"20240102","city":"SH","active_user_cnt":92001}]'

# Define agent
report_agent = Agent(
    role = "数据报表分析助手",
    goal = "负责查询统计结果报表数据，并进行可视化",
    backstory = "使用提供的工具查询统计结果数据，然后自主实现数据可视化",
    llm = llm,
    tools = [ReportQueryTool()],
    verbose = True
)
print("Agent(report_agent): ", report_agent)

# Define our task
report_task = Task(
    description = "查询2024年1月用户活跃统计结果，然后自主生成简单的可视化表格",
    agent = report_agent,
    tools = [ReportQueryTool()],
    expected_output="包含日期、城市、活跃用户数三个字段的简单表格"
)
print("Task(report_task): ", report_task)

上面代码中，ReportQueryTool 模拟输入一个日期范围作为筛选条件，模拟返回查询到的统计结果记录行，使用 JSON 格式返回。然后配置 Agent 和 Task 都与该 Tool 关联。

4.实现 CrewAI 执行入口程序 Crew

每个 CrewAI 程序都包含一个执行入口，使用 CrewAI 提供的 Crew 类来实现，代码如下所示：

# Define crew
crew = Crew(
    agents = [report_agent],
    tasks = [report_task],
    verbose = True,
    planning = False,
    full_output = True)

result = crew.kickoff()
print(result)

运行上面程序，可以看到执行过程，能够看到程序与 LLM 交互来实现推理并得到最终结果，示例结果输出如下所示：

# Agent: 数据报表分析助手
## Task: 查询用户活跃统计结果，然后自主生成简单的可视化表格


# Agent: 数据报表分析助手
## Using tool: 报表查询工具
## Tool Input:
"{\"start_date\": \"2024-01-01\", \"end_date\": \"2024-01-31\"}"
## Tool Output:
[{"dt":"20240101","city":"BJ","active_user_cnt":12765},{"dt":"20240101","city":"SH","active_user_cnt":823222},{"dt":"20240102","city":"BJ","active_user_cnt": 53025},{"dt":"20240102","city":"SH","active_user_cnt":92001}]


# Agent: 数据报表分析助手
## Final Answer:
| 日期       | 城市 | 活跃用户数 |
|------------|------|------------|
| 2024-01-01 | BJ   | 12,765    |
| 2024-01-01 | SH   | 823,222   |
| 2024-01-02 | BJ   | 53,025    |
| 2024-01-02 | SH   | 92,001    |
```


| 日期       | 城市 | 活跃用户数 |
|------------|------|------------|
| 2024-01-01 | BJ   | 12,765    |
| 2024-01-01 | SH   | 823,222   |
| 2024-01-02 | BJ   | 53,025    |
| 2024-01-02 | SH   | 92,001    |
```

通过工程方式开发构建

对于复杂系统，可以将一些经常修改的内容，如提示词等放进配置文件里面，以方便按需修改，通过通过工程的方式构建就会比较方便一些。
使用 CrewAI 的 CLI 命令可以创建一个 CrewAI 项目：

crewai create crew simple_report_app

可以看到生成的项目结构，如下所示：

simple_report_app/
├── .gitignore
├── pyproject.toml
├── README.md
├── .env
└── src/
    └── simple_report_app/
        ├── __init__.py
        ├── main.py
        ├── crew.py
        ├── tools/
        │   ├── custom_tool.py
        │   └── __init__.py
        └── config/
            ├── agents.yaml
            └── tasks.yaml

接下来，把前面我实现的代码，拆分并转换成对应的 YAML 配置文件、代码文件即可。

main.py

#!/usr/bin/env python

import sys
from simple_report_app.crew import SimpleReporterCrew

def run():
  """
  Run the crew.
  """
  inputs = {
    'topic': 'AI Agents'
  }
  SimpleReporterCrew().crew().kickoff(inputs=inputs)

crew.py

from crewai import Agent, Crew, Process, Task
from crewai.project import CrewBase, agent, crew, task
from custom_tool import ReportQueryTool

@CrewBase
class SimpleReporterCrew():
  """SimpleReporter crew"""

  @agent
  def reporter(self) -> Agent:
    return Agent(
      config = self.agents_config['reporter'],
      verbose = True,
      tools = [ReportQueryTool()]
    )

  @task
  def report_task(self) -> Task:
    return Task(
      config =s elf.tasks_config['report_task'],
    )

  @crew
  def crew(self) -> Crew:
    """Creates the SimpleReporter crew"""
    return Crew(
      agents = self.agents, # Automatically created by the @agent decorator
      tasks = self.tasks, # Automatically created by the @task decorator
      process = Process.sequential,
      verbose = True,
    )

custom_tool.py

from crewai.tools import BaseTool

# Define available tools
class ReportQueryTool(BaseTool):
    name: str = "报表查询工具"
    description: str = "根据输入日期范围，查询统计结果数据"

    def _run(self, start_date: str, end_date: str) -> str:
        """
        根据日期范围查询，结果是按天、城市统计的活跃用户数报表结果数据。
        参数:
            - start_date: 开始日期，格式为 yyyyMMdd
            - end_date: 截止日期，格式为 yyyyMMdd
        返回:
            - 查询结果以 JSON 格式字符串返回，结果包含按天、城市两个维度的活跃用户数统计结果
        """
        return '[{"dt":"20240101","city":"BJ","active_user_cnt":12765},{"dt":"20240101","city":"SH","active_user_cnt":823222},{"dt":"20240102","city":"BJ","active_user_cnt": 53025},{"dt":"20240102","city":"SH","active_user_cnt":92001}]'

agents.yaml

reporter:
  role: >
    数据报表分析助手
  goal: >
    负责查询统计结果报表数据，并进行可视化
  backstory: >
    使用提供的工具查询统计结果数据，然后自主实现数据可视化

tasks.yaml

report_task:
  description: >
    查询2024年1月用户活跃统计结果，然后自主生成简单的可视化表格
  expected_output: >
    包含日期、城市、活跃用户数三个字段的简单表格
  agent: reporter

我们可以修改项目配置文件 pyproject.toml 指定相应的配置，然后安装相关依赖：

cd simple_report_app
crewai install

如果安装没有问题，可以执行运行命令：

crewai run

输出和前面直接编码的方式，结果类似。因为每次都和 LLM 交互进行推理，有可能出现不一样的结果。如果得到的结果不满足预期，就要调整对应的提示词，最终达到我们期望的结果。

参考链接

本文基于署名-非商业性使用-相同方式共享 4.0许可协议发布，欢迎转载、使用、重新发布，但务必保留文章署名时延军（包含链接：http://shiyanjun.cn），不得用于商业目的，基于本文修改后的作品务必以相同的许可发布。如有任何疑问，请与我联系。

发表评论取消回复

石浩枫: 作者你自己看看你写的通顺吗，图layer一半有颜色一半没颜色，画的啥东西
gsgsgsl: 赞一个，前几年搞过kafka2.x版本的鉴权，几年有这方面需求，发现很多api变了，2.x版本的鉴权配置在3.x版...
dack: GPT-2中的相对位置编码请问有出处吗，在GPT-2的论文& #8221;Language Models are...
zhang: 你好，这一行“前面计算已经得到 QKT 矩阵，n=6，dk=8，则 A 的大小也是 6 x 6。”请问在下面的代码中为什么dk =...
derek: 何时才能出现伴侣Ai
丘比特: 请问博主，如果在窗口中用到广播状态，现在您有什么实现方案吗？
z: 寫的真好
方俊: 大佬好有耐心，从14年回复到19年哈哈
Yanjun: 图是用 Astah 和 OminiGraffle 画的
JacobZheng: 问个题外话，图是用什么工具画的啊
Derek Dekker: 感觉还挺难的
luosijie: 博主你好，请问您知道K距离方法出自哪篇文献吗，我该如何引用？

发表评论 取消回复

发表评论取消回复