通过 ChatGPT 生成的图像
在本博客中，我们将为生成式人工智能新闻搜索创建一个后端。我们将使用通过 Groq 的 LPU 提供服务的 Meta 的 Llama-3 8B 模型。
关于格罗克
如果您还没有听说过 Groq，那么让我为您介绍一下。 Groq 正在为大型语言模型 (LLM) 中文本生成的推理速度设定新标准。 Groq 提供 LPU（语言处理单元）接口引擎，这是一种新型的端到端处理单元系统，可为计算密集型应用程序提供最快的推理，并为其提供顺序组件，如法学硕士。
我们不会深入探讨与 GPU 相比，Groq 上的推理速度有多快。我们希望利用 Groq 和 Llama 3 文本生成功能提供的速度提升来创建生成式 AI 新闻搜索。这将类似于 Bing AI 搜索、Google AI 搜索或 PPLX。
为什么是骆驼3？
Meta 最近发布的 Llama 3 型号大受欢迎。更大的 70B Llama 3 型号目前在 LMSys LLM 排行榜上排名第五。在英语任务中，同一模型排名第二，仅次于 GPT-4。
根据 Meta 的 Llama 3 发布博客，8B 型号是同类产品中最好的，70B 型号优于 Gemini Pro 1.5 和 Claude 3 Sonnet。
基准图像取自 Meta Llama 3 发布博客
为了展示模型对现实场景和问题的理解，Meta 创建了高质量的人类评估集。该评估集包含 1800 个提示，涵盖 12 个关键用例：
- 寻求建议
- 头脑风暴
- 分类
- 封闭式问答
- 编码
- 创意写作
- 萃取
- 居住在一个角色/角色中
- 开放式问答
- 推理
- 重写
- 总结
该数据集对模型开发团队保密，以避免模型在此数据上意外过度拟合。他们将 Llama 3 70B 与 Claude Sonnet、Mistral Medium、GPT-3.5 和 Llama 2 等模型进行了测试。下图显示了 Llama 3 相对于上述所有模型的胜率。
人类评估取自 Meta 的 Llama 3 发布博客
有了所有这些有利于 Llama 3 的基准，我们可以决定将其用于我们的生成式 AI 新闻搜索。
一般来说，较小的模型提供更快的推理，因为它们不消耗大量的 VRAM，并且由于参数计算较少，token 生成速度更快，因此我们可以选择使用较小的 Llama 3 模型，即 Llama 3 8B 模型。
让我们对这个事情进行编码。
新闻API
我们将使用Newsdata.io提供的免费新闻 API根据搜索查询检索新闻内容。我们甚至可以使用 Google 的 RSS 提要来实现此目的，也可以是任何其他新闻 API。
您可以通过在 Newsdata 平台注册后生成的 API 令牌访问 Newsdata 新闻 API。一旦我们获得了 API 令牌，只需使用搜索查询进行 GET 调用、检索结果并将其传递给 LLM 即可。
我们将使用以下代码片段通过 Newsdata.io API 检索新闻。
#news.py
导入操作系统
导入httpx
从配置导入NEWS_API_KEY、NEWS_BASE_URL
async def getNews（query：str，max_size：int = 8）：
异步 使用httpx.AsyncClient（timeout= 60）作为客户端：
response = await client.get（
os.path.join（NEWS_BASE_URL，“news”）+
f“？apiKey= {NEWS_API_KEY} &q= {query} &size= {max_size} ”
尝试：
response.raise_for_status（）
返回response.json（） 除了
httpx.HTTPStatusError作为e：
print（
f“请求{e.request.url！r}时出现错误响应{e.response.status_code} ”）返回无
上面我们使用该httpx库通过 API 令牌和搜索词对 API 进行异步调用。如果响应状态代码是，200我们返回响应，否则我们打印异常并返回None。
格罗克接口
Groq 通过通过 API 密钥进行身份验证的 REST API 提供 Llama 3 8B 模型。我们还可以通过官方的Groq Python 库与 Llama 3 8B 模型进行交互。
以下是我们与 Groq 交互的方式。
# llms/groq.py
from groq import Groq, AsyncGroq从
输入导入回溯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 )) 输出 =等待self.client.chat.completions.create( messages=messages, model=model, **kwargs)返回output.choices[ 0 ].message.content except RateLimitError: raise RateLimitError except Exception as err: print ( f"ERROR: { str (err)} " ) print ( f" {traceback.format_exc()} " ) return ""类GroqLLMStream ( BaseLLM ): def __init__ ( self, api_key:联盟
[ 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" )
}] + messages
del 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" ]
输出=等待self.client.chat.completions.create(messages=messages,
model=model,
stream= True ,
**kwargs)输出中的块异步
： # print(chunk.choices[0]) Yield chunk.choices[ 0 ].delta.content或""
我通常更喜欢从 a 继承 LLM 类，BaseLLM这样我就可以通过基础控制所有常见的实用程序。以下是BaseLLM.
# llms/base.py
from abc import ABC，abstractmethod
fromtypes 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 async def __call__ ( self, model: str , messages: List [ Dict ], **kwargs ): pass
使用 Llama 3 8B，我们可以选择在上下文长度中使用 8192 个标记。其中，我们将保留 7000 个标记用于输入上下文，其余用于输出或生成。
输入上下文可能高于 7000 个令牌，在这种情况下，我们需要管理此上下文，以便在上下文中留下大量令牌用于输出生成。为此，我们编写了ContextManagement下面提供的实用程序。
# llms/ctx.py
来自输入import List , Dict , Literal , Union
from Transformers import AutoTokenizer
class ContextManagement :
def __init__ ( self ):
# 在 model_name 中断言 "mistral", "MistralCtx 仅适用于 Mistral 模型"
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 :
如果current_length + message_tokens >= max_length:
tokens_to_keep = max_length - current_length
如果tokens_to_keep > 0 :
content = self.__pad_content__(content, tokens_to_keep)
current_length += tokens_to_keep
else :
中断
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
})
当前消息角色 = 消息。得到（"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( "角色" ),
"内容" : 内容
})
else :
中断
else :
Managed_messages.append({
"角色" : message.get( "角色" ),
"内容" : 内容
})
current_length += message_tokens
current_message_role = message.get( "role" )
# print(managed_messages)
print ( f"TOTAL TOKENS: " , current_length)
return Managed_messages[::- 1 ]
上面我们使用 HuggingFacetokenizers库来标记我们的消息并计算标记，并仅保留符合我们上面商定的最大标记长度（即 7000）的消息和消息内容。
要使用标记生成器，
meta-llama/Meta-Llama-3–8B我们首先需要提供详细信息并接受 Meta 在 HuggingFace 上提供的使用条款，并通过使用命令huggingface-cli login或在.from_pretrainedAutoTokenizer
简单提示
我们将为生成人工智能新闻搜索应用程序使用一个非常简单的提示。下面提供了提示。
#提示.py SYSTEM_PROMPT = """你是一个新闻摘要机器人。当用户提供查询时，你将收到与该查询相关的几条新闻条目。你的任务是评估这些新闻条目与查询的相关性并保留如果有相关 的新闻，则应以简洁、专业和尊重的方式进行总结。摘要应以第一人称形式提供，并且必须以 Markdown 格式提供新闻文章的引用。不要告知用户已查看或找到的新闻条目的数量；仅专注于提供相关文章的简洁摘要。 如果找不到用户查询的相关新闻条目，请礼貌地回答，说明您无法提供。请记住，您的回复应直接解决用户的兴趣，而不透露后端流程或数据检索的细节。 例如，如果查询是关于“Lok Sabhaelections 2024”并且找到了相关文章，请提供。这些文章的摘要。如果文章不相关或无用，请恭敬地告知用户您无法提供所需信息。 ”“”
提示非常简单易懂。
中介
让我们将所有内容整合在一起并完成我们的生成式人工智能新闻搜索代理。
# agent.py
from llms.groq import GroqLLMStream
from configs import GROQ_API_KEY, GROQ_MODEL_NAME
from news import getNews
from Tips import SYSTEM_PROMPT
llm = GroqLLMStream(GROQ_API_KEY)
async def newsAgent ( query: str ):
returned_news_items = wait getNews(query)
if not returned_news_items:
yield "\n_无法获取与搜索查询相关的任何相关新闻。_"
return
returned_news_items = returned_news_items.get( "results" )
Used_meta_keys = [
"title" , "link" , "keywords" , "creator" , "description" , "国家" ,
"类别"
]
news_items = [{
k: d[k] for
k inUsed_meta_keys }
for d inretrieve_news_items ] messages = [{ "role" : "user" , "content" : f"Query: {query } \n\n新闻项目: {news_items} " }] async for chunk in llm(GROQ_MODEL_NAME, messages, system=SYSTEM_PROMPT, max_tokens= 1024 , temperature= 0.2 ):产量块
上面我们导入了我们为在 Groq 上与 Llama 3 交互、上下文管理、系统提示和新闻检索而编写的所有必要模块。之后，我们定义newsAgent将用户查询或搜索查询作为唯一参数的函数。
在本文中，newsAgent我们首先通过 Newsdata.io API 检索新闻，然后收集我们想要传递给 LLM 的相关密钥。然后，我们将查询、检索到的新闻项、系统提示以及模型名称传递到我们的流式 Groq 接口，并在生成和接收时生成块。
环境变量和配置
我们需要设置以下环境变量来运行 GenerativeAI 新闻搜索应用程序。
- 环境变量
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 密钥来检索新闻。
- 加载环境变量
从dotenv导入 o s import load_dotenv load_dotenv () GROQ_API_KEY = os.environ.获取（“GROQ_API_KEY”） GROQ_MODEL_NAME = os.environ。获取（“GROQ_MODEL_NAME”） NEWS_API_KEY = os.environ。获取（“NEWS_API_KEY”） NEWS_BASE_URL = os.environ。获取（“NEWS_BASE_URL”）
公开API
我们的 GenerativeAI 新闻搜索代理即将准备就绪。我们只需要通过流 API 公开它。为此，我们将使用 FastAPI 和 Uvicorn，如下面的代码所示。
# app.py
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 ,
allowed_methods=[ "*" ],
allowed_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"，主机= “0.0.0.0”，端口= 8899，重新加载= True）
在上面，我们导入了newsAgent所需的 FastAPI 和 Uvicorn 模块并设置了 FastAPI 应用程序。
我们创建一个索引端点只是为了健康检查。我们的新闻搜索代理通过/api/news返回流响应的路由公开。
完成该app.py文件后，我们可以使用以下命令启动服务器。
蟒蛇应用程序.py
服务器将在端口号8899 上启动。
现在，我们可以使用浏览器http://localhost:8899/api/news?query=searchtext并通过以下方式获取新闻。
作者的 GIF
整个代码库可在下面提供的链接中找到。
