53 lines
1.7 KiB
Python
53 lines
1.7 KiB
Python
"""FluentGerman.ai — Instruction assembly service."""
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models import Instruction, InstructionType
|
|
|
|
|
|
async def get_system_prompt(db: AsyncSession, user_id: int) -> str:
|
|
"""Assemble the full system prompt from global + personal + homework instructions."""
|
|
|
|
# Global instructions (no user_id)
|
|
result = await db.execute(
|
|
select(Instruction).where(
|
|
Instruction.user_id.is_(None),
|
|
Instruction.type == InstructionType.GLOBAL,
|
|
)
|
|
)
|
|
global_instructions = result.scalars().all()
|
|
|
|
# Personal + homework for this user
|
|
result = await db.execute(
|
|
select(Instruction).where(Instruction.user_id == user_id)
|
|
)
|
|
user_instructions = result.scalars().all()
|
|
|
|
parts: list[str] = []
|
|
|
|
if global_instructions:
|
|
parts.append("=== TEACHING METHOD ===")
|
|
for inst in global_instructions:
|
|
parts.append(f"[{inst.title}]\n{inst.content}")
|
|
|
|
personal = [i for i in user_instructions if i.type == InstructionType.PERSONAL]
|
|
if personal:
|
|
parts.append("\n=== PERSONAL INSTRUCTIONS ===")
|
|
for inst in personal:
|
|
parts.append(f"[{inst.title}]\n{inst.content}")
|
|
|
|
homework = [i for i in user_instructions if i.type == InstructionType.HOMEWORK]
|
|
if homework:
|
|
parts.append("\n=== CURRENT HOMEWORK ===")
|
|
for inst in homework:
|
|
parts.append(f"[{inst.title}]\n{inst.content}")
|
|
|
|
if not parts:
|
|
return (
|
|
"You are a helpful German language tutor. Help the student learn German "
|
|
"through conversation, corrections, and explanations."
|
|
)
|
|
|
|
return "\n\n".join(parts)
|