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

# 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

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

# 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

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

# 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."""

# 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_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/>"

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

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

python app.py