You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
117 lines
4.5 KiB
117 lines
4.5 KiB
from __future__ import annotations
|
|
import re
|
|
from fastapi import APIRouter, Request, HTTPException
|
|
from pydantic import BaseModel
|
|
|
|
from app.services.llm import LLMClient
|
|
from app.services.i18n import (
|
|
locale_from_request,
|
|
translate,
|
|
language_directive as _language_directive,
|
|
DEFAULT_LOCALE,
|
|
)
|
|
|
|
_THINK_BLOCK = re.compile(r"<think>.*?</think>", re.DOTALL)
|
|
|
|
|
|
def _strip_thinking(text: str, locale: str) -> str:
|
|
"""Strip LLM thinking blocks. The actual response content is what we
|
|
return to the client; the model's internal reasoning is never shown.
|
|
Falls back to a localised canned line when nothing is left after
|
|
stripping (rare, but happens when the model burns all tokens on thinking).
|
|
"""
|
|
cleaned = _THINK_BLOCK.sub("", text or "").strip()
|
|
if not cleaned:
|
|
return {
|
|
"en": "(no response)",
|
|
"zh": "(没有回应)",
|
|
}.get(locale, "(no response)")
|
|
return cleaned
|
|
|
|
router = APIRouter(prefix="/api")
|
|
|
|
|
|
class InterviewRequest(BaseModel):
|
|
question: str
|
|
|
|
|
|
@router.get("/simulation/{sim_id}/agents")
|
|
async def list_agents(sim_id: str, request: Request):
|
|
agents = await request.app.state.store.get_all_agents(sim_id)
|
|
return {"agents": [a.model_dump() for a in agents]}
|
|
|
|
|
|
@router.get("/simulation/{sim_id}/agent/{agent_id}")
|
|
async def get_agent(sim_id: str, agent_id: int, request: Request):
|
|
agent = await request.app.state.store.get_agent(sim_id, agent_id)
|
|
if not agent:
|
|
raise HTTPException(404, translate("agent_not_found", locale_from_request(request)))
|
|
return agent.model_dump()
|
|
|
|
|
|
@router.post("/simulation/{sim_id}/agent/{agent_id}/interview")
|
|
async def interview(sim_id: str, agent_id: int, req: InterviewRequest, request: Request):
|
|
store = request.app.state.store
|
|
llm: LLMClient = request.app.state.llm
|
|
locale = locale_from_request(request)
|
|
|
|
agent = await store.get_agent(sim_id, agent_id)
|
|
if not agent:
|
|
raise HTTPException(404, translate("agent_not_found", locale))
|
|
|
|
world_state = await store.get_world_state(sim_id)
|
|
world_name = world_state.blueprint.name if world_state else "the settlement"
|
|
|
|
head, tail = _language_directive(locale)
|
|
system = (
|
|
f"{head}\n\n"
|
|
f"You are {agent.name}, a {agent.age}-year-old {agent.role} in {world_name}.\n"
|
|
f"Your personality: honesty={agent.personality.honesty:.1f}, "
|
|
f"empathy={agent.personality.empathy:.1f}, "
|
|
f"confrontational={agent.personality.confrontational:.1f}\n"
|
|
f"Your core memories: {'; '.join(agent.core_memory[:5])}\n"
|
|
f"Your beliefs: {'; '.join(agent.beliefs[:5])}\n"
|
|
f"Your emotional state: {agent.emotional_state}\n\n"
|
|
"Someone approaches and asks you a question. Respond in character. "
|
|
"Be authentic to your personality. Reference specific events from your memory. "
|
|
"Keep it to 2-3 sentences.\n\n"
|
|
f"{tail}"
|
|
)
|
|
|
|
if agent.life_state:
|
|
life_lines = []
|
|
|
|
# Domain levels
|
|
for domain in ["finances", "career", "health"]:
|
|
val = getattr(agent.life_state, domain, 0.5)
|
|
if val < 0.3:
|
|
life_lines.append(f"Your {domain} situation is dire")
|
|
elif val < 0.5:
|
|
life_lines.append(f"Your {domain} is tight but you manage")
|
|
elif val > 0.7:
|
|
life_lines.append(f"Your {domain} is in good shape")
|
|
|
|
# Family
|
|
if agent.life_state.family:
|
|
family_strs = [f"{f.name} ({f.relation}, {f.age}, {f.status})" for f in agent.life_state.family]
|
|
life_lines.append(f"Your family: {', '.join(family_strs)}")
|
|
|
|
# Pressures
|
|
if agent.life_state.pressures:
|
|
pressure_strs = [p.description for p in agent.life_state.pressures[:3]]
|
|
life_lines.append(f"What weighs on you: {'; '.join(pressure_strs)}")
|
|
|
|
# Childhood
|
|
if agent.life_state.childhood_summary:
|
|
life_lines.append(f"Your upbringing: {agent.life_state.childhood_summary[:200]}")
|
|
|
|
if life_lines:
|
|
system += "\n\nYOUR LIFE SITUATION:\n" + "\n".join(life_lines)
|
|
system += "\n\nWhen answering, let your life situation color your responses naturally. Reference family, pressures, or your past when relevant."
|
|
|
|
response = await llm.generate(system=system, user=req.question, max_tokens=400)
|
|
cleaned = _strip_thinking(response, locale)
|
|
return {"response": cleaned, "emotional_state": agent.emotional_state}
|
|
|
|
|
|
# moved up to module top so it can be imported by tests / shared utilities
|
|
|