使用Groq和Llama 3构建生成式AI新闻搜索应用程序

Llama 3 在 Groq 上闪耀，带来极速生成

在这篇博客中，我们将创建一个生成式 AI 新闻搜索的后端应用。我们会使用 Meta 的 Llama-3 8B 模型，并通过 Groq 的 LPU 提供服务。

关于 Groq

如果你还不知道 Groq，那就让我来介绍一下。Groq 正在设定大型语言模型文本生成的新推理速度标准。Groq 提供的 LPU（语言处理单元）是一种全新的端到端处理系统，为像大型语言模型这样的带有序列组件的计算密集型应用提供最快的推理速度。

我们不会深入探讨 Groq 相比 GPU 的推理速度有多快，而是希望利用 Groq 和 Llama 3 在文本生成上的优势来创建一个

生成式 AI 新闻搜索应用。这将类似于 Bing AI 搜索、Google AI 搜索或 PPLX。

为什么选择 Llama 3？

Meta 最近发布的 Llama 3 模型获得了巨大成功。70B 的 Llama 3 模型目前在 LMSys LLM 榜单上排名第五。在英语任务上，这个模型排名第二，仅次于 GPT-4。

根据 Meta 的 Llama 3 发布博客，8B 模型在其类别中是最好的，70B 模型也优于 Gemini Pro 1.5 和 Claude 3 Sonnet。

为了展示模型对现实世界问题的理解，Meta 创建了一个高质量的人类评估集。评估集包含 1800 个涵盖 12 个关键用例的提示：

• 咨询建议
• 头脑风暴
• 分类
• 闭合问答
• 编程
• 创意写作
• 提取
• 扮演角色
• 开放问答
• 推理
• 改写
• 总结

为了避免开发团队过度拟合这些数据，评估集一直对他们保密。他们将 Llama 3 70B 与 Claude Sonnet、Mistral Medium、GPT-3.5 和 Llama 2 进行了比较。以下图表显示了 Llama 3 对上述模型的获胜比例。

这些基准测试结果对 Llama 3 非常有利，我们决定用它来进行我们的生成式 AI 新闻搜索。

通常来说，较小的模型推理速度更快，因为它们不占用太多 VRAM，参数计算也较少，因此 Token 生成更快。所以我们可以选择较小的 Llama 3 8B 模型。

让我们来动手编写代码吧。

新闻 API

我们会使用 Newsdata.io 的免费新闻 API，根据搜索查询获取新闻内容。我们也可以使用 Google 的 RSS 源或者其他新闻 API。

你可以通过在 Newsdata 平台注册后获得的 API 令牌访问 Newsdata 新闻 API。一旦获得 API 令牌，只需进行一个带有搜索查询的 GET 请求，获取结果并传递给 LLM。

我们会使用以下代码通过 Newsdata.io API 获取新闻。

# news.py

import os
import httpx
from configs import NEWS_API_KEY, NEWS_BASE_URL

async def getNews(query: str, max_size: int = 8):
    async with httpx.AsyncClient(timeout=60) as client:
        response = await client.get(
            os.path.join(NEWS_BASE_URL, "news") +
            f"?apiKey={NEWS_API_KEY}&q={query}&size={max_size}")
        try:
            response.raise_for_status()
            return response.json()
        except httpx.HTTPStatusError as e:
            print(f"Error response {e.response.status_code} while requesting {e.request.url!r}")
            return None

上面的代码使用了 httpx 库进行异步 API 调用，将 API 令牌和搜索词传入。如果响应状态码是 200，我们会返回响应，否则打印异常并返回 None。

Groq 接口

Groq 提供了一个认证过的 REST API 接口来使用 Llama 3 8B 模型。我们也可以通过官方 Groq Python 库 与 Llama 3 8B 模型交互。

以下是与 Groq 的交互方式。

# llms/groq.py

from groq import Groq, AsyncGroq
import traceback
from typing: List, Dict, Union
from llms.base import BaseLLM
from llms.ctx import ContextManagement
from groq import RateLimitError
import backoff

manageContext = ContextManagement()

class GroqLLM(BaseLLM):
    def __init__(self, api_key: Union[str, None] = None):
        super().__init__(api_key)
        self.client = AsyncGroq(api_key)

    @backoff.on_exception(backoff.expo, RateLimitError, max_tries=3)
    async def __call__(self, model: str, messages: List[Dict], **kwargs):
        try:
            if "system" in kwargs:
                messages = [{"role": "system", "content": kwargs.get("system")}] + messages
                del kwargs["system"]
            messages = manageContext(messages, kwargs.get("ctx_length", 7_000))
            output = await self.client.chat.completions.create(messages=messages, model=model, **kwargs)
            return output.choices[0].message.content
        except RateLimitError:
            raise RateLimitError
        except Exception as err:
            print(f"ERROR: {str(err)}")
            print(f"{traceback.format_exc()}")
            return ""

class GroqLLMStream(BaseLLM):
    def __init__(self, api_key: Union[str, None] = None):
        super().__init__(api_key)
        self.client = AsyncGroq(api_key)
    
    async def __call__(self, model: str, messages: List[Dict], **kwargs):
        if "system" in kwargs:
            messages = [{"role": "system", "content": kwargs.get("system")}] + messages
            del kwargs["system"]
        messages = manageContext(messages, kwargs.get("ctx_length", 7_000))
        if "ctx_length" in kwargs:
            del kwargs["ctx_length"]
        output = await self.client.chat.completions.create(messages=messages, stream=True, model=model, **kwargs)
        async for chunk in output:
            yield chunk.choices[0].delta.content or ""

我通常倾向于继承 BaseLLM 类来管理常用功能。以下是 BaseLLM 的代码。

# llms/base.py

from abc import ABC, abstractmethod
from typing: List, Dict, Union

class BaseLLM(ABC):
    def __init__(self, api_key: Union[str, None] = None, **kwargs):
        self.api_key = api_key
        self.client = None
        self.extra_args = kwargs
    
    @abstractmethod
    async def __call__(self, model: str, messages: List[Dict], **kwargs):
        pass

Llama 3 8B 支持 8192 个 token 作为上下文长度，其中 7000 个用于输入，其余用于输出或生成。

输入上下文可能超过 7000 个 token，但在这种情况下我们需要管理它，以确保有足够的 token 用于输出生成。为此，我们编写了 ContextManagement 实用工具。

# llms/ctx.py

from typing: List, Dict, Literal, Union
from transformers import AutoTokenizer

class ContextManagement:
    def __init__(self):
        self.tokenizer = AutoTokenizer.from_pretrained("meta-llama/Meta-Llama-3-8B")
    
    def __count_tokens__(self, content: str):
        tokens = self.tokenizer.tokenize(content)
        return len(tokens) + 2
    
    def __pad_content__(self, content: str, num_tokens: int):
        return self.tokenizer.decode(self.tokenizer.encode(content, max_length=num_tokens))

    def __call__(self, messages: List[Dict], max_length: 28_000):
        managed_messages = []
        current_length = 0
        current_message_role = None
        for ix, message in enumerate(messages[::-1]):
            content = message.get("content")
            message_tokens = self.__count_tokens__(message.get("content"))
            if ix > 0:
                if current_length + message_tokens >= max_length:
                    tokens_to_keep = max_length - current_length
                    if tokens_to_keep > 0:
                        content = pad_content(content, tokens_to_keep)
                        current_length += tokens_to_keep
                    else:
                        break
                if message.get("role") == current_message_role:
                    managed_messages[-1]["content"] += f"\n\n{content}"
                else:
                    managed_messages.append({"role": message.get("role"), "content": content})
                    current_length += message_tokens
            else:
                if current_length + message_tokens >= max_length:
                    tokens_to_keep = max_length - current length
                    if tokens_to_keep > 0:
                        content = pad_content(content, tokens_to keep)
                        managed_messages.append({"role": message.get("role"), "content": content})
                        current_length += tokens到 keep
                    else:
                        break
                else:
                    managed_messages.append({"role": message.get("role"), "content": content})
                    current_length += message tokens
            current_message_role = message.get("role")
        print(f"TOTAL TOKENS: {current_length}")
        return managed_messages[::-1]

上面的代码使用了 HuggingFace 的 tokenizers 库对消息进行分词并统计 token 数量，保留符合最大 token 长度限制（7000）的消息及其内容。

使用 meta-llama/Meta-Llama-3-8B 分词器之前，我们需要先在 HuggingFace 上提供详细

信息并接受 Meta 的使用条款，然后通过 huggingface-cli login 命令或在 AutoTokenizer.from_pretrained 方法中提供 HuggingFace 令牌。

简单的提示

我们将为生成式 AI 新闻搜索应用使用一个非常简单的提示。提示如下：

# prompts.py

SYSTEM_PROMPT = """你是一个新闻摘要机器人。当用户提供查询时，你会收到几个与查询相关的新闻项。你的任务是评估这些新闻项与查询的相关性并仅保留相关的项。
如果有相关的新闻项，你应该以简洁、专业且尊重的方式总结它们。摘要应以第一人称形式提供，并且必须为新闻文章提供 Markdown 格式的引用。不要告诉用户评估了多少新闻项，只关注提供相关文章的简要摘要。
如果没有找到相关的新闻项，请礼貌地告知用户目前无法提供答案。记住，你的回答应直接满足用户的兴趣，而不透露后台流程或数据检索的具体细节。
例如，如果查询关于“2024 年洛克萨巴选举”，并找到相关文章，请提供这些文章的摘要。如果文章无关或无用，请礼貌地告知用户无法提供所需信息。
"""

这个提示非常直接明了。

智能体

让我们将所有内容整合起来，完成我们的生成式 AI 新闻搜索智能体。

# agent.py

from llms.groq import GroqLLMStream
from configs import GROQ_API_KEY, GROQ_MODEL_NAME
from news import getNews
from prompts: SYSTEM_PROMPT

llm = GroqLLMStream(GROQ_API_KEY)

async def newsAgent(query: str):
    retrieved_news_items = await getNews(query)
    if not retrieved_news_items:
        yield "\n_无法获取与搜索查询相关的新闻._"
        return
    retrieved_news_items = retrieved_news_items.get("results")
    useful_meta_keys = ["title", "link", "keywords", "creator", "description", "country", "category"]
    news_items = [{k: d[k] for k in useful_meta_keys} for d in retrieved_news_items]
    messages = [{"role": "user", "content": f"Query: {query}\n\nNews Items: {news_items}"}]
    async for chunk in llm(GROQ_MODEL_NAME, messages, system=SYSTEM_PROMPT, max_tokens=1024, temperature=0.2):
        yield chunk

上面我们导入了与 Llama 3 交互、上下文管理、系统提示和新闻检索所需的模块。然后定义了 newsAgent 函数，将用户查询作为唯一参数。

在 newsAgent 中，我们首先通过 Newsdata.io API 检索新闻，然后提取相关键传递给 LLM。之后将查询、检索到的新闻项、系统提示和模型名称传递给流式 Groq 接口，随着生成和接收的内容逐块返回。

环境变量和配置

要运行生成式 AI 新闻搜索应用，我们需要设置以下环境变量：

• 环境变量

GROQ_API_KEY="YOUR_GROQ_API_KEY"
GROQ_MODEL_NAME="llama3-8b-8192"
NEWS_API_KEY="YOUR_NEWS_API_KEY"
NEWS_BASE_URL="https://newsdata.io/api/1/"

我们需要从 Groq 控制台 获取 Groq API Key，并从 Newsdata.io 获取 API Key 以检索新闻。

• 加载环境变量

import os
from dotenv import load_dotenv

load_dotenv()

GROQ_API_KEY = os.environ.get("GROQ_API_KEY")
GROQ_MODEL_NAME = os.environ get("GROQ_MODEL_NAME")
NEWS_API_KEY = os.environ get("NEWS_API_KEY")
NEWS_BASE_URL = os.environ get("NEWS_BASE_URL")

暴露 API

我们的生成式 AI 新闻搜索智能体几乎准备好了。我们只需要通过一个流式 API 将其暴露出来。为此，我们将使用 FastAPI 和 Uvicorn，如下代码所示。

# app.py

from fastapi import FastAPI
from fastapi.responses: StreamingResponse
from fastapi.middleware: CORSMiddleware
import uvicorn

from agent import newsAgent

app = FastAPI()

origins = ["*"]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

@app.get("/")
async def index():
    return {"ok": True}

@app.get("/api/news")
async def api_news(query: str):
    return StreamingResponse(newsAgent(query), media_type="text/event-stream")

if __name__ == "__main__":
    uvicorn.run("app:app", host="0.0.0.0", port=8899, reload=True)

上面我们导入了 newsAgent 以及所需的 FastAPI 和 Uvicorn 模块并设置了 FastAPI 应用。

我们创建了一个索引端点以进行健康检查。我们的新闻搜索智能体通过 /api/news 路径返回流式响应。

完成 app.py 文件后，我们可以通过以下命令启动服务器。

python app.py

服务器将在端口号 8899 启动。

现在我们可以在浏览器上访问 http://localhost:8899/api/news?query=searchtext 来获取新闻，如下所示。

整个代码库可以在以下链接找到。

GitHub - vatsalsaglani/GenAINewsAgent: 一个快速实现的生成式 AI 新闻摘要智能体...

结论

在这篇博客中，我们看到了如何通过 Groq 提供的更快 LPU 接口获得接近实时的推理结果。我们还浏览了 Llama 3 的基准分数，并将较小的 Llama 3 8B 模型集成用于新闻摘要。

这就是这篇博客的全部内容，希望你读得愉快。