Advanced chat

.env.example

@@ -2,7 +2,7 @@ DATABASE_URL=postgresql+asyncpg://fitness:fitness@db:5432/fitness
 OPENAI_API_KEY="your_openai_key_here"
 GOOGLE_API_KEY="your_google_key_here"
 TELEGRAM_BOT_TOKEN="your_telegram_bot_token_here"
-REDIS_HOST=redis
+REDIS_HOST=localhost
 REDIS_PORT=6379
 REDIS_DB=0
 LLM_CONFIG_PATH=llm/llm_config.yaml

.github/workflows/ci.yml

@@ -20,6 +20,10 @@ jobs:
           POSTGRES_DB: fitness
         ports:
           - 5432:5432
+      redis:
+        image: redis:7
+        ports:
+          - 6379:6379
 
     steps:
       - name: 🧾 Checkout repository
@@ -87,3 +91,6 @@ jobs:
 
       - name: ✅ Run Onboarding Test
         run: poetry run pytest tests/test_onboarding.py
+
+      - name: ✅ Run Redis Test
+        run: poetry run pytest tests/test_redis.py

llm/chat_memory.py

@@ -3,14 +3,17 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 from src.database.models.message import Message
 from redis.asyncio import Redis
+from src.database.models import SubSummary
 from llm.config_loader import CONFIG
 from llm.models import get_llm
+from src.database.connection import session_maker
 import logging
 
 logger = logging.getLogger(__name__)
 
 MAX_RECENT_MESSAGES = CONFIG.get("max_recent_messages", 20)
 SUMMARY_SIZE = CONFIG.get("summary_chunk_size", 10)
+MAX_SUMMARY_STACK = CONFIG.get("max_summary_stack", 10)  
 
 
 async def save_message_to_redis(telegram_id: int, role: str, content: str, redis_client: Redis | None = None):
@@ -74,9 +77,18 @@ async def create_sub_summary(telegram_id: int, redis_client: Redis | None = None
     response = await llm.ainvoke(prompt)
 
     summary_key = f"user:{telegram_id}:sub_summaries"
+
+    # Check if stack of sub-summaries is full
+    current_stack = await redis_client.llen(summary_key)
+    if current_stack >= MAX_SUMMARY_STACK:
+        # Save to Postgres
+        await save_sub_summaries_to_db(telegram_id, redis_client)
+        await redis_client.delete(summary_key)  # Clear stack in Redis
+
+    # Push new summary
     await redis_client.rpush(summary_key, response.content)
 
-    # Drop the first SUMMARY_SIZE messages (the oldest ones)
+    # Trim recent history
     await redis_client.ltrim(key, SUMMARY_SIZE, -1)
 
 
@@ -90,4 +102,18 @@ async def get_latest_sub_summary(telegram_id: int, redis_client: Redis | None =
     redis_client = redis_client or await get_redis()
     summary_key = f"user:{telegram_id}:sub_summaries"
     latest = await redis_client.lindex(summary_key, -1)
-    return latest
+    return latest
+
+
+async def save_sub_summaries_to_db(telegram_id: int, redis_client: Redis):
+    summary_key = f"user:{telegram_id}:sub_summaries"
+    summaries = await redis_client.lrange(summary_key, 0, -1)
+
+    if not summaries:
+        return
+
+    async with session_maker() as session:
+        for s in summaries:
+            sub_summary = SubSummary(telegram_id=telegram_id, summary=s)
+            session.add(sub_summary)
+        await session.commit()

llm/llm_config.yaml

@@ -12,4 +12,5 @@ yandex:
   iam_token: "IAM_TOKEN"
 
 max_recent_messages: 6
-summary_chunk_size: 3
+summary_chunk_size: 3
+max_summary_stack: 2

pyproject.toml

@@ -38,13 +38,13 @@ alembic = "^1.15.1"
 pytest = "^8.3.5"
 redis = "^5.2.1"
 asyncio = "^3.4.3"
-
+httpx = "0.28.1"
 
 [tool.poetry.group.dev.dependencies]
 pytest = "^8.3.5"
-httpx = "^0.28.1"
 pre-commit = "^4.1.0"
 pytest-asyncio = "^0.25.3"
+asgi-lifespan = "^2.1.0"
 
 [build-system]
 requires = ["poetry-core"]

src/database/models/__init__.py

@@ -3,6 +3,7 @@
 from .activity import Activity
 from .train_program import Program
 from .message import Message
+from .sub_summary import SubSummary
 
 
-__all__ = ["User", "Action", "Activity", "Program", "Message"]
+__all__ = ["User", "Action", "Activity", "Program", "Message",  "SubSummary"]

src/database/models/sub_summary.py

@@ -0,0 +1,12 @@
+from sqlalchemy import Column, DateTime, Integer, String, ForeignKey, func
+from src.database.models.base import Base
+
+__all__ = ["SubSummary"]
+
+class SubSummary(Base):
+    __tablename__ = "sub_summaries"
+
+    id = Column(Integer, primary_key=True, index=True)
+    telegram_id = Column(Integer, ForeignKey("users.telegram_id"), nullable=False, index=True)
+    summary = Column(String, nullable=False)
+    timestamp = Column(DateTime, server_default=func.now(), nullable=False)

src/logging_config.py

@@ -0,0 +1,12 @@
+import logging
+import sys
+
+def setup_logging(level: int = logging.INFO):
+    """Set up centralized logging configuration."""
+    logging.basicConfig(
+        level=level,
+        format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
+        handlers=[
+            logging.StreamHandler(sys.stdout),
+        ]
+    )

src/migrations/versions/55abc90e848f_add_subsummary.py

@@ -0,0 +1,41 @@
+"""Add subSummary
+
+Revision ID: 55abc90e848f
+Revises: d2f8b0a76127
+Create Date: 2025-03-30 17:30:54.302965
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = '55abc90e848f'
+down_revision: Union[str, None] = 'd2f8b0a76127'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table('sub_summaries',
+    sa.Column('id', sa.Integer(), nullable=False),
+    sa.Column('telegram_id', sa.Integer(), nullable=False),
+    sa.Column('summary', sa.String(), nullable=False),
+    sa.Column('timestamp', sa.DateTime(), server_default=sa.text('now()'), nullable=False),
+    sa.ForeignKeyConstraint(['telegram_id'], ['users.telegram_id'], ),
+    sa.PrimaryKeyConstraint('id')
+    )
+    op.create_index(op.f('ix_sub_summaries_id'), 'sub_summaries', ['id'], unique=False)
+    op.create_index(op.f('ix_sub_summaries_telegram_id'), 'sub_summaries', ['telegram_id'], unique=False)
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_index(op.f('ix_sub_summaries_telegram_id'), table_name='sub_summaries')
+    op.drop_index(op.f('ix_sub_summaries_id'), table_name='sub_summaries')
+    op.drop_table('sub_summaries')
+    # ### end Alembic commands ###

src/views/redis.py

@@ -2,7 +2,7 @@
 from src.dependencies.redis import get_redis
 import json
 
-router = APIRouter(prefix="/redis", tags=["redis"])
+router = APIRouter(tags=["redis"])
 
 
 @router.get("/ping")

tests/test_redis.py

@@ -0,0 +1,22 @@
+import pytest
+from httpx import AsyncClient, ASGITransport
+from api.app import get_application
+
+
+@pytest.mark.asyncio
+async def test_redis_ping():
+    app = get_application()
+
+    transport = ASGITransport(app=app)
+    async with AsyncClient(transport=transport, base_url="http://test") as client:
+        # manually trigger Redis startup
+        await app.router.startup()
+
+        res = await client.get("/api/v1/redis/ping")
+        assert res.status_code == 200
+        data = res.json()
+        assert data["status"] == "OK"
+        assert data["ping"] == "pong"
+
+        # manually trigger Redis shutdown
+        await app.router.shutdown()