使用 Llama3 + Groq构建新闻机器人（附源码）

              # news.py              

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

# 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:                          # 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"]                      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, 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

# llms/ctx.py              


from typing import List, Dict, Literal, Union              from 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 = 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(managed_messages)                      print(f"TOTAL TOKENS: ", current_length)                      return managed_messages[::-1]

SYSTEM_PROMPT = """你是一个新闻摘要机器人。当用户提供查询时，你将收到与该查询相关的几篇新闻。你的任务是评估这些新闻与查询的相关性，并仅保留相关的新闻。如果有相关的新闻，你应该以简洁、专业和尊重的方式对它们进行总结。摘要应以第一人称形式呈现，并且你必须以 markdown 格式提供新闻文章的引用。不要告诉用户已经审阅或找到的新闻数量；专注于提供相关文章的简洁摘要。在用户的查询中找不到相关新闻的情况下，礼貌地回复，说明你目前无法提供答案。请记住，你的回复应直接回应用户的兴趣，而不透露后端过程或数据检索的具体细节。例如，如果查询是关于“2024年洛克萨巴选举”，并且找到了相关文章，提供这些文章的摘要。如果文章与查询无关或无用，礼貌地告知用户你无法提供所需信息。"""

# agent.py              


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

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

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

python app.py