Type something to search...
to navigateto selectESC to close
构建更智能的 API：CrewAI 与 FastAPI 集成指南

构建更智能的 API：CrewAI 与 FastAPI 集成指南

目录

· 1. 引言 ∘ 客户支持代理 ∘ 个人财务顾问 ∘ 医疗助手 ∘ 学习伴侣 ∘ 项目管理助手 ∘ 创意合作伙伴 ∘ 智能家居管理器 ∘ 网络安全助手· 2. 搭建基础· 3. 理解 CrewAI ∘ 1. AI 代理 ∘ 2. 工具 ∘ 3. 过程 ∘ 4. 任务· 4. 使用 CrewAI 设置 Fast API· 结论· 了解更多

1. 引言

在2025年，我们有不同的方法来开发软件，特别是随着最新的AI改进，AI通过各种工具和应用进入了我们的生活，即使是我们最好的IDE，如Intellij IDEA或Microsoft Visual Code。

在介绍Crew AI之前，我想简要解释一下什么是LLM。**大型语言模型（LLMs）是可以理解和生成自然语言文本的机器学习模型。这是一种人工智能（AI）**程序，可以识别和生成文本，执行其他任务。

简单来说，LLM已经接收了足够的数据，以便能够识别和解释人类语言或其他类型的数据。LLM可以自动化复杂和顺序的工作流程和任务。例如，您可以使用LLM构建助手，能够代表您自主在线订购产品并安排其在应用中的交付。这些基于LLM的助手被称为代理。

代理是一个LLM驱动的助手，被分配特定的任务和工具来完成这些任务。在其基本形式中，一个典型的AI代理可能配备有记忆，以存储和管理用户交互，与外部数据源通信，并使用功能来执行其任务。代理可以执行的常见示例包括以下内容。

客户支持代理

AI 代理可以作为 24/7 客户服务代表，处理常见问题解答，解决客户问题，并将复杂查询升级到人工代理。例如，电子商务应用中的 AI 代理可以实时协助用户跟踪订单、处理退货或提供产品推荐。

个人财务顾问

在财务应用中的AI代理可以充当虚拟顾问，帮助用户管理预算、分析消费模式，并根据他们的财务目标建议投资机会。例如，在分析用户的风险状况后，它可能会推荐特定的共同基金或ETF。

医疗助理

一个专注于医疗的代理可以帮助患者预约医生、提醒他们的用药时间表，或回答基本的健康查询。例如，代理可以通过分析可穿戴设备的数据来帮助用户监测慢性病，并提供健康见解。

学习伴侣

在教育中，AI 代理可以作为辅导员，通过个性化学习计划引导学习者。它可以通过解释困难的概念、建议额外的资源，甚至创建练习测验来帮助用户巩固学习。

项目管理助手

集成到项目管理工具中的代理可以帮助组织任务、设定截止日期和自动安排会议。例如，它可以分析团队进展，识别瓶颈，并建议解决方案以提高生产力。

创意合作伙伴

AI代理可以在艺术领域充当共同创作者。例如，在内容创作中，代理可以帮助生成创意、撰写剧本或创建图形设计。它可能会分析趋势，以建议与特定受众产生共鸣的创意格式。

智能家居管理器

在智能家居生态系统中，AI代理可以自动化和优化家庭操作。例如，它可以根据房主的偏好和行为调整照明、控制恒温器，甚至推荐节能技巧。

网络安全助手

一个AI代理可以实时监控网络活动，检测异常，并对潜在威胁做出响应。它可能通过阻止可疑IP或通知管理员关键漏洞来采取主动措施。

这些例子说明了AI代理如何适应各种场景，提高效率、便利性和用户体验。请告诉我是否希望深入了解其中的任何内容！

多代理平台已经开发出来以管理如此复杂的AI工作流， crewAI就是其中之一。在本文中，我将使用crewai开发一个工作流，并使其可以通过fastapi从外部调用，作为额外内容，还将提供一个后台任务机制以支持并发请求。

2. 设置舞台

让我们创建一个文件夹（你可以随意命名），我称之为 app。

mkdir -p app

创建虚拟环境。我假设你已经安装了 python，顺便说一下，我在 python 3.12 上尝试过，但在 python 3.13 版本的 crew ai 依赖项上遇到了问题。

python -m venv .venv

激活虚拟环境，然后安装依赖项，在此之前创建 requirements.txt 文件。

crewai
fastapi
uvicorn
python-dotenv
pydantic
celery
requests

然后，其他的就来了……这将激活虚拟环境并安装项目所需的依赖项。

source venv/bin/activate
pip install -r requirements.txt

3. 理解 CrewAI

crewAI 是一个由 João Moura 创建的开源多智能体编排框架。这个基于 Python 的框架利用 人工智能 (AI) 协作，通过编排角色扮演的自主 AI 代理，使它们共同作为一个紧密的团队或“船员”来完成任务。crewAI 的目标是提供一个强大的框架来自动化多智能体工作流。

我不会深入探讨 crewai 的工作原理，因为本文旨在创建一个由 crew ai 支持的快速 API。有关 crewai 工作原理的更多信息，您可以查看 该文档。

该图提供了 Crew 的概念框架，重点关注 AI 代理在完成任务以实现特定结果中的角色和协作。以下是关键组件的详细分解：

1. AI代理

  • 该框架显示了多个AI代理，它们在顶部以黑框的形式呈现。
  • 这些代理协同工作，可以委派任务或相互提问。

2. 工具

  • 一个特定部分强调 工具，代理可以使用这些工具来执行他们的任务。工具是增强代理能力的外部工具或功能。

3. 过程

  • 过程 定义了 AI 代理如何协作。
  • 这包括：
  • 任务如何分配。
  • 代理之间如何互动。
  • 代理如何执行其工作。

4. 任务

  • 任务 在底部显示，代表代理需要处理的单个操作或责任。
  • 任务可以：
  • 通过指定使用哪些工具来覆盖代理工具。
  • 将特定任务分配给特定代理。

4. 使用 CrewAI 设置 Fast API

在跳入演示之前，我想向您保证，我为您准备了一个 repository，请随意使用。您也可以打开我希望解决的问题。

  • 在这个 repository 中，我创建了 analyzer.py Python 脚本，并在其中添加了我的端点，您将能够调用以启动 crewai 作为后台任务。
from fastapi import APIRouter, HTTPException, BackgroundTasks
from app.models.models import TopicRequest, TaskResponse
from app.services.services import BotService

router = APIRouter()

@router.post("/analyze", response_model=TaskResponse)
async def analyze_topic(request: TopicRequest, background_tasks: BackgroundTasks):
    task_id = BotService.create_task(request.topic)
    background_tasks.add_task(BotService.process_task, task_id, request.topic)
    return BotService.get_task_status(task_id)

@router.get("/task/{task_id}", response_model=TaskResponse)
async def get_task_status(task_id: str):
    task = BotService.get_task_status(task_id)
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    return task

此代码片段定义了一个 FastAPI 路由器，具有用于创建和管理与主题分析相关的后台任务的端点。

  • services.py 中的 BotService 类管理分析主题的任务生命周期，使用 UrlInsightBot，包括任务创建、异步处理、状态更新和日志记录。
import asyncio
import uuid
import logging
from typing import Dict
from app.models.models import TaskStatus, TaskResponse
from app.crew.crew import UrlInsightBot

## Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

class BotService:
    _tasks: Dict[str, TaskResponse] = {}
    
    @classmethod
    def create_task(cls, topic: str) -> str:
        task_id = str(uuid.uuid4())
        cls._tasks[task_id] = TaskResponse(
            task_id=task_id,
            status=TaskStatus.PENDING
        )
        logger.info(f"Task {task_id} created with status PENDING for topic: {topic}")
        return task_id

    @classmethod
    async def process_task(cls, task_id: str, topic: str):
        try:
            cls._tasks[task_id].status = TaskStatus.PROCESSING
            logger.info(f"Task {task_id} status changed to PROCESSING")
            bot = UrlInsightBot()
            result = await bot.crew().kickoff_async(inputs={'topic': topic})
            cls._tasks[task_id].status = TaskStatus.COMPLETED
            cls._tasks[task_id].result = result
            logger.info(f"Task {task_id} completed successfully with status {cls._tasks[task_id].status}")
        except Exception as e:
            cls._tasks[task_id].status = TaskStatus.FAILED
            cls._tasks[task_id].error = str(e)
            logger.error(f"Task {task_id} failed with error: {e}")
            
    @classmethod
    async def process_task_sleep(cls, task_id: str, topic: str):
        try:
            cls._tasks[task_id].status = TaskStatus.PROCESSING
            logger.info(f"Task {task_id} status changed to PROCESSING")
            # Simulate processing
            await asyncio.sleep(5)  # Simulate a long-running task
            cls._tasks[task_id].status = TaskStatus.COMPLETED
            cls._tasks[task_id].result = f"Processed topic: {topic}"
            logger.info(f"Task {task_id} completed successfully with status {cls._tasks[task_id].status}")
        except Exception as e:
            cls._tasks[task_id].status = TaskStatus.FAILED
            cls._tasks[task_id].error = str(e)

    @classmethod
    def get_task_status(cls, task_id: str) -> TaskResponse:
        return cls._tasks.get(task_id)
  • 此代码使用 CrewAI 框架定义了一个 UrlInsightBot crew 类，使用 YAML 文件配置代理和任务，并设置两个代理，研究员和报告分析师，及其各自的配置。
from crewai import Agent, Crew, Process, Task
from crewai.project import CrewBase, agent, crew, task
import logging

## If you want to run a snippet of code before or after the crew starts, 
## you can use the @before_kickoff and @after_kickoff decorators
## https://docs.crewai.com/concepts/crews#example-crew-class-with-decorators

## Suppress logs from LiteLLM and httpx
logging.getLogger("LiteLLM").setLevel(logging.WARNING)

@CrewBase
class UrlInsightBot():
 """UrlInsightBot crew"""

 # Learn more about YAML configuration files here:
 # Agents: https://docs.crewai.com/concepts/agents#yaml-configuration-recommended
 # Tasks: https://docs.crewai.com/concepts/tasks#yaml-configuration-recommended
 agents_config = 'config/agents.yaml'
 tasks_config = 'config/tasks.yaml'

 # If you would like to add tools to your agents, you can learn more about it here:
 # https://docs.crewai.com/concepts/agents#agent-tools
 @agent
 def researcher(self) -> Agent:
  return Agent(
   config=self.agents_config['researcher'],
  )

 @agent
 def reporting_analyst(self) -> Agent:
  return Agent(
   config=self.agents_config['reporting_analyst'],
  )

 # To learn more about structured task outputs, 
 # task dependencies, and task callbacks, check out the documentation:
 # https://docs.crewai.com/concepts/tasks#overview-of-a-task
 @task
 def research_task(self) -> Task:
  return Task(
   config=self.tasks_config['research_task'],
  )

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

 @crew
 def crew(self) -> Crew:
  """Creates the UrlInsightBot crew"""
  # To learn how to add knowledge sources to your crew, check out the documentation:
  # https://docs.crewai.com/concepts/knowledge#what-is-knowledge

  return Crew(
   agents=self.agents, # Automatically created by the @agent decorator
   tasks=self.tasks, # Automatically created by the @task decorator
   process=Process.sequential,
  )

在您克隆了 repository 并设置项目后，您可以使用以下命令运行项目，并使用 curl 进行调用。

cd ./crewai/url_insight_api
uvicorn app.main:app --reload --port 8000 --log-config config/log_config.yaml

测试服务器

## Start analysis
curl -X POST http://localhost:8000/api/v1/analyze \
  -H "Content-Type: application/json" \
  -d '{"topic": "AI LLMs"}'

## Check task status (replace <task_id> with actual ID from previous response)
curl http://localhost:8000/api/v1/task/<task_id>

您可以进行并发请求，您会看到它支持并发。

结论

将 CrewAI 与 FastAPI 集成展示了将 AI 驱动的代理协作与强大的 Python Web 框架相结合的力量，以创建高效、可扩展和智能的应用程序。通过利用 CrewAI 的能力，开发人员可以无缝管理流程，将任务委派给 AI 代理，并实现优化的结果。使用 FastAPI 确保系统不仅快速可靠，而且高度可扩展，使其适用于自动化操作、提供副驾驶协助或简化复杂工作流程等实际应用。

随着 AI 代理的不断发展，像 CrewAI 这样的框架将成为构建模拟人类协作和决策的创新系统的必备工具。通过这种展示的方法，开发人员现在拥有了利用这一潜力的蓝图，提供更智能的应用程序，将生产力和效率提升到新的高度。

🔥 喜欢这篇文章吗？别忘了 点赞、关注 和 分享 给你的朋友们！你的支持帮助我们创作更多这样的内容。如果你想阅读更多类似的文章，可以考虑在 这里 订阅。

🌟 在 Ko-Fi 上支持我们：如果你觉得这篇文章对你有帮助，可以考虑在 Ko-Fi 上请我们喝咖啡。你的支持对我们意义重大，帮助我们持续提供内容！

继续阅读

如果您喜欢这篇文章，您可能会发现以下内容同样有启发性：

  1. 超级增强您的 Supabase：如何像专业人士一样实现数据库钩子

阅读这篇文章时，您会对数据库网络钩子感到惊讶。

2. 揭开 Spring Security 的一次性令牌：无状态认证的终极解决方案

探索 Spring Security 一次性令牌及其强大之处，以及与传统令牌方法的区别。

3. 改变您开发方式的 Monorepo 设置技巧！💥

通过这些简单的 Monorepo 技巧来转变您的开发工作流程——您将惊讶于节省了多少时间！

Tags :
Share :

Related Posts

使用 ChatGPT 搜索网络功能的 10 种创意方法

使用 ChatGPT 搜索网络功能的 10 种创意方法

例如，提示和输出 你知道可以使用 ChatGPT 的“搜索网络”功能来完成许多任务，而不仅仅是基本的网络搜索吗？ 对于那些不知道的人，ChatGPT 新的“搜索网络”功能提供实时信息。 截至撰写此帖时，该功能仅对使用 ChatGPT 4o 和 4o-mini 的付费会员开放。 ![](https://images.weserv.nl/?url=https://cdn-im

阅读更多
在人工智能和技术领域保持领先地位的 10 项必学技能 📚

在人工智能和技术领域保持领先地位的 10 项必学技能 📚

在人工智能和科技这样一个动态的行业中，保持领先意味着不断提升你的技能。无论你是希望深入了解人工智能模型性能、掌握数据分析，还是希望通过人工智能转变传统领域如法律，这些课程都是你成功的捷径。以下是一个精心策划的高价值课程列表，可以助力你的职业发展，并让你始终处于创新的前沿。 1. 生成性人工智能简介课程: [生成性人工智能简介](https://genai.works

阅读更多
10 个强大的 Perplexity AI 提示，让您的营销任务自动化

10 个强大的 Perplexity AI 提示，让您的营销任务自动化

在当今快速变化的数字世界中，营销人员总是在寻找更智能的方法来简化他们的工作。想象一下，有一个个人助理可以为您创建受众档案，建议营销策略，甚至为您撰写广告文案。这听起来像是一个梦想？ 多亏了像 Perplexity 这样的 AI 工具，这个梦想现在成为现实。通过正确的提示，您可以将 AI 转变为您的 个人营销助理。在本文中，我将分享 10 个强大的提示，帮助您自动

阅读更多
10+ 面向 UI/UX 设计师的顶级 ChatGPT 提示

10+ 面向 UI/UX 设计师的顶级 ChatGPT 提示

人工智能技术，如机器学习、自然语言处理和数据分析，正在重新定义传统设计方法。从自动化重复任务到实现个性化用户体验，人工智能使设计师能够更加专注于战略思维和创造力。随着这一趋势的不断增长，UI/UX 设计师越来越多地采用 AI 驱动的工具来促进他们的工作。利用人工智能不仅能提供基于数据的洞察，还为满足多样化用户需求的创新设计解决方案开辟了机会。 1. 用户角色开发 目的

阅读更多
在几分钟内完成数月工作的 100 种人工智能工具

在几分钟内完成数月工作的 100 种人工智能工具

人工智能（AI）的快速发展改变了企业的运作方式，使人们能够在短短几分钟内完成曾经需要几周或几个月的任务。从内容创作到网站设计，AI工具帮助专业人士节省时间，提高生产力，专注于创造力。以下是按功能分类的100个AI工具的全面列表，以及它们在现实世界中的使用实例。 1. 研究工具 研究可能耗时，但人工智能工具使查找、分析和组织数据变得更加容易。**ChatGPT, Cop

阅读更多
你从未知道的 17 个令人惊叹的 GitHub 仓库

你从未知道的 17 个令人惊叹的 GitHub 仓库

Github 隐藏的宝石!! 立即收藏的代码库 学习编程相对简单，但掌握编写更好代码的艺术要困难得多。GitHub 是开发者的宝藏，那里“金子”是其他人分享的精心编写的代码。通过探索 GitHub，您可以发现如何编写更清晰的代码，理解高质量代码的样子，并学习成为更熟练开发者的基本步骤。 1. notwaldorf/emoji-translate *谁需

阅读更多