导读
通过本文你将了解：
什么是推理速度出奇快的Groq
Llama 3的基准效果
-
Llama3+Groq构建新闻机器人的源代码
关注公众号，后台输入"Llama3+Groq" 获取源代码
没错，如果你还没听过Groq，你已经Out了，这里不讨论Groq的原理，也不夸赞Groq有多快，你应该自己试一试，绝对惊掉下巴。
在本文中，我们将为生成式 AI 新闻搜索创建一个后端。我们将使用 Meta 的 Llama-3 8B 模型通过 Groq 的 LPU 进行提供。
关于 Groq
如果你还没有听说过 Groq，那让我来启发你一下。Groq正在为大型语言模型（LLMs）的文本生成推理速度设定新的标准。
Groq提供了一种新型的端到端处理单元系统，即LPU（语言处理单元）接口引擎，为具有顺序组件的计算密集型应用提供了最快的推理能力，就像LLMs中那样。
我不会深入探讨Groq相比GPU的推理速度有多快(以后也许会有相关文章)，你应该自己体验一下，反正我第一次体验的时候被震惊到了（相对于一个字一个字蹦的chatgpt）
本文将利用Groq和Llama 3提供的速度增加以及文本生成能力，来创建一个生成式AI新闻搜索。这将类似于必应AI搜索、谷歌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 个关键用例：
寻求建议
头脑风暴
分类
封闭问题回答
coding
创意写作
提取
扮演角色/人物
开放问题回答
推理
重写
摘要
这个集合对模型开发团队保密，以避免意外过度拟合他们的模型。
他们测试了 Llama 3 70B 与 Claude Sonnet、Mistral Medium、GPT-3.5 和 Llama 2 等模型的对比。下图显示了 Llama 3 对上述所有模型的胜率。
有了所有这些对 Llama 3 的基准支持，我们可以决定将其用于我们的生成式 AI 新闻搜索。
一般来说，较小的模型提供更快的推理，因为它们不会消耗大量的 VRAM，而且由于参数计算较少，令牌生成速度更快，因此我们可以选择使用较小的 Llama 3 模型，即 Llama 3 8B 模型。
新闻 API
我们将使用来自 Newsdata.io 的免费新闻 API 来根据搜索查询检索新闻内容。我们甚至可以使用来自 Google 的 RSS 订阅来实现这一点，或者可以使用其他任何新闻 API。
可以通过在 Newsdata 平台上注册后生成 API 令牌来访问 Newsdata 新闻 API。一旦我们获得了 API 令牌，就只需使用搜索查询进行 GET 调用，检索结果，并将其传递给 LLM。
我们将使用以下代码片段来使用 Newsdata.io API 检索新闻。
import os import httpx from configs import NEWS_API_KEY, NEWS_BASE_URL
asyncdef getNews(query: str, max_size: int = 8): asyncwith 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 库](https://github.com/groq/groq-python) 与 Llama 3 8B 模型进行交互。
以下是与 Groq 交互的方式。
# llms/groq.py
from groq import Groq, AsyncGroq import traceback from typing import 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=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"] 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.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=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, model=model, stream=True, **kwargs) async for chunk in output: yield chunk.choices[0].delta.content or ""
from abc import ABC, abstractmethod from typing import 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 asyncdef__call__(self, model: str, messages: List[Dict], **kwargs): pass
使用 Llama 3 8B，我们可以在上下文长度中使用 8192 个标记。其中，我们将保留 7000 个标记用于输入上下文，其余用于输出或生成。
输入上下文可以高于 7000 个标记，那么在这种情况下，我们需要管理这个上下文，以便在上下文中留下足够多的标记用于输出生成。为此，我们编写了下面提供的 ContextManagement 实用程序。
from typing import 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: int = 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 = self.__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_message_role = message.get("role") 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 = self.__pad_content__(content, tokens_to_keep) current_length += tokens_to_keep managed_messages.append({ "role": message.get("role"), "content": content }) 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 库来对我们的消息进行标记化，并计算标记数，并仅保留符合我们上面约定的最大标记长度，即 7000。
要使用 meta-llama/Meta-Llama-3–8B 标记器，我们首先需要提供我们的详细信息，并接受 HuggingFace 上 Meta 提供的使用条款，并通过使用 huggingface-cli login 命令或在 AutoTokenizer 的 from_pretrained 方法中提供标记来将我们的 HuggingFace 标记添加到我们的机器上。
简单的Prompts
我们将为我们的生成式AI新闻搜索应用程序使用一个非常简单的提示。提示如下所示。
SYSTEM_PROMPT = """你是一个新闻摘要机器人。当用户提供查询时，你将收到与该查询相关的几篇新闻。你的任务是评估这些新闻与查询的相关性，并仅保留相关的新闻。如果有相关的新闻，你应该以简洁、专业和尊重的方式对它们进行总结。摘要应以第一人称形式呈现，并且你必须以 markdown 格式提供新闻文章的引用。不要告诉用户已经审阅或找到的新闻数量；专注于提供相关文章的简洁摘要。在用户的查询中找不到相关新闻的情况下，礼貌地回复，说明你目前无法提供答案。请记住，你的回复应直接回应用户的兴趣，而不透露后端过程或数据检索的具体细节。例如，如果查询是关于“2024年洛克萨巴选举”，并且找到了相关文章，提供这些文章的摘要。如果文章与查询无关或无用，礼貌地告知用户你无法提供所需信息。"""
Agent
让我们把一切都整合在一起，完成我们的生成式人工智能新闻搜索代理。
from llms.groq import GroqLLMStream from configs import GROQ_API_KEY, GROQ_MODEL_NAME from news import getNews from prompts import SYSTEM_PROMPT
llm = GroqLLMStream(GROQ_API_KEY)
asyncdef newsAgent(query: str): retrieved_news_items = await getNews(query) ifnot retrieved_news_items: yield"\n_Cannot fetch any relevant news related to the search query._" 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}" }] asyncfor 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 接口，并在生成和接收到的时候产生块。
环境变量和配置
我们需要设置以下环境变量来运行我们的 GenerativeAI News Search 应用程序。
环境变量
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 Key 和 Newsdata.io 获取 API 密钥来检索新闻。
加载环境变量
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
我们的 GenerativeAI 新闻搜索代理几乎已经准备就绪。我们只需要通过流式 API 。为此，我们将使用 FastAPI 和 Uvicorn，如下所示的代码。
from fastapi import FastAPI from fastapi.responses import StreamingResponse from fastapi.middleware.cors import 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("/") asyncdef index(): return {"ok": True}
@app.get("/api/news") asyncdef 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 文件后，我们可以使用以下命令启动服务器。
服务器将在端口号 8899 上启动。
现在，我们可以转到我们的浏览器，访问 [http://localhost:8899/api/news?query=search](http://localhost:8899/api/news?query=search)text，以以下方式获取我们的新闻。
结论
本文如介绍了如何利用Groq提供的更快的LPU接口来实现接近实时的推断。
我们还一览了Llama 3的基准分数，并集成了较小的Llama 3 8B模型用于新闻摘要。
粤ICP备14082021号
免费POC，
零成本试错
