Commit a28e107
Changed files (1)
src
llm
src/llm/summary.py
@@ -1,10 +1,7 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
-import base64
import json
import re
-from pathlib import Path
-from typing import TYPE_CHECKING
from loguru import logger
from pyrogram.client import Client
@@ -19,9 +16,6 @@ from messages.progress import modify_progress
from messages.sender import send2tg
from messages.utils import equal_prefix
-if TYPE_CHECKING:
- from io import BytesIO
-
HELP = f"""🤖**GPT总结历史消息** (最多{COMBINATION_MAX_HISTORY}条)
当前模型:
- 文本模型: **{GPT.TEXT_MODEL_NAME}**
@@ -122,8 +116,11 @@ def get_summay_model(history: list[dict]) -> dict:
return config
-async def get_contexts(client: Client, history: list[dict]) -> list[dict]:
- """Get GPT contexts based on parsed chat history."""
+async def get_contexts(client: Client, history: list[dict]) -> list[dict]: # noqa: ARG001
+ """Get GPT contexts based on parsed chat history.
+
+ Currently, we only summarize text contents.
+ """
contexts = [
{
"role": "system", # system prompt
@@ -160,23 +157,25 @@ async def get_contexts(client: Client, history: list[dict]) -> list[dict]:
]
user_contexts = []
for info in history:
+ if info["text"].startswith("/"): # commands
+ continue
if info["mtype"] == "text" and info["text"]:
content = {"user": info["full_name"], "time": f"{info['datetime']:%H:%M:%S}", "message": info["text"]}
user_contexts.append({"type": "text", "text": json.dumps(content, ensure_ascii=False)})
- continue
- if info["mtype"] == "photo":
- content = {"user": info["full_name"], "time": f"{info['datetime']:%H:%M:%S}", "message": "[image]如下"}
- user_contexts.append({"type": "text", "text": json.dumps(content, ensure_ascii=False)})
- res: BytesIO = await client.download_media(info["file_id"], in_memory=True) # type: ignore
- ext = Path(res.name).suffix.removeprefix(".").replace("jpg", "jpeg")
- b64 = base64.b64encode(res.getvalue()).decode("utf-8")
- user_contexts.append({"type": "image_url", "image_url": {"url": f"data:image/{ext};base64,{b64}"}})
- if info["text"]:
- content = {"user": info["full_name"], "time": f"{info['datetime']:%H:%M:%S}", "message": info["text"]}
- user_contexts.append({"type": "text", "text": json.dumps(content, ensure_ascii=False)})
- else:
- content = {"user": info["full_name"], "time": f"{info['datetime']:%H:%M:%S}", "message": f"[{info['mtype']}] {info['text']}".strip()}
- user_contexts.append({"type": "text", "text": json.dumps(content, ensure_ascii=False)})
+ # continue
+ # if info["mtype"] == "photo":
+ # content = {"user": info["full_name"], "time": f"{info['datetime']:%H:%M:%S}", "message": "[image]如下"}
+ # user_contexts.append({"type": "text", "text": json.dumps(content, ensure_ascii=False)})
+ # res: BytesIO = await client.download_media(info["file_id"], in_memory=True) # type: ignore
+ # ext = Path(res.name).suffix.removeprefix(".").replace("jpg", "jpeg")
+ # b64 = base64.b64encode(res.getvalue()).decode("utf-8")
+ # user_contexts.append({"type": "image_url", "image_url": {"url": f"data:image/{ext};base64,{b64}"}})
+ # if info["text"]:
+ # content = {"user": info["full_name"], "time": f"{info['datetime']:%H:%M:%S}", "message": info["text"]}
+ # user_contexts.append({"type": "text", "text": json.dumps(content, ensure_ascii=False)})
+ # else:
+ # content = {"user": info["full_name"], "time": f"{info['datetime']:%H:%M:%S}", "message": f"[{info['mtype']}] {info['text']}".strip()}
+ # user_contexts.append({"type": "text", "text": json.dumps(content, ensure_ascii=False)})
contexts.append({"role": "user", "content": user_contexts})
logger.trace(contexts)
return contexts