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

#克隆代码仓库
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

python appreviews.py -i

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。

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"""
    try:
        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"""
    try:
        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"""
    try:
        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"""
    try:
        
        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 []

#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
  
)

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()