使用 Amazon Bedrock，Claude3 和 CrewAI 构建应用商城用户评论分析工具

1. 架构介绍

该用户评论分析工具的整体架构图如下：

1.1 Amazon Bedrock 和 Claude3

Claude3 是一种新型的大型语言模型，由 Anthropic 公司开发，现在 Amazon Bedrock 上已经提供 Claude3 的 Sonnet，Haiku 及 Opus 模型，本文将使用 Sonnet 模型。与其他语言模型相比，Claude3 在以下几个方面表现出色：更强的推理和分析能力、更好的常识理解、更高的鲁棒性和一致性、更注重安全性和可控性。

1.2 CrewAI Agent 框架介绍

CrewAI（ https://github.com/joaomdmoura/crewai/ ）是基于 LangChain 开发的一个开源智能体框架,旨在编排和协调多个自主 AI 智能体的团队合作。它允许你创建和管理一组 AI 助手，共同努力实现统一目标，就像船上的船员或项目中的工作团队一样。 CrewAI 的核心思想是创建一个协作环境，让智能体能够共同努力，发挥各自的长处，并根据需要利用外部工具和资源。这种方法旨在克服单个 LLM 的局限性，实现更复杂和先进的问题解决能力。

CrewAI 的一些关键特点包括：

专注于协作：不同于许多专注于单一智能体的 AI 框架，CrewAI 被设计为智能体之间可以无缝协作，共享信息和任务，以获得更好的结果。这种“协作智能”使 CrewAI 能够应对单个智能体难以完成的复杂问题。

采用角色扮演智能体：CrewAI 团队中的每个智能体都可以扮演特定角色,如数据工程师、营销人员或客户服务代表。这使你能够根据项目需求量身定制团队。

简单灵活：CrewAI 被设计为易于使用，即使对于不熟悉 AI 的人也是如此。它也非常灵活，你可以根据具体需求进行定制。例如，每个智能体都可以使用不同的大语言模型，以满足其不同的角色和要求。

CrewAI 的核心组件包括：

智能体（Agents）：智能体是 LLM 的单个实例，每个智能体都具有独特的能力和专长。它们可以被视为虚拟团队成员，各自贡献自己的专业知识来完成整体任务。

任务（Tasks）：任务代表需要解决的具体目标或问题。它们可以是简单的查询，也可以是需要多个智能体协同努力的复杂多步骤过程。

工具（Tools）：工具是智能体可以利用的专门软件组件或外部资源，用于增强其能力。例如搜索引擎、数据库、API 和特定领域的软件工具等。

流程（Processes）：流程定义了解决任务的工作流程和策略。它们概述了步骤顺序、参与的智能体、要使用的工具以及智能体之间的通信和协调机制。

团队（Crews）：团队由针对特定任务组建的多个智能体组成。团队成员具有互补的技能和专长，根据任务需求进行精心挑选。

记忆（Memory）：记忆是一个共享的知识库，智能体可以在任务执行过程中访问和贡献信息。它作为存储和检索相关信息的中央存储库，确保智能体可以访问必要的上下文，并在彼此的工作基础上继续前进。

2. 部署项目

本文源代码已经发布在（代码地址），可以在亚马逊云上的 EC2 进行部署，创建一台 m5.xlarge 机型的 EC2 实例，执行以下安装部署步骤：

#克隆代码仓库
git clone https://github.com/stevensu1977/appstore-review-llm.git
#安装及激活python vitual environment环境（目的是多python/lib依赖版本在EC2上的隔离）
python -m venv .venv
. .venv/bin/activate
pip install -r requirements.txt
#创建需要的目录
mkdir output upload
#该评论分析工具的命令行参数如下
python appreviews.py -h
usage: appreviews.py [-h] [-i] [app_name]
  app_name：要抓取用户评论的app应用，如TikTok(抖音）
  interaction：启用交互式方式，用户可以输入app，并显示执行进度
#执行命令行启动工具应用
python appreviews.py --app_name TikTok
#交互式执行工具应用
python appreviews.py -i 
#启动图形界面
streamlit run main.py
       3. 用户评论分析 
       本文测试的数据都是公开可以访问到的评论数据，读者可以根据需要自行选择，我们以亚马逊云控制台应用为例进行测试。首先启动图形界面“streamlit run main”，streamlit 将在 localhost:8501 上面启动界面，使用浏览器访问即可看见下面的界面。 
       我们选择 Google Play，App 名字输入亚马逊云控制台的应用名字：AWS Console，Rank 选择选择 All（Rank 代表评论的几星，1～5 星，或者 All 全部），点击 Submit，稍后在右侧菜单会出现 Processing，表示 CrewAI Agent 已经开始进行分析，详细 CrewAI 分析的过程见章节 4。 
       Agent 调用 Tools 工具加载数据： 
        定义分析目标和指标，主要是 Google Play 和 Apple Store 的 app 用户评论、自定义数据加载。 
        使用智能体（Agent）调用工具函数（Tools）分别完成数据加载，数据总结。 
       为了实现分析流程，我们分别编写了工具函数（Tools），智能体（Agents），任务（Task）。代码主要分为 main.py（streamlit UI 实现），appreviews.py（Tools、Agents、Task）实现。 
       4.1 CrewAI 与 Amazon Bedrock Claude3 集成 
       CrewAI 直接使用 LangChain 的 LLM 包装实现类，直接使用 BedrockChat，我们通过 initialize_llm 来初始化 Amazon Bedrock 客户端，请注意这里默认使用的是俄勒冈（us-west-2）。 
        from langchain_community.chat_models import BedrockChat
#Config Amazon Bedrock claude3
def initialize_llm():
    Initialize a Large Language Model (LLM) instance using the Bedrock Runtime service.
    Returns:
        BedrockChat: An instance of the BedrockChat class, which represents the initialized LLM.
    model_id = "anthropic.claude-3-sonnet-20240229-v1:0"
    model_kwargs = {
        "max_tokens": 2048,
        "temperature": 0.0,
        "top_k": 250,
        "top_p": 1,
        "stop_sequences": ["\n\nHuman"],
    # Check if AWS_REGION environment variable is set
    aws_region = os.environ.get("AWS_REGION", "us-west-2")
    bedrock_runtime = boto3.client(
        service_name="bedrock-runtime",
        region_name=aws_region,
    bedrock_llm = BedrockChat(
        client=bedrock_runtime,
        model_id=model_id,
        model_kwargs=model_kwargs,
    return bedrock_llm
bedrock_llm = initialize_llm() #后面会直接把这个bedrock_llm作为参数传递给CrewAI的Agents。
       4.2 Tools 
       在初始化后，我们需要编写一些工具函数（Tools），让 Agents 可以访问网络、本地文件系统。我们分别编写了 4 个工具函数，1 个普通函数。CrewAI 可以直接使用 langchain.agents 包中的 tool 修饰器来包装一个函数，需要注意的是这个函数必须带有功能描述，比如”””Tool to found Apple App Store App reviews by app name ,country and rank””” ，这样 Claude3 才会理解函数的作用，以及需要哪些参数，Claude3 会返回正确的函数调用（Function Calling）。 
        from langchain.agents import tool 
......
@tool
def get_apple_app_review(app_name:str, country:str, rank:int = -1):
    """Tool to found Apple App Store App reviews by app name ,country and rank"""
        app = AppStore(country=country, app_name=app_name)
        app.review(how_many=100)
        result = app.reviews
        return result
    except Exception as e:
        print(e)
        return []
@tool
def load_local_file(file_name: str,page:int=1):
    """Tool to load local file , page must start 1"""
        with open(file_name, "rb") as file:
            result = chardet.detect(file.read())
            encoding = result["encoding"]
        file_extension = file_name.split(".")[-1].lower()
        if file_extension == "csv":
            with open(file_name, "r", encoding=encoding) as file:
                reader = csv.reader(file)
                data = list(reader)
            return split_data(data,page)
        elif file_extension == "json":
            with open(file_name, "r", encoding=encoding) as file:
                data = json.load(file)
            return split_data(data,page)
        elif file_extension == "txt":
            with open(file_name, "r", encoding=encoding) as file:
                data = file.read().splitlines()
            return data
        else:
            print(f"Unsupported file extension: {file_extension}")
            return []
    except Exception as e:
        print(f"Error loading file: {e}")
        return []
# Tool to found Google App Store AppID by app name   
@tool
def get_google_app_id(app_name:str):
    """Tool to found Google App Store AppID by app name"""
        result = search(
            app_name,
            lang="en",  # defaults to 'en'
            country="us",  # defaults to 'us'
            n_hits=1  # defaults to 30 (= Google's maximum)
        if len(result)>0:
            return result[0]['appId']
        else:
            return None
    except Exception as e:
        print(e)
        return None
# Tool to found Google App Store App reviews by app id
@tool
def get_google_play_app_review(app_id:str,country:str, rank:int = -1 ):
    """Tool to found Google App Store App reviews by app id ,country and rank"""
        filter_score = None
        if rank > 0 and rank <6:
            filter_score = rank
        print(f"{country=}")
        result, continuation_token = reviews(
        app_id,
        lang='en', # defaults to 'en'
        country="us" if country=="" else country , # defaults to 'us'
        sort=Sort.NEWEST, # defaults to Sort.NEWEST
        count=100, # defaults to 100
        filter_score_with= filter_score
        if continuation_token:
            print(f'Have continuation token: {continuation_token}')
        # if len(result) > 0:
        #     process_app_review_with_llm(result)
        with open(f"output/{app_id}.json", "w", encoding="utf-8") as file:
            json.dump(result, file, cls=DateTimeEncoder,ensure_ascii=False, indent=4)
        return [{"username":review["userName"],"content":review["content"],"score":review["score"]} for review in result]
    except Exception as e:
        print(e)
        return []
       4.3 Agents 
       我们编写了 2 个 Agent，review_loader 和 researcher，分别负责了数据加载预处理和评论总结。其中 role, goal, backstory 都会复制到最终的提示词，提示词模版可以参见 src/crewai/translations/en.json。 
        #data loader agent
review_loader = Agent(
    role='App Review Research Analyst',
    goal='If user input is app id and store type google or apple ,you load App review  from a app id',
    backstory='Specializes in handling and interpreting app reviews documents',
    verbose=True,
    tools=[get_google_play_app_review,get_apple_app_review,load_local_file],
    allow_delegation=False,
    llm=bedrock_llm,
    memory=True,
#summarize agent
researcher = Agent(
  role='Expert App Store reviews summarizer',
  goal='Found prices issue , network issue, game content issue in every user\'s reviews content, you need keep username',
  backstory="""You are a renowned expert summarizer of AI related 
  Google App Store User reviews, known for your insightful 
  and engaging articles. You transform complex concepts into 
  compelling narratives.""",
  verbose=True,
  allow_delegation=True,
  llm=bedrock_llm
       4.4  Tasks、Crew 组装 
       编写完 Agents 后，我们还需要编写 Task，同时需要把 Task 和 Agent 关联起来，最后 Agents、Tasks 组装到 Crew 类中，通过 kickoff 来调用。 
        review_loading_task = Task(
        description=f"""Use store: {store} , app_id: {app_id} and country:{country}, rank: {rank}, if store is Custom File , please load local "upload/{file}", if result more than 50, you need every time process 50 result, return reviews""",
        agent=review_loader,
        expected_output='A refined finalized version of the blog post in markdown format'
content_creation_task = Task(
    description="""Using the insights provided by the research Analyst agent, if have different score , please calculate every score the percentage
    develop an engaging  content, you need use markdown format and translate to chinese. """,
    agent=researcher,
    expected_output=f'A refined finalized version of the blog post in markdown format,need include store: {store} , app_id: {app_id} and country:{country},translate to chinese.'
# Instantiate a Crew with the review_reader and researcher agents, and the tasks to be performed
# Set the logging level to 2 (more verbose)
crew = Crew(
        agents=[review_loader, researcher],
        tasks=[review_loading_task, content_creation_task],
        verbose=2, 
......
#最终执行是在main.py里面调用kickoff()
crew_result = crew.kickoff()
       5. 总结 
       在本文中，我们探讨了如何利用 Amazon Bedrock、Claude3 大型语言模型和 CrewAI 智能体框架来构建一个应用商店评论分析系统。我们构建了一个端到端的分析流程，包括数据加载、关键信息提取、问题分析和报告生成等步骤。通过结合 Claude3 的强大自然语言处理能力和 CrewAI 智能体的协作能力，我们能够高效地处理大量的应用商店评论数据。系统可以自动提取关键信息、分析用户情感倾向，并生成总结性报告。我们使用了亚马逊云控制台作为示例，分别分析了来自 Google Play 和 Apple App Store 的用户评论，通过对不同评分等级的数据进行分析，系统可以关注不同群体的反馈，从而获得更全面的用户洞见。总地来说，将 Amazon Bedrock、Claude3 和 CrewAI 结合使用，为构建智能化的应用商店评论分析系统提供了一种有前景的解决方案。未来，我们计划进一步扩展和优化这一系统，以支持更多功能和更大规模的数据处理需求。 
       *前述特定亚马逊云科技生成式人工智能相关的服务仅在亚马逊云科技海外区域可用，亚马逊云科技中国仅为帮助您了解行业前沿技术和发展海外业务选择推介该服务。 
        Anthropic’s Claude on Amazon Bedrock 
        CrewAI 
        Streamlit