3926ab676f
RAG proxy that intercepts SkyrimNet LLM requests and enriches them with relevant Tamrielic lore from CHIM's Oghma Infinium database. Features: - FastAPI proxy compatible with OpenAI API - ChromaDB semantic search for lore retrieval - NPC profile extraction from SkyrimNet prompts - Google Sheets ingestion for CHIM's Oghma data - Kubernetes deployment manifests - Debug endpoint for RAG operation monitoring Collections ingested to iris-dev ChromaDB: - oghma_lore: 1951 entries (scholar knowledge) - oghma_basic: 1949 entries (commoner knowledge) - oghma_visual: 1151 entries (Omnisight perception) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
154 lines
5.1 KiB
Python
154 lines
5.1 KiB
Python
"""Lore Injector - Injects retrieved lore into SkyrimNet prompts."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import structlog
|
|
|
|
from .models import InjectionResult, LoreEntry, NPCProfile
|
|
|
|
logger = structlog.get_logger()
|
|
|
|
|
|
class LoreInjector:
|
|
"""Injects Oghma lore into SkyrimNet chat messages."""
|
|
|
|
DEFAULT_TEMPLATE = """
|
|
## Relevant Lore Knowledge
|
|
|
|
Based on your background as a {race} {profession} in {location}, you would know:
|
|
|
|
{lore_items}
|
|
|
|
Note: Reference this knowledge naturally when relevant to the conversation. Do not recite it.
|
|
"""
|
|
|
|
def __init__(self, template: str | None = None, position: str = "after_bio"):
|
|
"""
|
|
Initialize injector.
|
|
|
|
Args:
|
|
template: Jinja-style template for injection block
|
|
position: Where to inject - 'after_bio', 'before_conversation', 'system_suffix'
|
|
"""
|
|
self.template = template or self.DEFAULT_TEMPLATE
|
|
self.position = position
|
|
|
|
def inject(
|
|
self,
|
|
messages: list[dict],
|
|
npc_profile: NPCProfile,
|
|
lore_entries: list[LoreEntry],
|
|
query_time_ms: float,
|
|
) -> tuple[list[dict], InjectionResult]:
|
|
"""
|
|
Inject lore into chat messages.
|
|
|
|
Args:
|
|
messages: Original chat messages
|
|
npc_profile: Extracted NPC profile
|
|
lore_entries: Retrieved lore entries
|
|
query_time_ms: Time taken for retrieval
|
|
|
|
Returns:
|
|
Tuple of (modified messages, injection result)
|
|
"""
|
|
if not lore_entries:
|
|
return messages, InjectionResult(
|
|
npc_profile=npc_profile,
|
|
lore_entries=[],
|
|
injection_text="",
|
|
query_time_ms=query_time_ms,
|
|
)
|
|
|
|
# Build injection text
|
|
injection_text = self._build_injection_text(npc_profile, lore_entries)
|
|
|
|
# Clone messages to avoid modifying original
|
|
modified_messages = [dict(msg) for msg in messages]
|
|
|
|
# Find injection point
|
|
injected = False
|
|
for i, msg in enumerate(modified_messages):
|
|
if msg.get("role") == "system":
|
|
content = msg.get("content", "")
|
|
|
|
if self.position == "after_bio":
|
|
# Inject after character bio section
|
|
bio_markers = ["## Background", "## Personality", "## Speech Style"]
|
|
for marker in bio_markers:
|
|
if marker in content:
|
|
# Insert before this section
|
|
idx = content.index(marker)
|
|
modified_messages[i]["content"] = (
|
|
content[:idx] + injection_text + "\n\n" + content[idx:]
|
|
)
|
|
injected = True
|
|
break
|
|
|
|
elif self.position == "system_suffix":
|
|
# Append to end of system message
|
|
modified_messages[i]["content"] = content + "\n\n" + injection_text
|
|
injected = True
|
|
|
|
if injected:
|
|
break
|
|
|
|
# Fallback: prepend to first user message if no system message found
|
|
if not injected and self.position == "before_conversation":
|
|
for i, msg in enumerate(modified_messages):
|
|
if msg.get("role") == "user":
|
|
content = msg.get("content", "")
|
|
modified_messages[i]["content"] = (
|
|
f"[Context for the NPC you're speaking with]\n{injection_text}\n\n"
|
|
f"[Player speaks]\n{content}"
|
|
)
|
|
injected = True
|
|
break
|
|
|
|
if injected:
|
|
logger.info(
|
|
"Injected lore",
|
|
npc_name=npc_profile.name,
|
|
entries_count=len(lore_entries),
|
|
position=self.position,
|
|
)
|
|
else:
|
|
logger.warning("Could not find injection point", position=self.position)
|
|
|
|
result = InjectionResult(
|
|
npc_profile=npc_profile,
|
|
lore_entries=lore_entries,
|
|
injection_text=injection_text if injected else "",
|
|
query_time_ms=query_time_ms,
|
|
)
|
|
|
|
return modified_messages, result
|
|
|
|
def _build_injection_text(
|
|
self,
|
|
npc_profile: NPCProfile,
|
|
lore_entries: list[LoreEntry],
|
|
) -> str:
|
|
"""Build the injection text block."""
|
|
# Build lore items list
|
|
lore_items = []
|
|
for entry in lore_entries:
|
|
# Truncate very long entries
|
|
content = entry.content
|
|
if len(content) > 300:
|
|
content = content[:297] + "..."
|
|
lore_items.append(f"- **{entry.topic}**: {content}")
|
|
|
|
lore_items_text = "\n".join(lore_items)
|
|
|
|
# Fill template
|
|
injection_text = self.template.format(
|
|
race=npc_profile.race or "person",
|
|
profession=npc_profile.profession or "citizen",
|
|
location=npc_profile.location or "Skyrim",
|
|
lore_items=lore_items_text,
|
|
name=npc_profile.name,
|
|
)
|
|
|
|
return injection_text.strip()
|