import json import random import httpx from app.config import settings class AIBackend: def __init__(self): self.backend = settings.AI_BACKEND async def chat(self, messages: list[dict], temperature: float = 0.7, max_tokens: int = 512) -> str: if self.backend == "ollama": return await self._ollama_chat(messages, temperature, max_tokens) elif self.backend == "deepseek": return await self._deepseek_chat(messages, temperature, max_tokens) else: return "" async def _ollama_chat(self, messages: list[dict], temperature: float, max_tokens: int) -> str: try: async with httpx.AsyncClient(timeout=60) as client: resp = await client.post( f"{settings.OLLAMA_BASE_URL}/api/chat", json={ "model": settings.OLLAMA_MODEL, "messages": messages, "stream": False, "options": {"temperature": temperature, "num_predict": max_tokens}, }, ) data = resp.json() return data.get("message", {}).get("content", "") except Exception: return "" async def _deepseek_chat(self, messages: list[dict], temperature: float, max_tokens: int) -> str: try: async with httpx.AsyncClient(timeout=60) as client: resp = await client.post( f"{settings.DEEPSEEK_BASE_URL}/v1/chat/completions", headers={"Authorization": f"Bearer {settings.DEEPSEEK_API_KEY}"}, json={"model": settings.DEEPSEEK_MODEL, "messages": messages, "temperature": temperature, "max_tokens": max_tokens}, ) result = resp.json() return result["choices"][0]["message"]["content"] except Exception: return "" ai_backend = AIBackend() DM_INTROS = { "hardcore": "欢迎来到这场推理盛宴。真相就藏在你面前的线索之中，仔细观察，大胆推理。", "emotional": "这是一个关于人心的故事。有时候，真相并不重要，重要的是你在这个过程中感受到了什么。", "欢乐": "欢迎来到这场欢乐的聚会！记住，每个人都在演戏，但真相只有一个！", "恐怖": "黑暗中有什么在注视着你……准备好了吗？", } async def dm_generate(script: dict, phase_index: int, released_clues: list[str], context: str = "") -> dict: phases = script.get("phases", []) phase = phases[phase_index] if phase_index < len(phases) else phases[-1] title = script["title"] script_type = script.get("type", "hardcore") type_intro = DM_INTROS.get(script_type, "") clues_str = ", ".join(released_clues) if released_clues else "暂无" prompt = f"""你是一个剧本杀主持人（DM），请以主持人的身份发言。 当前剧本: {title} 当前阶段: {phase['name']} 阶段描述: {phase['description']} 已发放线索: {clues_str} 上下文: {context} 请简短发言（50字以内），推动剧情发展。""" result = await ai_backend.chat([{"role": "user", "content": prompt}]) if result: # Try to parse as JSON cleaned = result.strip().removeprefix("```json").removesuffix("```").strip() try: return json.loads(cleaned) except json.JSONDecodeError: pass return {"action": "narrate", "content": result} voice = phase["description"][:80] if phase_index == 0 and context == "opening": voice = f"{type_intro}\n\n欢迎来到【{title}】。{script.get('background', '')[:100]}" return {"action": "narrate", "content": f"【{phase['name']}】{voice}"} async def npc_generate(role: dict, background: str, phase_name: str, recent_messages: list[str], player_message: str = "", is_private: bool = False, known_clues: list[str] = None, phase_goal: str = "", all_roles: list[dict] = None) -> dict: personality = role.get("personality", role.get("name", "")) clues_str = " | ".join(known_clues[-5:]) if known_clues else "暂无线索" roles_str = "" if all_roles: rlist = [r["name"] for r in all_roles if r["id"] != role.get("id")] roles_str = "其他玩家: " + ", ".join(rlist) prompt = f"""你正在扮演一个剧本杀角色。请严格按照角色设定发言。 角色信息: - 姓名: {role['name']} - 公开身份: {role.get('publicProfile', '')} - 性格: {personality} - 你的目标: {role.get('goal', '')} - 你的秘密: {role.get('secretProfile', '')} 当前背景: {background[:80]}... 当前阶段: {phase_name} {("阶段目标: " + phase_goal) if phase_goal else ""} {roles_str} 你已知道的线索: {clues_str} 最近发言: {' | '.join(recent_messages[-4:])} {'【私聊】' + player_message if is_private and player_message else ''} {'【有人对你说】' + player_message if not is_private and player_message else ''} 【发言规则】 1. 以角色身份说人话，简短自然（30字以内） 2. 你的秘密绝对不能主动说出来 3. 被问到时可以撒谎、回避、转移话题 4. 结合已公开的线索来推理和回应 5. 如果你知道某些线索的真相（如你就是凶手），可以故意误导他人 直接输出你的角色发言内容，不要JSON、不要解释。""" result = await ai_backend.chat([{"role": "user", "content": prompt}]) if result: cleaned = result.strip().removeprefix("```json").removesuffix("```").strip() try: return json.loads(cleaned) except json.JSONDecodeError: pass if result.startswith("{"): return {"action": "speak", "content": result} return {"action": "speak", "content": result} return {"action": "speak", "content": _rule_npc_reply(role, player_message, is_private, known_clues)} def _rule_npc_reply(role: dict, player_message: str, is_private: bool, known_clues: list[str] = None) -> str: name = role["name"] personality = role.get("personality", "") goal = role.get("goal", "") clues = known_clues or [] if not player_message: if clues: replies = [ f"{name}沉思道：\"这些线索……我觉得需要重新梳理一下。\"", f"{name}说：\"我注意到了一些细节，但现在还不方便说。\"", f"{name}看了看法医报告：\"时间线和线索对不上，肯定有人撒谎。\"", ] return random.choice(replies) replies = [ f"{name}环顾四周，若有所思。", f"{name}清了清嗓子：\"各位，我觉得我们应该整理一下思路。\"", f"{name}沉默地看着大家。", f"{name}低声说：\"这件事没有那么简单……\"", ] return random.choice(replies) if "凶手" in player_message or "杀人" in player_message or "你杀" in player_message: denials = [ f"{name}脸色一变：\"你凭什么这么说？证据呢？\"", f"{name}冷笑一声：\"如果我是凶手，我还会坐在这里？\"", f"{name}摇头：\"我没有理由杀他。\"", ] return random.choice(denials) if "时间" in player_message or "在哪" in player_message or "案发" in player_message: times = [ f"{name}回忆道：\"我当时在……让我想想。\"", f"{name}说：\"那段时间我一个人在房间里。\"", f"{name}皱眉：\"我不太确定具体时间，但我确实听到了什么声音。\"", ] return random.choice(times) if "线索" in player_message or "证据" in player_message or "发现" in player_message: clues_replies = [ f"{name}点头：\"这个线索确实值得注意。\"", f"{name}沉思：\"但这个线索也可能是在误导我们。\"", f"{name}说：\"我也有一个发现，但还不确定是否相关。\"", ] return random.choice(clues_replies) if "知道" in player_message or "秘密" in player_message: secrets = [ f"{name}回避了你的目光：\"我什么也不知道。\"", f"{name}沉默了一会儿：\"每个人都有不想说的秘密，不是吗？\"", f"{name}说：\"我只能告诉你，事情不是你看到的那样。\"", ] return random.choice(secrets) if "动机" in player_message or "为什么" in player_message or "目的" in player_message: motives = [ f"{name}说：\"每个人都有自己的理由，但有些理由……\"", f"{name}耸耸肩：\"动机？也许我们需要先搞清楚发生了什么。\"", f"{name}看着你：\"你确定你想知道真正的动机？\"", ] return random.choice(motives) generic = [ f"{name}思考了一下：\"这个嘛，我说不好。\"", f"{name}回答：\"我不太确定，但我觉得我们应该继续调查。\"", f"{name}说：\"你说得有一定道理，但可能还有其他可能性。\"", f"{name}点头：\"有意思，继续说。\"", ] return random.choice(generic)