Thorsten
f37dc9fa76
FastAPI + Postgres backend with fastapi-users auth, games/characters CRUD, and a WebSocket chat endpoint where an x.ai Grok DM narrates play, rolls dice, and maintains character sheets via tool calls. React + Tailwind frontend covers registration/login, game list and creation, participation-code join flow, character viewer, and a real-time chat session view. Docker Compose skeleton (Postgres + backend) included; Caddy/frontend container wiring is deferred to the deploy milestone. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
60 lines
2.2 KiB
Python
60 lines
2.2 KiB
Python
import uuid
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.character import Character
|
|
from app.models.message import Message
|
|
from app.models.user import User
|
|
|
|
# Rough char-based budget for Phase 1 (~4 chars/token, leaves headroom for system prompt + output).
|
|
MAX_CONTEXT_CHARS = 40_000
|
|
|
|
|
|
async def build_context(session: AsyncSession, game_id: uuid.UUID) -> list[dict]:
|
|
messages = (
|
|
await session.execute(
|
|
select(Message).where(Message.game_id == game_id).order_by(Message.id.asc())
|
|
)
|
|
).scalars().all()
|
|
if not messages:
|
|
return []
|
|
|
|
user_ids = {m.user_id for m in messages if m.user_id is not None}
|
|
character_ids = {m.character_id for m in messages if m.character_id is not None}
|
|
|
|
names: dict[uuid.UUID, str] = {}
|
|
if user_ids:
|
|
rows = (await session.execute(select(User.id, User.name).where(User.id.in_(user_ids)))).all()
|
|
names = {row.id: row.name for row in rows}
|
|
|
|
char_names: dict[uuid.UUID, str] = {}
|
|
if character_ids:
|
|
rows = (
|
|
await session.execute(select(Character.id, Character.name).where(Character.id.in_(character_ids)))
|
|
).all()
|
|
char_names = {row.id: row.name for row in rows}
|
|
|
|
entries: list[dict] = []
|
|
for m in messages:
|
|
if m.sender_type == "dm":
|
|
entries.append({"role": "assistant", "content": m.content})
|
|
elif m.sender_type == "player":
|
|
player_name = names.get(m.user_id, "Spieler") if m.user_id else "Spieler"
|
|
character_name = char_names.get(m.character_id) if m.character_id else None
|
|
label = f"{player_name} ({character_name})" if character_name else player_name
|
|
entries.append({"role": "user", "content": f"{label}: {m.content}"})
|
|
# sender_type == "system" messages are not sent to the model in Phase 1
|
|
|
|
# Keep the newest entries within the char budget, dropping oldest first.
|
|
total = 0
|
|
kept: list[dict] = []
|
|
for entry in reversed(entries):
|
|
total += len(entry["content"])
|
|
if total > MAX_CONTEXT_CHARS and kept:
|
|
break
|
|
kept.append(entry)
|
|
kept.reverse()
|
|
|
|
return kept
|