1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
| class NPCDialogueManager:
"""NPC对话管理器"""
def __init__(self, llm_client, memory_system):
self.llm = llm_client
self.memory = memory_system
self.conversation_sessions = {}
def start_conversation(self, npc_id, player_id):
"""开始对话"""
session_id = f"{npc_id}_{player_id}_{int(time.time())}"
self.conversation_sessions[session_id] = {
"npc_id": npc_id,
"player_id": player_id,
"messages": [],
"start_time": time.time(),
"state": "active"
}
# 获取NPC上下文
context = self.build_npc_context(npc_id)
return session_id, context
def process_input(self, session_id, player_input):
"""处理玩家输入"""
session = self.conversation_sessions[session_id]
# 1. 检索相关记忆
relevant_memories = self.memory.retrieve_memories(
session["npc_id"],
player_input
)
# 2. 构建提示
prompt = self.build_prompt(
session["npc_id"],
player_input,
session["messages"],
relevant_memories
)
# 3. LLM生成
try:
response = self.llm.generate(
prompt=prompt,
max_tokens=150,
temperature=0.8,
stop_sequences=["\n", "玩家:"]
)
npc_response = {
"text": response,
"timestamp": time.time(),
"memories_accessed": len(relevant_memories)
}
# 4. 更新对话历史
session["messages"].append({
"role": "player",
"content": player_input
})
session["messages"].append({
"role": "npc",
"content": response
})
# 5. 存储记忆
self.memory.store_memory(
session["npc_id"],
{
"content": f"玩家说: {player_input}\n我回应: {response}",
"type": "conversation",
"emotion": self.detect_emotion(response)
},
importance=self.calculate_importance(player_input)
)
return npc_response
except Exception as e:
# 降级处理
return self.fallback_response(session["npc_id"])
def build_prompt(self, npc_id, player_input, history, memories):
"""构建完整提示"""
prompt = f"""
{self.get_system_prompt(npc_id)}
相关记忆：
{self.format_memories(memories)}
对话历史：
{self.format_history(history)}
玩家说：{player_input}
你的回应：
"""
return prompt
def detect_emotion(self, text):
"""检测对话情感"""
# 简单情感检测
emotions = {
"开心": ["高兴", "哈哈", "太好了", "喜欢"],
"生气": ["气死", "讨厌", "滚", "烦"],
"悲伤": ["难过", "伤心", "可惜"],
"惊讶": ["什么？", "天哪", "真的"]
}
detected = "neutral"
for emotion, keywords in emotions.items():
if any(kw in text for kw in keywords):
detected = emotion
break
return detected
|