用Llama 3和Groq打造生成式AI新闻摘要

浏览次数： 1583

在这篇文章中，我们要一起打造一个生成式 AI 新闻搜索后台。我们的选择是 Meta 的 Llama-3 8B 模型，这一切都将通过 Groq 的 LPU 来实现。

关于 Groq

如果你还不知道 Groq，那就让我来介绍一下。Groq 正在重新定义大型语言模型（LLM）中文本生成的推理速度标准。它提供了一种名为 LPU（语言处理单元）的接口引擎，这是一种全新的端到端处理单元系统，为那些包含序列组件的计算密集型应用提供了前所未有的快速推理能力，正如在大型语言模型中看到的那样。

我们这里不打算深入讨论为什么 Groq 的推理速度能远超 GPU。我们的目标是利用 Groq 提供的速度优势和 Llama 3 在文本生成方面的能力，创建一个与 Bing AI 搜索、Google AI 搜索或 PPLX 类似的生成式 AI 新闻搜索服务。

为什么是 Llama 3？

Meta 最近发布的 Llama 3 模型非常受欢迎。其中更大的 70B Llama 3 模型目前在 LMSys LLM 排行榜中位居第五。在英语任务方面，这一模型排名第二，仅次于 GPT-4。

Meta 在其发布博客中提到，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 的优势，我们决定将其用于我们的生成式 AI 新闻搜索。

一般来说，较小的模型因为不需要大量的 VRAM 并且参数计算更少，所以推理速度更快，因此我们选择使用较小的 Llama 3 模型，即 Llama 3 8B 模型。

开始编码。

新闻 API

我们将利用 Newsdata.io 提供的免费新闻 API 根据搜索查询获取新闻内容。我们也可以使用来自 Google 的 RSS 源，或者其他任何新闻 API 来实现这一点。

你可以通过 API 令牌访问 Newsdata 新闻 API，这个令牌可以在 Newsdata 平台注册后获得。一旦我们拿到 API 令牌，就只需要发起一个带有搜索查询的 GET 请求，获取结果，并将其传递给 LLM。

下面是我们用来通过 Newsdata.io API 获取新闻的代码片段。

# news.py
import osimport httpxfrom 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 resposne {e.response.status_code} while requesting {e.request.url!r}")return None

我们使用 httpx 库异步调用 API，传入 API 令牌和搜索词。如果响应状态码是 200，我们返回响应，否则打印异常信息并返回 None。

Groq 接口

Groq 通过通过 API 密钥进行身份验证的 REST API 提供 Llama 3 8B 模型。我们还可以通过官方的 Groq Python 库与 Llama 3 8B 模型进行交互。

以下是我们将如何与 Groq 进行交互。

Groq 通过 REST API 提供 Llama 3 8B 模型，并通过 API 密钥进行认证。我们还可以通过官方的 Groq Python 库与 Llama 3 8B 模型进行交互。

下面是我们如何使用 Groq。

# llms/groq.py
from groq import Groq, AsyncGroqimport tracebackfrom typing import List, Dict, Unionfrom llms.base import BaseLLMfrom llms.ctx import ContextManagementfrom groq import RateLimitErrorimport backoff
manageContext = ContextManagement()
class GroqLLM(BaseLLM):
def __init__(self, api_key: Union[str, None] = None):super().__init__(api_key)self.client = AsyncGroq(api_key=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")}] + messagesdel kwargs["system"]if "ctx_length" in kwargs:del kwargs["ctx_length"]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.contentexcept RateLimitError:raise RateLimitErrorexcept 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=api_key)
async def __call__(self, model: str, messages: List[Dict], **kwargs):if "system" in kwargs:# print(f"System in Args")messages = [{"role": "system","content": kwargs.get("system")}] + messagesdel kwargs["system"]# print(f"KWARGS KEYS: {kwargs.keys()}")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, model=model, stream=True, **kwargs)async for chunk in output:# print(chunk.choices[0])yield chunk.choices[0].delta.content or ""

我习惯于让 LLM 类从 BaseLLM 继承，这样可以通过基类控制所有常见的功能。下面是 BaseLLM 的定义。

# llms/base.py
from abc import ABC, abstractmethodfrom typing import List, Dict, Union
class BaseLLM(ABC):
def __init__(self, api_key: Union[str, None] = None, **kwargs):self.api_key = api_keyself.client = Noneself.extra_args = kwargs
@abstractmethodasync def __call__(self, model: str, messages: List[Dict], **kwargs):pass

使用 Llama 3 8B 模型，我们可以在上下文长度中使用 8192 个词元。其中我们将 7000 个词元用于输入上下文，其余用于输出或生成。

如果输入上下文超过 7000 个词元，那么我们需要管理这个上下文，以便为输出生成保留足够的词元。为此，我们编写了下面的 ContextManagement 工具。

# llms/ctx.py
from typing import List, Dict, Literal, Unionfrom transformers import AutoTokenizer
class ContextManagement:
def __init__(self):# assert "mistral" in model_name, "MistralCtx only available for Mistral models"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: int = 28_000):managed_messages = []current_length = 0current_message_role = Nonefor 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_lengthif tokens_to_keep > 0:content = self.__pad_content__(content, tokens_to_keep)current_length += tokens_to_keepelse:breakif 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_message_role = message.get("role")current_length += message_tokenselse:if current_length + message_tokens >= max_length:tokens_to_keep = max_length - current_lengthif tokens_to_keep > 0:content = self.__pad_content__(content, tokens_to_keep)current_length += tokens_to_keepmanaged_messages.append({"role": message.get("role"),"content": content})else:breakelse:managed_messages.append({"role": message.get("role"),"content": content})current_length += message_tokenscurrent_message_role = message.get("role")# print(managed_messages)print(f"TOTAL TOKENS: ", current_length)return managed_messages[::-1]

我们使用 HuggingFace 的 tokenizers 库来对我们的消息进行标记化和计算词元，并只保留符合我们之前设定的最大词元长度（7000）的消息内容。

要使用 meta-llama/Meta-Llama-3–8B 分词器，我们首先需要在 HuggingFace 上提供我们的详细信息并接受 Meta 提供的使用条款，并通过使用 huggingface-cli login 命令或在 AutoTokenizer 的 from_pretrained 方法中提供 token 将我们的 HuggingFace token 添加到我们的机器上。

提示词

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

# prompts.py
SYSTEM_PROMPT = """You are a news summary bot. When a user provides a query, you will receive several news items related to that query. Your task is to assess the relevance of these news items to the query and retain only the ones that are pertinent.If there are relevant news items, you should summarize them in a concise, professional, and respectful manner. The summary should be delivered in the first person, and you must provide citations for the news articles in markdown format. Do not inform the user about the number of news items reviewed or found; focus solely on delivering a succinct summary of the relevant articles.In cases where no relevant news items can be found for the user's query, respond politely stating that you cannot provide an answer at this time. Remember, your responses should directly address the user's interest without revealing the backend process or the specifics of the data retrieval.For example, if the query is about "Lok Sabha elections 2024" and relevant articles are found, provide a summary of these articles. If the articles are unrelated or not useful, inform the user respectfully that you cannot provide the required information."""

提示非常直接易懂。

代理

让我们把所有的部分整合起来，完成我们的生成式 AI 新闻搜索代理。

# agent.py
from llms.groq import GroqLLMStreamfrom configs import GROQ_API_KEY, GROQ_MODEL_NAMEfrom news import getNewsfrom prompts import 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_无法获取与搜索查询相关的新闻。_""returnretrieved_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}\\n\\n新闻条目: {news_items}" }]async for chunk in llm(GROQ_MODEL_NAME, messages, system=SYSTEM_PROMPT, max_tokens=1024, temperature=0.2):yield chunk

上面我们导入了与 Groq 上的 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 API 密钥和 Newsdata.io 的 API 密钥来检索新闻。

加载环境变量

import osfrom 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 FastAPIfrom fastapi.responses import StreamingResponsefrom fastapi.middleware.cors import CORSMiddlewareimport 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)