实验 5与多智能体聊天CrewAIChatGPT 和 Streamlit

与多智能体聊天就像您想象的那么简单

“与多智能体聊天” 是“与一切聊天”系列中的一篇文章。本部分将重点介绍如何应用 CrewAI 平台构建多智能体应用程序

对于那些不知道的人，“与一切聊天”系列专注于为您提供构建 LLM 应用程序的技术知识和技巧。我创建的所有应用程序都是使用流行框架：Streamlit、Langchain 和 OpenAI（LLM 模型）。

您可以在这里找到“与一切聊天”系列：我的 GitHub

难度等级：中级 🎖🎖️️

基于LLM的代理的快速出现和采用使2024年成为人工智能演变的重要转折点。这些代理旨在自主和协作地执行任务，改变了企业和个人解决问题和提高生产力的方式。

随着这些代理变得越来越可靠，并继续为未来的业务建立坚实的基础层，它们的协作将创造一个强大的层次，帮助弥合技术与业务之间的差距。

我将在另一篇文章中深入探讨这个主题。在这篇文章中，我们将重点关注如何通过将概念应用于构建聊天应用程序来实际运作。

我使用的技术栈是CrewAI、Streamlit、Langchain和GPT作为工作模型。

我将在本文中涵盖的内容：

1. 什么是基于LLM的代理
2. CrewAI：多代理平台
3. “与多代理聊天”应用
4. 实施
5. 结果与未来工作

展示案例（对于那些想先看到结果以激励自己的读者）

1. 什么是基于LLM的代理

什么是代理？

在人工智能和自动化系统领域，代理是能够自主和独立地执行特定任务的软件实体。这些代理被设计为根据预定义的规则和目标进行操作，与环境和其他代理互动，以高效地完成任务。

代理的关键特征

为了帮助您理解代理的特征，我在下面的图片中进行了说明：

1. 目标导向： 您必须记住的重要一点是，每个代理都是为了实现一个或多个特定目标而设计的。每个代理都有特定的目标来指导其行动和决策。代理可以根据重要性和紧急性对任务进行优先级排序。
2. 适应性： 代理具有适应变化情况和环境的能力。它们可以根据来自环境的反馈（例如文本、音频、视频等）调整其行为。代理可以根据新信息和变化的条件改变其操作。高级代理还可以通过经验学习来提高性能。
3. 互动性： 代理可以与周围环境和其他代理进行互动。这使它们能够协调并共同工作以实现共同目标。代理使用通信协议与其他代理交换信息，而无需人类干预。
4. 反应性： 代理可以通过使用工具（例如搜索互联网、抓取网站、使用代码解释器总结下载内容等）采取主动措施来完成这些任务。这主要是通过 API 调用和函数调用来实现的。
5. 最后但同样重要的是，代理的大脑使代理能够自主工作，基于 LLM。基于此，代理可以在没有持续人类干预的情况下做出自己的决策。我们在这里有很多选择，例如 GPT、Claude、Llama 等。

2. CrewAI: 多智能体平台

为了演示智能体如何协同工作，我将使用CrewI，这是一个旨在优化智能体工作流程的平台。如果您想了解更多细节，请 点击此链接

本节将带您了解CrewAI的一些核心概念。首先，让我们看一个例子：

示例

目标：“从用户输入的个人传记信息中提取三个关键点。”

实现代码创建了两个代理协同工作，以实现“从用户输入的个人传记信息中提取三个关键点”的目标（在本例中为埃隆·马斯克）：

• 搜索代理： 配备使用“Serper API”工具从互联网搜索信息的能力
• 摘要代理： 具备从收集到的信息中总结和提取3个关键点的能力（搜索任务的输出）

目标被分为两个任务：搜索任务和摘要任务

实现代码

from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool
import os

# 在此处替换 OPENAI_API_KEY & SERPER_API_KEY
# 获取 OPENAI_API_KEY 的链接: https://platform.openai.com/api-keys
# 获取 SERPER_API_KEY 的链接: https://serper.dev/api-key
os.environ["OPENAI_API_KEY"] = "sk-xxxxxx"
os.environ["OPENAI_MODEL_NAME"] = "gpt-3.5-turbo"
os.environ["SERPER_API_KEY"] = "799fab83yyyyy"

# 定义一个工具
search_tool = SerperDevTool()

# 定义一个搜索代理
search_agent = Agent(
    role = "搜索代理",
    goal = "谁是 {input}",
    backstory = """你是一个在互联网上搜索传记的专家""",
    tools = [search_tool],
)

# 定义一个摘要代理
# 创建代理时，必须为其设置目标和背景故事。
summary_agent = Agent(
    role = "传记摘要代理",
    goal="在200字以内总结任何传记",
    backstory="你是一个从传记中提取重要和简明信息的专家",
)

# 定义一个搜索任务，以查找用户输入的 {input} 的传记
search_task = Task(
    description = "搜索 {input} 的传记",
    expected_output = "完整的传记",
    agent = search_agent,
    tools = [search_tool],
)

# 定义一个摘要任务，从 search_task 的输出中提取关键信息
summary_task = Task(
    description = "仅提取传记中的3个亮点",
    expected_output="包含3个亮点的简短传记",
    agent = summary_agent,
    context = [search_task]
)

# 定义包含代理和任务的团队
# Process.sequential: 代理按特定顺序执行任务。
my_crew = Crew(
    agents = [search_agent, summary_agent],
    tasks = [search_task, summary_task],
    process = Process.sequential,
    verbose = True
)

my_crew.kickoff(inputs={"input":"Elon musk"})

运行上述代码并获得结果：

Here are three highlights from Elon Musk's biography:

1. Elon Musk was born in Pretoria, South Africa, to model Maye and businessman and engineer Errol Musk. He briefly attended the University of Pretoria before immigrating to Canada.

2. Musk founded X.com in 1999, which later became PayPal, SpaceX in 2002, and Tesla. He is widely known for his work with electric vehicles and space exploration.

3. Elon Musk is a creative genius known for his tenacious resilience. Despite his extraordinary success, he remains human enough to allow others to see him at his most vulnerable.

3. “多智能体聊天”应用

到目前为止，我假设您已经理解了所有内容。现在，我们将进入一个更复杂的示例：“多智能体聊天”应用。我们应用的高层设计如下面的图像所示：

该应用的目标是“创建关于用户输入的{topic}的新闻通讯。”

• search_agent 将使用“Serper API工具”搜索并返回关于{topic}的5个URL。
• download_agent 将使用“下载”工具下载上述每个URL的内容。
• summary_agent 将使用“创建新闻通讯”工具从文章摘要列表和URL列表中创建新闻通讯。

4. 实施

我将带您了解我们代码的主要部分。

4.1. 工具

首先，我们需要创建3个技能（或工具）

搜索工具： 我使用了Serper API作为第一个工具。这个工具的目的是搜索并返回任何主题的5个结果

import json
import os

import requests
from langchain.tools import tool

os.environ["SERPER_API_KEY"] = "799fab83yyyyy"

class SearchTools():

    @tool("search_internet")
    def search_internet(query):
        """用于搜索互联网
        关于给定主题并返回相关结果"""
        print(f"DEBUG:SearchTools:{query}")
        top_result_to_return = 5

        try:
            url = "https://google.serper.dev/search"
            payload = json.dumps({"q": query})
            headers = {
                'X-API-KEY': os.environ['SERPER_API_KEY'],
                'content-type': 'application/json'
            }
            response = requests.request("POST", url, headers=headers, data=payload)
            # 检查是否有organic键
            if 'organic' not in response.json():
                return "抱歉，我找不到关于该主题的任何信息，可能是您的serper api密钥有误。"
            else:
                results = response.json()['organic']
                string = []
                for result in results[:top_result_to_return]:
                    try:
                        string.append('\n'.join([
                            f"标题: {result['title']}", f"链接: {result['link']}",
                            f"摘要: {result['snippet']}", "\n-----------------"
                        ]))
                    except KeyError:
                        next

                return '\n'.join(string)
        except Exception as e:
            return f"SearchTools:Exception:{e}"

内容下载工具： 我使用了newspaper4k库作为第二个工具。这个工具的目的是下载并总结下载的内容，生成150字的文本。

from crewai import Agent, Task, Crew, Process
import re
from langchain.tools import tool
import newspaper

class BrowserTools():    

    @tool("using_newspaper4k_scrape_and_summarize_website")
    def using_newspaper4k_scrape_and_summarize_website(website):
        """用于抓取和总结网站内容"""
        print(f"DEBUG:BrowserTools:{type(website)}:URL:{website}")
        try:
            link = ""
            if isinstance(website, dict):
                # 检查并从网站获取链接
                link = website.get("website")["title"]
            else:
                # website是字符串
                pattern = r'"website":\s*"([^"]+)"'
                match = re.search(pattern, website)

                if match:
                    link = match.group(1)
                else:
                    url_pattern = r'https?://[^\s<>"]+|www\.[^\s<>"]+'
                    url_match = re.match(url_pattern, website)
                    if url_match:
                        link = website
                    else:
                        print("未找到链接")
            print(f"DEBUG:URL:{link}")

            article = newspaper.article(link)
            content = f"标题: {article.title}. 内容: {article.text}"

            summary_agent = Agent(
                role='摘要代理',
                goal='在150字以内总结以下内容: {content}',
                backstory="您是一位著名CEO的助手",
                allow_delegation=False,
            )

            summary_task = Task(
                description="在150字以内总结以下内容: {content}",
                expected_output="一个摘要",
                agent=summary_agent,
            )

            crew = Crew(
                agents=[summary_agent],
                tasks=[summary_task],
            )
            result = crew.kickoff(inputs={"content": content})
            return result
        except Exception as e:
            return f"BrowserTools:Exception:{e}"

摘要工具： 这个工具的目的是汇总所有摘要内容

from langchain.tools import tool

class NewsletterTools():

    @tool("create_newsletter")
    def create_newsletter(summaries):
        """
        在创建汇总所有摘要内容的新闻通讯时非常有用
        """
        print(f"DEBUG:NewsletterTools:{summaries}")
        try:
            newsletter = ""
            for summary in summaries:
                # 假设每个摘要都包含'title'和'content'
                title = summary['title']
                content = summary['description'][:150]  # 摘要不超过150字
                newsletter += f"标题: {title}\n内容: {content}\n\n"
            return newsletter
        except Exception as e:
            return f"NewsletterTools:Exception:{e}"

4.2. 代理

这很简单，对吧！

• search_agent 将使用“搜索工具”搜索并返回 5 个 URL
• download_agent 将使用“下载”工具下载上述每个 URL 的内容。
• summary_agent 将根据文章摘要列表和 URL 列表创建新闻通讯

# Define Search Agent
search_agent = Agent(
    role='Search Agent',
    goal="Search for the latest news about the topic {topic}",
    backstory="You are an expert at searching for information on the internet and always keep up with the latest news.",
    memory = True,
    verbose = True,
    callbacks=[MyCustomHandler("SearchAgent")],
    tools = [SearchTools.search_internet]
)

# Define Download Agent
download_agent = Agent(
    role='Download Agent',
    goal="Download and summarize the main content from the list of URL",
    backstory="You are an expert at downloading and summarizing content from articles on the internet.",
    memory=True,
    verbose=True,
    callbacks=[MyCustomHandler("DownloadAgent")],
    tools = [BrowserTools.using_newspaper4k_scrape_and_summarize_website]
)

# Define Newsletter Agent
newsletter_agent = Agent(
    role='Newsletter Agent',
    goal='Create a newsletter aggregating news from a list of article summaries',
    backstory='You are an expert at aggregating news and creating engaging and easy-to-read newsletters.',
    callbacks=[MyCustomHandler("NewsletterAgent")],
    memory=True,
    verbose=True,
    tools = [NewsletterTools.create_newsletter]
)

4.3. 任务

我们将目标分解为 3 个任务：

# search_task: search for topic via internet
search_task = Task(
    description=(
        "Search and return a list of URLs related to the topic: {topic}."
    ),
    expected_output='List of URLs.',
    agent=search_agent,
)

# download_task: download the content from each received URL
download_task = Task(
    description=(
        "Download content from each URL in the list and summarize the main content of each URL"
    ),
    expected_output='A summary of the main content of URL',
    agent=download_agent,
    context = [search_task]
)

# create_newsletter_task: aggregating the summary results from download_task
create_newsletter_task = Task(
    description=(
        "Create a newsletter from a list of article summaries and the URL list"
    ),
    expected_output='A newsletter aggregating articles including a title and brief description.',
    context = [search_task, download_task],
    agent=newsletter_agent,
)

4.4. 如何在 Streamlit 上展示我们的结果

我们如何在 Streamlit 上展示我们的结果？

# Avatar Photos for our bots
avatars = {"SearchAgent": "https://cdn-icons-png.flaticon.com/512/10885/10885144.png",
           "DownloadAgent": "https://cdn-icons-png.flaticon.com/512/4021/4021729.png",
           "NewsletterAgent": "https://cdn-icons-png.flaticon.com/512/5822/5822082.png"}

# Init Session State
if "messages" not in st.session_state:
    st.session_state["messages"] = [{"role": "assistant", "content": "您感兴趣的话题是什么？"}]


# Handle responses from CrewAI and show it on streamlit chat_message
class MyCustomHandler(BaseCallbackHandler):
    def __init__(self, agent_name: str) -> None:
        self.agent_name = agent_name

    def on_chain_start(self, serialized: Dict[str, Any], inputs: Dict[str, Any], **kwargs: Any) -> None:
        """打印出我们正在进入一个链。
        如果觉得太吵，可以关闭它
        """
        # content = "DEBUG: Show you behind stories..:" + inputs['input']
        # st.session_state.messages.append({"role": "assistant", "content": content})
        # st.chat_message("assistant").write(content)

    def on_chain_end(self, outputs: Dict[str, Any], **kwargs: Any) -> None:
        """打印出我们完成了一个链。"""
        st.session_state.messages.append({"role": self.agent_name, "content": outputs['output']})
        st.chat_message(self.agent_name, avatar=avatars[self.agent_name]).write(outputs['output'])

主页：

def main_page():
    st.title("💬 CrewAI: 创建新闻通讯")

    agents = [search_agent, download_agent, newsletter_agent]
    tasks = [search_task, download_task, create_newsletter_task]

    for msg in st.session_state.messages:
        if msg["role"] in avatars.keys():
            st.chat_message(msg["role"], avatar=avatars[msg["role"]]).write(msg["content"])
        else:
            st.chat_message(msg["role"]).write(msg["content"])

    if prompt := st.chat_input():
        st.session_state.messages.append({"role": "user", "content": prompt})
        st.chat_message("user").write(prompt)

        crew = Crew(
            agents=agents,
            tasks=tasks,
            process=Process.sequential,
            manager_llm=llm,
            output_log_file="crewai.log",
        )

        final = crew.kickoff(inputs={"topic": prompt})


if __name__ == '__main__':
    main_page()

5. 结果与未来工作

运行上述代码并获得结果：

search_agent 返回的结果：

download_agent 返回的结果：

最后，我们的结果（来自 newsletter_agent）：

未来工作

我希望你到目前为止享受这个旅程。还有许多想法可以考虑并与 CrewAI 实现。

以下是其中之一：

加密分析

以下是一些关于如何使用 CrewAI 构建加密交易数据分析应用程序的想法，包括代理和任务定义。

代理想法

1) 市场数据收集器

• 角色：从各种来源收集市场数据。
• 目标：收集加密交易所的价格、交易量和技术指标。
• 背景故事：在加密市场方面的专家，能够快速收集和处理数据。

2) 新闻分析师

• 角色：分析与加密相关的新闻。
• 目标：总结与加密相关的新闻文章、新闻稿和市场分析。
• 背景故事：一位新闻专家，能够快速准确地分析影响加密市场的新闻。

3) 技术分析师

• 角色：进行技术分析。
• 目标：分析价格图表和技术指标以预测市场趋势。
• 背景故事：一位拥有多年阅读图表和指标经验的技术分析专家。

4) 情绪分析师

• 角色：分析市场情绪。
• 目标：通过社交媒体和论坛分析市场情绪，以评估投资者情绪。
• 背景故事：一位情绪分析专家，能够从社交数据源中读取市场情绪。

5) 投资策略师

• 角色：制定投资策略。
• 目标：结合其他代理的分析，提供全面的投资建议。
• 背景故事：一位投资策略师，能够结合多个来源的信息做出明智的投资决策。

任务想法

收集市场数据

from crewai import Task

market_data_task = Task(
    description="""
    Collect market data from crypto exchanges, including prices, trading volumes, and technical indicators.
    """,
    expected_output="A dataset containing prices, trading volumes, and technical indicators from exchanges.",
    agent=market_data_collector
)

分析加密新闻

from crewai import Task

news_analysis_task = Task(
    description="""
    Summarize news articles, press releases, and market analyses related to crypto.
    """,
    expected_output="A report summarizing the latest news, press releases, and market analyses related to crypto.",
    agent=news_analyzer

)

进行技术分析

from crewai import Task

technical_analysis_task = Task(
    description="""
    Analyze price charts and technical indicators to predict market trends.
    """,
    expected_output="A technical analysis report including price charts and market trend predictions.",
    agent=technical_analyst
)

评估市场情绪

from crewai import Task

sentiment_analysis_task = Task(
    description="""
    Analyze market sentiment through social media and forums to gauge investor sentiment.
    """,
    expected_output="A report analyzing market sentiment from social media and forums.",
    agent=sentiment_analyst
)

制定投资策略：

from crewai import Task

investment_strategy_task = Task(
    description="""
    Combine analyses from other agents to provide comprehensive investment recommendations.
    """,
    expected_output="A comprehensive investment strategy report combining technical analysis, news analysis, and sentiment analysis.",
    agent=investment_strategist,
    context=[market_data_task, news_analysis_task, technical_analysis_task, sentiment_analysis_task]
)

团队

from crewai import Crew

# Create a Crew with the agents and tasks
crypto_analytics_crew = Crew(
    agents=[market_data_collector, news_analyzer, technical_analyst, sentiment_analyst, investment_strategist],
    tasks=[market_data_task, news_analysis_task, technical_analysis_task, sentiment_analysis_task, investment_strategy_task]
)

# Kick off the Crew with specific input information
result = crypto_analytics_crew.kickoff(inputs={"input": "Bitcoin"})
print(result)

在你离开之前！🤟

如果你觉得这篇文章对你有帮助并希望表示支持，请为我的文章鼓掌 10 次。 👏 这真的会激励我，并将这篇文章推荐给更多人。

Stackademic 🎓

感谢您阅读到最后。在您离开之前：

• 请考虑 点赞 和 关注 作者！ 👏
• 关注我们 X | LinkedIn | YouTube | Discord
• 访问我们的其他平台： In Plain English | CoFeed | Differ
• 更多内容请访问 Stackademic.com