使用crewai掌握AI代理创建：构建动态智能代理的逐步指南

Rifx.Online
Natural Language Processing , AI Applications , AI Research
08 Mar, 2025

AI Agent 及其重要性简介

AI Agent 是可以自主或在极少人工干预下执行任务的软件实体。它们被设计用于感知其环境、做出决策并采取行动以实现特定目标。在自然语言处理 (NLP) 的背景下，AI Agent 可用于自动化文本摘要、翻译、情感分析等任务。

在当今的 AI 驱动世界中，AI Agent 的重要性怎么强调都不为过。随着对智能自动化的需求不断增长，各组织越来越多地转向 AI Agent 来简化其运营、提高效率并获得竞争优势。根据 Grand View Research 的一份报告，全球 AI 市场预计将以 36.6% 的复合年增长率从 2023 年增长到 2030 年，而 AI Agent 在这一增长中发挥着重要作用。

了解 CrewAI 及其功能

CrewAI 是一个强大的框架，允许开发人员构建能够执行复杂任务的智能 Agent。它提供了一个灵活的架构，可以创建具有多个工具的 Agent，每个工具都专门用于特定任务。这种模块化方法提供了更大的灵活性和可扩展性，使得为各种应用程序开发和部署 AI Agent 变得更加容易。

CrewAI 的一些主要功能包括：

模块化架构：CrewAI 允许开发人员创建具有多个工具的 Agent，每个工具都为特定任务而设计。这种模块化方法实现了更大的灵活性和可扩展性。
动态任务执行：CrewAI 中的 Agent 可以根据收到的输入动态地执行任务，从而实现更智能和自适应的行为。
与 LLM 集成：CrewAI 支持与大型语言模型 (LLM)（例如 GPT-4）集成，使 Agent 能够利用高级 NLP 模型的强大功能来执行文本生成、摘要等任务。
可定制的工作流程：开发人员可以自定义其 Agent 的工作流程，定义任务的顺序和执行每个任务的条件。

设置环境

在深入构建我们的 AI Agent 之前，我们需要设置我们的环境并安装必要的库。这包括安装必要的库并设置我们的 Azure OpenAI 凭据。

pip install crewai pydantic openai

import os

os.environ["OPENAI_BASE_URL"] = "your_azure_openai_endpoint"
os.environ["OPENAI_API_KEY"] = "your_api_key"
os.environ["OPENAI_API_VERSION"] = "2024-06-01-preview"
os.environ["OPENAI_DEPLOYMENT_NAME"] = "your_gpt4_deployment_name"

设计 Agent 架构

我们的目标是创建一个可以对会议记录进行分类并生成针对特定会议类型量身定制的摘要的 Agent。为了实现这一目标，我们将设计一个具有以下组件的 Agent：

分类工具：此工具将会议记录分类为以下五类之一：scrum、网络研讨会、知识转移 (KT) 会议、研讨会或一般会议。
摘要工具：我们将为每种会议类型创建专用的摘要工具，每个工具都有其自己的提示，该提示是根据该会议类型的特定要求量身定制的。
Agent：Agent 将协调工作流程，使用分类工具确定会议类型，然后选择适当的摘要工具来生成摘要。

实现分类工具

我们工作流程中的第一步是对会议记录进行分类。我们将创建一个 ClassifyMeetingTool，它使用 LLM 分析记录并确定会议类型。

from crewai.tools import BaseTool
from pydantic import BaseModel, Field
from openai import AzureOpenAI

class ClassifyMeetingTool(BaseTool):
    name: str = "ClassifyMeeting"
    description: str = "Identifies the meeting type using AI."

class Input(BaseModel):
        text: str = Field(..., description="Snippet from the meeting transcript.")

def _run(self, text: str) -> str:
        system_prompt = """Analyze the meeting transcript and classify it into one of these categories:
        - scrum: Daily standups, sprint planning, sprint review, or retrospective meetings
        - webinar: Educational presentations, training sessions with a primary speaker
        - kt: Technical knowledge transfer sessions, system documentation reviews
        - workshop: Interactive sessions with hands-on activities
        - general: Any other type of meeting

Return ONLY the category name, nothing else."""

        client = AzureOpenAI(
            azure_endpoint=os.getenv("OPENAI_BASE_URL"),
            api_key=os.getenv("OPENAI_API_KEY"),
            api_version=os.getenv("OPENAI_API_VERSION"),
        )
        messages = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": text[:500]}
        ]
        response = client.chat.completions.create(
            model=os.getenv("OPENAI_DEPLOYMENT_NAME"),
            messages=messages,
        )
        return response.choices[0].message.content.strip().lower()

实现总结工具

接下来，我们将为每种会议类型创建专门的总结工具。每个工具都将拥有针对该会议类型特定需求的提示。

class ScrumSummarizationTool(BaseTool):
    name: str = "ScrumSummarizationTool"
    description: str = "Summarizes scrum meetings with task details."

class Input(BaseModel):
        transcription: str = Field(..., description="Scrum meeting transcript.")

def _run(self, transcription: str) -> str:
        system_prompt = """Summarize the following scrum call, focusing on project updates, tasks, and blockers. Keep the summary concise."""
        return generate_completion(system_prompt, transcription)

class WebinarSummarizationTool(BaseTool):
    name: str = "WebinarSummarizationTool"
    description: str = "Summarizes webinars with key takeaways."

class Input(BaseModel):
        transcription: str = Field(..., description="Webinar meeting transcript.")

def _run(self, transcription: str) -> str:
        system_prompt = """Summarize the following webinar, focusing on key takeaways, Q&A highlights, and notable insights. Keep the summary concise."""
        return generate_completion(system_prompt, transcription)

class KTSummarizationTool(BaseTool):
    name: str = "KTSummarizationTool"
    description: str = "Summarizes KT sessions with technical insights."

class Input(BaseModel):
        transcription: str = Field(..., description="KT session meeting transcript.")

def _run(self, transcription: str) -> str:
        system_prompt = """Summarize the following KT session, focusing on technical concepts, action items, and key recommendations. Keep the summary concise."""
        return generate_completion(system_prompt, transcription)

class WorkshopSummarizationTool(BaseTool):
    name: str = "WorkshopSummarizationTool"
    description: str = "Summarizes workshops with practical takeaways."

class Input(BaseModel):
        transcription: str = Field(..., description="Workshop meeting transcript.")

def _run(self, transcription: str) -> str:
        system_prompt = """Summarize the following workshop, focusing on interactive activities, practical takeaways, and new skills acquired. Keep the summary concise."""
        return generate_completion(system_prompt, transcription)

class GeneralSummarizationTool(BaseTool):
    name: str = "GeneralSummarizationTool"
    description: str = "Summarizes general meetings with discussion points."

class Input(BaseModel):
        transcription: str = Field(..., description="General meeting transcript.")

def _run(self, transcription: str) -> str:
        system_prompt = """Summarize the following meeting, focusing on main discussion points and consensus. Keep the summary concise."""
        return generate_completion(system_prompt, transcription)

实现代理

现在我们有了工具，我们可以创建编排工作流程的代理。

from crewai import Agent
from typing import Optional

class MeetingSummarizationAgent(Agent):
    def __init__(self, tools):
        super().__init__(
            role="AI Meeting Summarizer",
            goal="Generate tailored summaries for different meeting types",
            backstory="This agent uses dedicated tools to tailor summaries dynamically.",
            tools=tools,
            verbose=True,
        )

def generate_summary(self, transcription: str) -> str:
        classifier = self.get_tool_by_name("ClassifyMeeting")
        if not classifier:
            return "No meeting type classifier tool found."

        meeting_type = classifier.run(text=transcription[:500])

        tool_mapping = {
            "scrum": "ScrumSummarizationTool",
            "webinar": "WebinarSummarizationTool",
            "kt": "KTSummarizationTool",
            "workshop": "WorkshopSummarizationTool",
            "general": "GeneralSummarizationTool"
        }

        summarizer_name = tool_mapping.get(meeting_type, "GeneralSummarizationTool")
        summarizer = self.get_tool_by_name(summarizer_name)
        if not summarizer:
            return f"No summarization tool found for {meeting_type}."

        return summarizer.run(transcription=transcription)

测试代理

让我们用一个示例的文字记录来测试我们的代理，您可以传递您的文字记录或提供您的文字记录文件的路径。您也可以根据需要添加尽可能多的工具，目前我将向您展示 2-3 个工具以及一个简单的文字记录。

## Initialize tools
classifier = ClassifyMeetingTool()
scrum_tool = ScrumSummarizationTool()
webinar_tool = WebinarSummarizationTool()
..... add more tools

## Initialize agent
agent = MeetingSummarizationAgent([
    classifier,
    scrum_tool,
    webinar_tool,
    ....
])

## Test with a sample transcript
transcript = "Daily standup meeting: Team updates on sprint tasks and blockers."
summary = agent.generate_summary(transcript)
print(summary)

人工智能代理在人工智能竞赛中的未来

人工智能代理有望在正在进行的人工智能竞赛中发挥关键作用。随着组织继续寻求利用人工智能以获得竞争优势的方式，创建能够自主执行复杂任务的智能代理的能力将变得越来越有价值。根据 MarketsandMarkets 的一份报告，全球医疗保健人工智能市场预计将从 2021 年的 66 亿美元增长到 2026 年的 254 亿美元，在预测期内的复合年增长率为 30.5%。这种增长是由医疗保健领域对人工智能的日益采用所驱动的，这预计将为人工智能代理在医疗诊断、治疗计划和患者监测等领域创造新的机会。