221 lines
9.2 KiB
Python
221 lines
9.2 KiB
Python
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)
|