AI游戏开发与智能NPC完全教程
1. AI在游戏开发中的应用概述
游戏是AI技术最早、最深入的应用领域之一。从1997年Deep Blue击败国际象棋冠军，到AlphaStar在《星际争霸2》中达到大师水平，再到今天LLM驱动的NPC能够进行自由对话——AI正在彻底重塑游戏开发的面貌。
当前AI在游戏中的主要应用方向：
| 应用领域 | 典型技术 | 代表案例 |
|---|---|---|
| NPC行为 | 行为树、效用AI、LLM对话 | 《中土世界》复仇系统 |
| 寻路导航 | A*、NavMesh、Flow Field | 《星际争霸2》单位寻路 |
| 内容生成 | 程序化生成、GAN、扩散模型 | 《No Man's Sky》宇宙生成 |
| 难度调节 | 强化学习、玩家建模 | 《求生之路》AI导演 |
| 测试自动化 | 自动化脚本、强化学习 | 各大工作室QA自动化 |
| 反作弊 | 异常检测、行为分析 | 在线竞技游戏反外挂 |
2. 游戏AI基础
2.1 有限状态机（FSM）
有限状态机是最基础的游戏AI架构，通过定义状态和转换规则来控制NPC行为。
from typing import Callable, Dict, Optional
import time
class State:
def __init__(self, name: str, on_enter: Callable = None,
on_update: Callable = None, on_exit: Callable = None):
self.name = name
self.on_enter = on_enter or (lambda npc: None)
self.on_update = on_update or (lambda npc, dt: None)
self.on_exit = on_exit or (lambda npc: None)
class Transition:
def __init__(self, target_state: str, condition: Callable,
action: Callable = None):
self.target = target_state
self.condition = condition
self.action = action or (lambda npc: None)
class FiniteStateMachine:
def __init__(self, npc):
self.npc = npc
self.states: Dict[str, State] = {}
self.transitions: Dict[str, list] = {}
self.current_state: Optional[str] = None
def add_state(self, state: State):
self.states[state.name] = state
if state.name not in self.transitions:
self.transitions[state.name] = []
def add_transition(self, from_state: str, transition: Transition):
self.transitions[from_state].append(transition)
def set_initial(self, state_name: str):
self.current_state = state_name
self.states[state_name].on_enter(self.npc)
def update(self, dt: float):
if self.current_state is None:
return
# 检查转换条件
for t in self.transitions.get(self.current_state, []):
if t.condition(self.npc):
self.states[self.current_state].on_exit(self.npc)
t.action(self.npc)
self.current_state = t.target
self.states[self.current_state].on_enter(self.npc)
return
# 执行当前状态更新
self.states[self.current_state].on_update(self.npc, dt)
# 示例：守卫NPC的有限状态机
class GuardNPC:
def __init__(self, name: str, x: float = 0, y: float = 0):
self.name = name
self.x, self.y = x, y
self.health = 100
self.alert_level = 0
self.patrol_points = [(0, 0), (10, 0), (10, 10), (0, 10)]
self.current_patrol_idx = 0
self.target_x, self.target_y = None, None
self.speed = 2.0
self.sight_range = 8.0
self.enemy_nearby = False
def enter_patrol(npc):
print(f"[{npc.name}] 开始巡逻")
target = npc.patrol_points[npc.current_patrol_idx]
npc.target_x, npc.target_y = target
def update_patrol(npc, dt):
if npc.target_x is not None:
dx = npc.target_x - npc.x
dy = npc.target_y - npc.y
dist = (dx**2 + dy**2) ** 0.5
if dist < 0.5:
npc.current_patrol_idx = (npc.current_patrol_idx + 1) % len(npc.patrol_points)
target = npc.patrol_points[npc.current_patrol_idx]
npc.target_x, npc.target_y = target
print(f"[{npc.name}] 到达巡逻点，前往下一个: {target}")
else:
npc.x += (dx / dist) * npc.speed * dt
npc.y += (dy / dist) * npc.speed * dt
def enter_alert(npc):
print(f"[{npc.name}] ⚠️ 发现可疑目标！进入警戒状态")
npc.alert_level = 50
def update_alert(npc, dt):
npc.alert_level = max(0, npc.alert_level - 5 * dt)
if npc.alert_level <= 0:
npc.enemy_nearby = False
def enter_attack(npc):
print(f"[{npc.name}] ⚔️ 进入战斗！")
def update_attack(npc, dt):
npc.health -= 2 * dt # 模拟战斗损耗
# 构建FSM
guard = GuardNPC("守卫A")
fsm = FiniteStateMachine(guard)
fsm.add_state(State("patrol", on_enter=enter_patrol, on_update=update_patrol))
fsm.add_state(State("alert", on_enter=enter_alert, on_update=update_alert))
fsm.add_state(State("attack", on_enter=enter_attack, on_update=update_attack))
fsm.add_transition("patrol", Transition("alert", lambda n: n.enemy_nearby))
fsm.add_transition("alert", Transition("attack", lambda n: n.alert_level > 80))
fsm.add_transition("alert", Transition("patrol", lambda n: n.alert_level <= 0))
fsm.add_transition("attack", Transition("patrol", lambda n: n.health < 20))
fsm.set_initial("patrol")
# 模拟运行
guard.enemy_nearby = True
for i in range(10):
fsm.update(1.0)
2.2 行为树（Behavior Tree）
行为树比FSM更具扩展性和可维护性，是现代游戏AI的主流架构。
from enum import Enum
from typing import List, Callable, Optional
class NodeStatus(Enum):
SUCCESS = "success"
FAILURE = "failure"
RUNNING = "running"
class BTNode:
def __init__(self, name: str):
self.name = name
def tick(self, blackboard: dict) -> NodeStatus:
raise NotImplementedError
class Sequence(BTNode):
"""顺序节点：依次执行子节点，任一失败则失败"""
def __init__(self, name: str, children: List[BTNode]):
super().__init__(name)
self.children = children
def tick(self, blackboard: dict) -> NodeStatus:
for child in self.children:
status = child.tick(blackboard)
if status != NodeStatus.SUCCESS:
return status
return NodeStatus.SUCCESS
class Selector(BTNode):
"""选择节点：依次尝试子节点，任一成功则成功"""
def __init__(self, name: str, children: List[BTNode]):
super().__init__(name)
self.children = children
def tick(self, blackboard: dict) -> NodeStatus:
for child in self.children:
status = child.tick(blackboard)
if status != NodeStatus.FAILURE:
return status
return NodeStatus.FAILURE
class Condition(BTNode):
"""条件节点：检查黑板中的条件"""
def __init__(self, name: str, check: Callable):
super().__init__(name)
self.check = check
def tick(self, blackboard: dict) -> NodeStatus:
return NodeStatus.SUCCESS if self.check(blackboard) else NodeStatus.FAILURE
class Action(BTNode):
"""动作节点：执行具体行为"""
def __init__(self, name: str, action: Callable):
super().__init__(name)
self.action = action
def tick(self, blackboard: dict) -> NodeStatus:
return self.action(blackboard)
class Inverter(BTNode):
"""反转装饰器：反转子节点结果"""
def __init__(self, child: BTNode):
super().__init__(f"NOT({child.name})")
self.child = child
def tick(self, blackboard: dict) -> NodeStatus:
status = self.child.tick(blackboard)
if status == NodeStatus.SUCCESS:
return NodeStatus.FAILURE
elif status == NodeStatus.FAILURE:
return NodeStatus.SUCCESS
return NodeStatus.RUNNING
# 构建战斗NPC行为树
def check_health_low(bb):
return bb.get("health", 100) < 30
def check_has_enemy(bb):
return bb.get("enemy_visible", False)
def check_in_range(bb):
return bb.get("distance_to_enemy", 999) < 5.0
def action_flee(bb):
print("🏃 逃跑中...")
bb["health"] = min(100, bb["health"] + 5)
return NodeStatus.SUCCESS
def action_attack(bb):
print("⚔️ 攻击敌人！")
bb["enemy_health"] = bb.get("enemy_health", 100) - 15
return NodeStatus.SUCCESS
def action_patrol(bb):
print("🚶 巡逻中...")
return NodeStatus.SUCCESS
def action_approach(bb):
print("🏃 接近敌人...")
bb["distance_to_enemy"] = max(0, bb.get("distance_to_enemy", 10) - 3)
return NodeStatus.SUCCESS
# 组装行为树
combat_tree = Selector("根节点", [
# 优先级1：血量低时逃跑
Sequence("逃跑行为", [
Condition("血量低", check_health_low),
Action("逃跑", action_flee)
]),
# 优先级2：发现敌人时战斗
Sequence("战斗行为", [
Condition("有敌人", check_has_enemy),
Selector("战斗选择", [
Sequence("近战", [
Condition("在攻击范围内", check_in_range),
Action("攻击", action_attack)
]),
Action("接近敌人", action_approach)
])
]),
# 优先级3：默认巡逻
Action("巡逻", action_patrol)
])
# 测试行为树
blackboard = {"health": 80, "enemy_visible": True, "distance_to_enemy": 8, "enemy_health": 100}
print("=== 行为树执行 ===")
for tick in range(8):
print(f"\nTick {tick + 1} (HP:{blackboard['health']}, 敌距:{blackboard.get('distance_to_enemy', '-')})")
combat_tree.tick(blackboard)
2.3 效用系统（Utility AI）
效用系统通过为每个行为计算"效用值"来选择最优行为，更加灵活和可调优。
from typing import List, Callable
import math
class Consideration:
"""单一考量因素"""
def __init__(self, name: str, evaluator: Callable,
curve: str = "linear", params: dict = None):
self.name = name
self.evaluator = evaluator
self.curve = curve
self.params = params or {}
def score(self, blackboard: dict) -> float:
raw = self.evaluator(blackboard)
raw = max(0.0, min(1.0, raw)) # 归一化到[0,1]
if self.curve == "linear":
return raw
elif self.curve == "exponential":
exp = self.params.get("exponent", 2)
return raw ** exp
elif self.curve == "logistic":
midpoint = self.params.get("midpoint", 0.5)
slope = self.params.get("slope", 10)
return 1.0 / (1.0 + math.exp(-slope * (raw - midpoint)))
elif self.curve == "step":
threshold = self.params.get("threshold", 0.5)
return 1.0 if raw >= threshold else 0.0
return raw
class UtilityAction:
"""带效用评估的行动"""
def __init__(self, name: str, considerations: List[Consideration],
action: Callable, weight: float = 1.0):
self.name = name
self.considerations = considerations
self.action = action
self.weight = weight
def evaluate(self, blackboard: dict) -> float:
if not self.considerations:
return 0.0
scores = [c.score(blackboard) for c in self.considerations]
# 使用乘法组合（所有因素都重要）
combined = 1.0
for s in scores:
combined *= s
return combined * self.weight
class UtilityAI:
"""效用AI决策系统"""
def __init__(self, actions: List[UtilityAction]):
self.actions = actions
def decide(self, blackboard: dict) -> UtilityAction:
best_action = None
best_score = -1
for action in self.actions:
score = action.evaluate(blackboard)
if score > best_score:
best_score = score
best_action = action
return best_action
def rank_actions(self, blackboard: dict) -> List[tuple]:
scored = [(a, a.evaluate(blackboard)) for a in self.actions]
return sorted(scored, key=lambda x: x[1], reverse=True)
# 构建村民NPC效用AI
def eval_hunger(bb):
return bb.get("hunger", 0) / 100.0
def eval_threat(bb):
return 1.0 - (bb.get("threat_level", 0) / 100.0)
def eval_energy(bb):
return bb.get("energy", 100) / 100.0
def eval_resource_need(bb):
return 1.0 - (bb.get("resources", 50) / 100.0)
eat_action = UtilityAction("进食", [
Consideration("饥饿度", eval_hunger, curve="exponential", params={"exponent": 2}),
], lambda bb: print("🍖 进食中...") or "eating", weight=1.2)
flee_action = UtilityAction("逃离", [
Consideration("威胁程度", eval_threat, curve="step", params={"threshold": 0.3}),
], lambda bb: print("🏃 逃离危险！") or "fleeing", weight=2.0)
gather_action = UtilityAction("采集资源", [
Consideration("资源需求", eval_resource_need, curve="linear"),
Consideration("精力", eval_energy, curve="logistic", params={"midpoint": 0.3}),
Consideration("安全", eval_threat, curve="linear"),
], lambda bb: print("🪓 采集资源中...") or "gathering", weight=0.8)
rest_action = UtilityAction("休息", [
Consideration("精力低", lambda bb: 1.0 - bb.get("energy", 100) / 100.0,
curve="exponential", params={"exponent": 3}),
], lambda bb: print("😴 休息中...") or "resting", weight=0.9)
village_ai = UtilityAI([eat_action, flee_action, gather_action, rest_action])
# 测试不同场景
scenarios = [
{"name": "日常", "hunger": 20, "threat_level": 10, "energy": 80, "resources": 60},
{"name": "危险", "hunger": 30, "threat_level": 85, "energy": 70, "resources": 40},
{"name": "疲惫", "hunger": 50, "threat_level": 5, "energy": 10, "resources": 30},
{"name": "饥饿", "hunger": 90, "threat_level": 5, "energy": 60, "resources": 50},
]
for s in scenarios:
name = s.pop("name")
print(f"\n--- 场景: {name} ---")
rankings = village_ai.rank_actions(s)
for action, score in rankings:
bar = "█" * int(score * 20)
print(f" {action.name}: {score:.4f} {bar}")
best = village_ai.decide(s)
print(f" → 选择: {best.name}")
best.action(s)
3. 寻路算法
3.1 A*算法
A*是游戏中最常用的寻路算法，结合了Dijkstra算法的最优性和贪心搜索的效率。
import heapq
from typing import List, Tuple, Optional
class GridMap:
def __init__(self, width: int, height: int):
self.width = width
self.height = height
self.walls = set()
def add_wall(self, x: int, y: int):
self.walls.add((x, y))
def is_walkable(self, x: int, y: int) -> bool:
return (0 <= x < self.width and 0 <= y < self.height
and (x, y) not in self.walls)
def get_neighbors(self, x: int, y: int) -> List[Tuple[int, int]]:
neighbors = []
# 四方向 + 对角线
for dx, dy in [(-1,0),(1,0),(0,-1),(0,1),(-1,-1),(-1,1),(1,-1),(1,1)]:
nx, ny = x + dx, y + dy
if self.is_walkable(nx, ny):
# 对角线移动需要两个相邻格都可通行
if abs(dx) + abs(dy) == 2:
if self.is_walkable(x + dx, y) and self.is_walkable(x, y + dy):
neighbors.append((nx, ny))
else:
neighbors.append((nx, ny))
return neighbors
def render(self, path: list = None, start: tuple = None, end: tuple = None):
path_set = set(path) if path else set()
for y in range(self.height - 1, -1, -1):
row = ""
for x in range(self.width):
if (x, y) == start:
row += "S "
elif (x, y) == end:
row += "E "
elif (x, y) in path_set:
row += "* "
elif (x, y) in self.walls:
row += "█ "
else:
row += ". "
print(row)
def heuristic(a: Tuple[int, int], b: Tuple[int, int]) -> float:
"""对角线距离启发函数"""
dx = abs(a[0] - b[0])
dy = abs(a[1] - b[1])
return max(dx, dy) + (1.414 - 1) * min(dx, dy)
def astar(grid: GridMap, start: Tuple[int, int],
end: Tuple[int, int]) -> Optional[List[Tuple[int, int]]]:
open_set = []
heapq.heappush(open_set, (0, start))
came_from = {}
g_score = {start: 0}
f_score = {start: heuristic(start, end)}
while open_set:
current = heapq.heappop(open_set)[1]
if current == end:
# 重建路径
path = [current]
while current in came_from:
current = came_from[current]
path.append(current)
return path[::-1]
for neighbor in grid.get_neighbors(*current):
dx = abs(neighbor[0] - current[0])
dy = abs(neighbor[1] - current[1])
move_cost = 1.414 if (dx + dy == 2) else 1.0
tentative_g = g_score[current] + move_cost
if tentative_g < g_score.get(neighbor, float('inf')):
came_from[neighbor] = current
g_score[neighbor] = tentative_g
f_score[neighbor] = tentative_g + heuristic(neighbor, end)
heapq.heappush(open_set, (f_score[neighbor], neighbor))
return None # 无路径
# 使用示例
grid = GridMap(15, 10)
# 添加障碍物
for x in range(3, 10):
grid.add_wall(x, 5)
for y in range(2, 7):
grid.add_wall(7, y)
start = (1, 1)
end = (13, 8)
path = astar(grid, start, end)
if path:
print(f"路径长度: {len(path)} 步")
print(f"路径代价: {sum(1.414 if abs(path[i][0]-path[i-1][0])+abs(path[i][1]-path[i-1][1])==2 else 1.0 for i in range(1, len(path))):.2f}")
grid.render(path, start, end)
else:
print("未找到路径！")
3.2 导航网格（NavMesh）
NavMesh将可行走区域表示为凸多边形的集合，适用于复杂3D地形。
import math
from typing import List, Tuple, Optional
from dataclasses import dataclass
@dataclass
class Vec2:
x: float
y: float
def distance_to(self, other: 'Vec2') -> float:
return math.sqrt((self.x - other.x)**2 + (self.y - other.y)**2)
def __sub__(self, other):
return Vec2(self.x - other.x, self.y - other.y)
def __add__(self, other):
return Vec2(self.x + other.x, self.y + other.y)
def __mul__(self, scalar):
return Vec2(self.x * scalar, self.y * scalar)
@dataclass
class NavTriangle:
"""导航网格三角形"""
id: int
vertices: List[Vec2] # 3个顶点
center: Vec2
neighbors: List[int] # 相邻三角形ID
def __post_init__(self):
if not self.center:
self.center = Vec2(
sum(v.x for v in self.vertices) / 3,
sum(v.y for v in self.vertices) / 3
)
class SimpleNavMesh:
"""简化版导航网格"""
def __init__(self):
self.triangles: Dict[int, NavTriangle] = {}
def add_triangle(self, tri_id: int, vertices: List[Vec2]):
center = Vec2(
sum(v.x for v in vertices) / 3,
sum(v.y for v in vertices) / 3
)
tri = NavTriangle(tri_id, vertices, center, [])
self.triangles[tri_id] = tri
def build_connections(self):
"""基于共享边构建三角形连接关系"""
edges = {} # (sorted vertex tuple) -> triangle_id
for tri_id, tri in self.triangles.items():
verts = tri.vertices
for i in range(3):
v1, v2 = verts[i], verts[(i + 1) % 3]
edge_key = tuple(sorted([(v1.x, v1.y), (v2.x, v2.y)]))
if edge_key in edges:
neighbor_id = edges[edge_key]
if neighbor_id not in tri.neighbors:
tri.neighbors.append(neighbor_id)
if tri_id not in self.triangles[neighbor_id].neighbors:
self.triangles[neighbor_id].neighbors.append(tri_id)
else:
edges[edge_key] = tri_id
def find_path(self, start: Vec2, end: Vec2) -> Optional[List[Vec2]]:
"""在NavMesh上寻路"""
start_tri = self._find_triangle(start)
end_tri = self._find_triangle(end)
if start_tri is None or end_tri is None:
return None
# A*搜索三角形序列
open_set = [(0, start_tri)]
came_from = {}
g_score = {start_tri: 0}
while open_set:
current = heapq.heappop(open_set)[1]
if current == end_tri:
path = [end]
tri = end_tri
while tri in came_from:
tri = came_from[tri]
path.append(self.triangles[tri].center)
path.reverse()
path[0] = start
return path
for neighbor_id in self.triangles[current].neighbors:
neighbor = neighbor_id
cost = self.triangles[current].center.distance_to(
self.triangles[neighbor].center
)
tentative_g = g_score[current] + cost
if tentative_g < g_score.get(neighbor, float('inf')):
came_from[neighbor] = current
g_score[neighbor] = tentative_g
f = tentative_g + self.triangles[neighbor].center.distance_to(end)
heapq.heappush(open_set, (f, neighbor))
return None
def _find_triangle(self, point: Vec2) -> Optional[int]:
"""找到点所在的三角形"""
for tri_id, tri in self.triangles.items():
if self._point_in_triangle(point, tri.vertices):
return tri_id
return None
@staticmethod
def _point_in_triangle(p: Vec2, verts: List[Vec2]) -> bool:
"""使用重心坐标判断点是否在三角形内"""
v0, v1, v2 = verts
d00 = (v1.x - v0.x) * (v1.x - v0.x) + (v1.y - v0.y) * (v1.y - v0.y)
d01 = (v1.x - v0.x) * (v2.x - v0.x) + (v1.y - v0.y) * (v2.y - v0.y)
d11 = (v2.x - v0.x) * (v2.x - v0.x) + (v2.y - v0.y) * (v2.y - v0.y)
d20 = (p.x - v0.x) * (v1.x - v0.x) + (p.y - v0.y) * (v1.y - v0.y)
d21 = (p.x - v0.x) * (v2.x - v0.x) + (p.y - v0.y) * (v2.y - v0.y)
denom = d00 * d11 - d01 * d01
if abs(denom) < 1e-10:
return False
v = (d11 * d20 - d01 * d21) / denom
w = (d00 * d21 - d01 * d20) / denom
u = 1 - v - w
return (u >= 0) and (v >= 0) and (u + v <= 1)
# 构建简单的NavMesh
import heapq
mesh = SimpleNavMesh()
# 创建一个8x8网格的NavMesh（每个格子2个三角形）
for gy in range(4):
for gx in range(4):
base_id = (gy * 4 + gx) * 2
x0, y0 = gx * 2, gy * 2
# 左下三角形
mesh.add_triangle(base_id, [
Vec2(x0, y0), Vec2(x0+2, y0), Vec2(x0, y0+2)
])
# 右上三角形
mesh.add_triangle(base_id + 1, [
Vec2(x0+2, y0), Vec2(x0+2, y0+2), Vec2(x0, y0+2)
])
mesh.build_connections()
path = mesh.find_path(Vec2(1, 1), Vec2(7, 7))
if path:
print("NavMesh路径:")
for p in path:
print(f" ({p.x:.1f}, {p.y:.1f})")
4. 智能NPC行为设计
4.1 需求驱动的NPC系统
模拟NPC的基本需求（饥饿、疲劳、社交等），驱动自主行为。
import random
from dataclasses import dataclass, field
from typing import Dict, List
@dataclass
class Need:
name: str
value: float = 50.0 # 0-100
decay_rate: float = 1.0 # 每tick衰减
critical_threshold: float = 20.0
comfort_threshold: float = 60.0
def update(self, dt: float):
self.value = max(0, min(100, self.value - self.decay_rate * dt))
@property
def is_critical(self) -> bool:
return self.value < self.critical_threshold
@property
def is_comfortable(self) -> bool:
return self.value > self.comfort_threshold
@dataclass
class Personality:
"""NPC性格特征（大五人格模型）"""
openness: float = 0.5 # 开放性
conscientiousness: float = 0.5 # 尽责性
extraversion: float = 0.5 # 外向性
agreeableness: float = 0.5 # 宜人性
neuroticism: float = 0.5 # 神经质
@dataclass
class Memory:
"""NPC记忆系统"""
events: List[dict] = field(default_factory=list)
relationships: Dict[str, float] = field(default_factory=dict) # NPC ID -> 好感度
def add_event(self, event: dict):
self.events.append(event)
# 保留最近100条记忆
if len(self.events) > 100:
self.events = self.events[-100:]
def get_relationship(self, npc_id: str) -> float:
return self.relationships.get(npc_id, 0.0)
def adjust_relationship(self, npc_id: str, delta: float):
current = self.get_relationship(npc_id)
self.relationships[npc_id] = max(-100, min(100, current + delta))
class SmartNPC:
"""需求驱动的智能NPC"""
def __init__(self, name: str, personality: Personality = None):
self.name = name
self.personality = personality or Personality()
self.needs = {
"hunger": Need("饥饿", decay_rate=0.8),
"energy": Need("精力", decay_rate=0.5, critical_threshold=15),
"social": Need("社交", decay_rate=0.3),
"fun": Need("娱乐", decay_rate=0.4),
"safety": Need("安全", value=80, decay_rate=0.1),
}
self.memory = Memory()
self.current_action = "idle"
self.inventory = {"food": 3, "gold": 50}
self.position = (0, 0)
def update(self, dt: float):
"""每帧更新NPC状态"""
# 更新需求
for need in self.needs.values():
need.update(dt)
# 执行当前行为
self._execute_action(dt)
# 决策下一行为
self._decide_next_action()
def _decide_next_action(self):
"""基于需求优先级决定行为"""
# 找到最紧急的需求
urgent_needs = [
(name, need) for name, need in self.needs.items()
if need.is_critical
]
if urgent_needs:
urgent_needs.sort(key=lambda x: x[1].value)
most_urgent = urgent_needs[0][0]
self.current_action = self._get_action_for_need(most_urgent)
elif not all(n.is_comfortable for n in self.needs.values()):
# 舒适度不足，提升最弱的需求
weakest = min(self.needs.items(), key=lambda x: x[1].value)
self.current_action = self._get_action_for_need(weakest[0])
else:
# 所有需求满足，执行性格驱动的行为
self.current_action = self._personality_driven_action()
def _get_action_for_need(self, need_name: str) -> str:
action_map = {
"hunger": "eating",
"energy": "sleeping",
"social": "chatting",
"fun": "playing",
"safety": "seeking_shelter"
}
return action_map.get(need_name, "idle")
def _personality_driven_action(self) -> str:
p = self.personality
if p.extraversion > 0.7:
return "exploring"
elif p.conscientiousness > 0.7:
return "working"
elif p.openness > 0.7:
return "studying"
return "relaxing"
def _execute_action(self, dt: float):
action_effects = {
"eating": {"hunger": 15 * dt},
"sleeping": {"energy": 20 * dt},
"chatting": {"social": 12 * dt, "fun": 5 * dt},
"playing": {"fun": 15 * dt, "energy": -3 * dt},
"seeking_shelter": {"safety": 10 * dt},
"exploring": {"fun": 8 * dt, "energy": -2 * dt},
"working": {"hunger": -3 * dt, "energy": -4 * dt},
"studying": {"fun": 5 * dt, "energy": -2 * dt},
"relaxing": {"energy": 5 * dt, "fun": 3 * dt},
}
effects = action_effects.get(self.current_action, {})
for need_name, delta in effects.items():
if need_name in self.needs:
self.needs[need_name].value = max(0, min(100,
self.needs[need_name].value + delta))
def status_report(self) -> str:
lines = [f"[{self.name}] 行为: {self.current_action}"]
for name, need in self.needs.items():
bar_len = int(need.value / 5)
bar = "█" * bar_len + "░" * (20 - bar_len)
status = "⚠️" if need.is_critical else "✅" if need.is_comfortable else " "
lines.append(f" {status} {name}: [{bar}] {need.value:.1f}")
return "\n".join(lines)
# 模拟NPC一天的生活
npc = SmartNPC("铁匠汉斯", Personality(
openness=0.3, conscientiousness=0.8, extraversion=0.4,
agreeableness=0.6, neuroticism=0.3
))
print("=== 模拟NPC一天 ===")
for hour in range(24):
npc.update(2.0) # 每步2小时
if hour % 4 == 0:
print(f"\n--- 第{hour}小时 ---")
print(npc.status_report())
5. LLM驱动的NPC对话系统
5.1 上下文感知的对话引擎
from dataclasses import dataclass, field
from typing import List, Dict, Optional
import json
@dataclass
class DialogueContext:
"""对话上下文管理"""
npc_name: str
npc_role: str
personality: str
knowledge: List[str]
current_mood: str = "neutral"
relationship_level: int = 50 # 0-100
conversation_history: List[dict] = field(default_factory=list)
world_state: Dict = field(default_factory=dict)
def add_message(self, role: str, content: str):
self.conversation_history.append({"role": role, "content": content})
# 保留最近20轮对话
if len(self.conversation_history) > 40:
self.conversation_history = self.conversation_history[-40:]
def build_system_prompt(self) -> str:
knowledge_text = "\n".join(f"- {k}" for k in self.knowledge)
return f"""你是游戏中的NPC角色。
名字: {self.npc_name}
身份: {self.npc_role}
性格: {self.personality}
当前心情: {self.current_mood}
与玩家关系: {self.relationship_level}/100
你知道的信息:
{knowledge_text}
世界状态:
{json.dumps(self.world_state, ensure_ascii=False, indent=2)}
行为准则:
1. 始终保持角色一致，不要跳出角色
2. 根据与玩家的关系调整语气（关系好则友善，关系差则冷淡）
3. 只分享你知道的信息，不要编造不存在的知识
4. 回复简洁，每次对话不超过3句话
5. 可以拒绝不当请求或提出交换条件"""
class LLMDialogueEngine:
"""LLM驱动的NPC对话引擎"""
def __init__(self):
self.npcs: Dict[str, DialogueContext] = {}
def register_npc(self, context: DialogueContext):
self.npcs[context.npc_name] = context
def generate_prompt(self, npc_name: str, player_message: str) -> dict:
"""生成发送给LLM的完整prompt"""
ctx = self.npcs.get(npc_name)
if not ctx:
return {"error": f"NPC '{npc_name}' not found"}
ctx.add_message("user", player_message)
messages = [{"role": "system", "content": ctx.build_system_prompt()}]
messages.extend(ctx.conversation_history)
return {
"messages": messages,
"temperature": 0.8,
"max_tokens": 150,
"stop": ["\n\n"]
}
def process_response(self, npc_name: str, llm_response: str) -> str:
"""处理LLM回复，更新NPC状态"""
ctx = self.npcs[npc_name]
ctx.add_message("assistant", llm_response)
# 简单的情感分析（实际应用中可用更复杂的模型）
negative_words = ["讨厌", "滚", "不", "别", "烦", "恨"]
positive_words = ["谢谢", "好的", "喜欢", "开心", "欢迎"]
for word in positive_words:
if word in llm_response:
ctx.current_mood = "happy"
ctx.relationship_level = min(100, ctx.relationship_level + 2)
break
for word in negative_words:
if word in llm_response:
ctx.current_mood = "annoyed"
ctx.relationship_level = max(0, ctx.relationship_level - 2)
break
return llm_response
# 使用示例
engine = LLMDialogueEngine()
# 注册铁匠NPC
blacksmith = DialogueContext(
npc_name="铁匠汉斯",
npc_role="村庄铁匠，负责打造武器和修理装备",
personality="粗犷但善良，喜欢喝酒，讨厌小偷。说话直接，偶尔幽默。",
knowledge=[
"最近北方的矿山出现了怪物，矿工们不敢去了",
"最好的铁矿石来自龙脊山脉",
"村长的女儿上周失踪了，村长很着急",
"东边的森林里有一把传说中的剑，但没人找到过",
"自己的锤子是祖传的，从不借给别人"
],
current_mood="neutral",
relationship_level=50,
world_state={
"time_of_day": "下午",
"weather": "阴天",
"village_alert_level": "低",
"recent_events": ["北方矿山出现怪物", "村长女儿失踪"]
}
)
engine.register_npc(blacksmith)
# 模拟对话
conversations = [
"你好，我想买一把铁剑",
"北方矿山的怪物是怎么回事？",
"你能帮我打造一把特殊的武器吗？",
"你知道村长女儿去哪了吗？",
]
print("=== 铁匠汉斯对话系统 ===\n")
for msg in conversations:
print(f"玩家: {msg}")
prompt = engine.generate_prompt("铁匠汉斯", msg)
# 模拟LLM回复（实际使用时替换为真实API调用）
simulated_responses = {
"你好，我想买一把铁剑": "哦，买剑啊？普通的15金币，精钢的30。你带够钱了吗？",
"北方矿山的怪物是怎么回事？": "听说是些大虫子，黑乎乎的，有牛那么大。矿工老张差点没命跑回来。你可别去送死。",
"你能帮我打造一把特殊的武器吗？": "特殊的？那得看你有什么材料了。龙脊山脉的黑铁石能打出好东西，但那地方可不好去。",
"你知道村长女儿去哪了吗？": "唉，这事整个村子都在议论。有人说看见她往东边森林去了...那地方邪门，希望她没事吧。",
}
response = simulated_responses.get(msg, "...嗯，让我想想。")
engine.process_response("铁匠汉斯", response)
ctx = engine.npcs["铁匠汉斯"]
print(f"汉斯: {response}")
print(f" [心情:{ctx.current_mood} 关系:{ctx.relationship_level}]\n")
6. 程序化内容生成（PCG）
6.1 地牢生成算法
import random
from typing import List, Tuple, Optional
class Room:
def __init__(self, x: int, y: int, w: int, h: int):
self.x, self.y = x, y
self.w, self.h = w, h
@property
def center(self) -> Tuple[int, int]:
return (self.x + self.w // 2, self.y + self.h // 2)
def intersects(self, other: 'Room', padding: int = 1) -> bool:
return not (self.x + self.w + padding <= other.x or
other.x + other.w + padding <= self.x or
self.y + self.h + padding <= other.y or
other.y + other.h + padding <= self.y)
class DungeonGenerator:
"""程序化地牢生成器"""
FLOOR = '.'
WALL = '#'
DOOR = '+'
CORRIDOR = '~'
CHEST = 'C'
SPAWN = 'S'
def __init__(self, width: int = 60, height: int = 40, seed: int = None):
self.width = width
self.height = height
self.grid = [[self.WALL] * width for _ in range(height)]
self.rooms: List[Room] = []
self.corridors: List[List[Tuple[int, int]]] = []
random.seed(seed)
def generate(self, n_rooms: int = 8, min_size: int = 4, max_size: int = 10,
max_attempts: int = 200) -> 'DungeonGenerator':
# 1. 放置房间
for _ in range(max_attempts):
if len(self.rooms) >= n_rooms:
break
w = random.randint(min_size, max_size)
h = random.randint(min_size, max_size)
x = random.randint(1, self.width - w - 1)
y = random.randint(1, self.height - h - 1)
room = Room(x, y, w, h)
if not any(room.intersects(r) for r in self.rooms):
self.rooms.append(room)
self._carve_room(room)
# 2. 连接房间（最小生成树 + 额外边）
self._connect_rooms()
# 3. 放置物品
self._place_objects()
return self
def _carve_room(self, room: Room):
for y in range(room.y, room.y + room.h):
for x in range(room.x, room.x + room.w):
self.grid[y][x] = self.FLOOR
def _connect_rooms(self):
if len(self.rooms) < 2:
return
# 简化版：按房间中心距离排序连接
connected = {0}
unconnected = set(range(1, len(self.rooms)))
while unconnected:
best_dist = float('inf')
best_pair = None
for c in connected:
for u in unconnected:
c1 = self.rooms[c].center
c2 = self.rooms[u].center
dist = abs(c1[0] - c2[0]) + abs(c1[1] - c2[1])
if dist < best_dist:
best_dist = dist
best_pair = (c, u)
if best_pair:
c, u = best_pair
self._carve_corridor(self.rooms[c].center, self.rooms[u].center)
connected.add(u)
unconnected.remove(u)
# 添加一些额外连接增加多样性
for _ in range(len(self.rooms) // 3):
a, b = random.sample(range(len(self.rooms)), 2)
self._carve_corridor(self.rooms[a].center, self.rooms[b].center)
def _carve_corridor(self, start: Tuple[int, int], end: Tuple[int, int]):
x, y = start
corridor = [(x, y)]
# L形走廊
if random.random() < 0.5:
# 先水平再垂直
while x != end[0]:
x += 1 if end[0] > x else -1
self.grid[y][x] = self.CORRIDOR if self.grid[y][x] == self.WALL else self.grid[y][x]
corridor.append((x, y))
while y != end[1]:
y += 1 if end[1] > y else -1
self.grid[y][x] = self.CORRIDOR if self.grid[y][x] == self.WALL else self.grid[y][x]
corridor.append((x, y))
else:
while y != end[1]:
y += 1 if end[1] > y else -1
self.grid[y][x] = self.CORRIDOR if self.grid[y][x] == self.WALL else self.grid[y][x]
corridor.append((x, y))
while x != end[0]:
x += 1 if end[0] > x else -1
self.grid[y][x] = self.CORRIDOR if self.grid[y][x] == self.WALL else self.grid[y][x]
corridor.append((x, y))
self.corridors.append(corridor)
def _place_objects(self):
# 在随机房间放置宝箱
for room in random.sample(self.rooms, min(3, len(self.rooms))):
x = random.randint(room.x + 1, room.x + room.w - 2)
y = random.randint(room.y + 1, room.y + room.h - 2)
self.grid[y][x] = self.CHEST
# 在第一个房间放置出生点
if self.rooms:
cx, cy = self.rooms[0].center
self.grid[cy][cx] = self.SPAWN
def render(self) -> str:
return "\n".join("".join(row) for row in self.grid)
def get_tile(self, x: int, y: int) -> str:
if 0 <= x < self.width and 0 <= y < self.height:
return self.grid[y][x]
return self.WALL
# 生成地牢
dungeon = DungeonGenerator(60, 30, seed=42).generate(n_rooms=8)
print(dungeon.render())
print(f"\n房间数: {len(dungeon.rooms)}")
for i, room in enumerate(dungeon.rooms):
print(f" 房间{i}: ({room.x},{room.y}) {room.w}x{room.h} 中心{room.center}")
7. 游戏测试自动化
import random
from dataclasses import dataclass, field
from typing import List, Callable
@dataclass
class TestResult:
test_name: str
passed: bool
duration_ms: float
details: str = ""
class GameTestAgent:
"""AI游戏测试代理"""
def __init__(self, game_state: dict):
self.game_state = game_state
self.actions_log: List[str] = []
self.bugs_found: List[dict] = []
self.position = game_state.get("player_pos", [0, 0])
def execute_action(self, action: str) -> dict:
"""执行游戏动作并返回结果"""
self.actions_log.append(action)
if action == "move_north":
self.position[1] += 1
elif action == "move_south":
self.position[1] -= 1
elif action == "move_east":
self.position[0] += 1
elif action == "move_west":
self.position[0] -= 1
elif action == "attack":
return {"type": "combat", "damage_dealt": random.randint(10, 30)}
elif action == "interact":
return {"type": "interaction", "success": random.random() > 0.3}
return {"type": "movement", "position": self.position.copy()}
def check_invariants(self) -> List[TestResult]:
"""检查游戏不变量"""
results = []
# 检查玩家位置合法性
pos = self.position
in_bounds = (0 <= pos[0] < self.game_state.get("map_width", 100) and
0 <= pos[1] < self.game_state.get("map_height", 100))
results.append(TestResult("边界检查", in_bounds, 0.5,
f"位置{pos}" if in_bounds else f"越界! 位置{pos}"))
# 检查生命值
hp = self.game_state.get("player_hp", 100)
hp_valid = 0 <= hp <= self.game_state.get("max_hp", 100)
results.append(TestResult("生命值范围", hp_valid, 0.2,
f"HP={hp}" if hp_valid else f"HP异常: {hp}"))
# 检查资源
gold = self.game_state.get("gold", 0)
gold_valid = gold >= 0
results.append(TestResult("金币非负", gold_valid, 0.1,
f"金币={gold}" if gold_valid else f"金币为负: {gold}"))
return results
def fuzz_test(self, n_actions: int = 100) -> dict:
"""模糊测试：随机执行动作并检查游戏状态"""
actions = ["move_north", "move_south", "move_east", "move_west",
"attack", "interact"]
bugs = 0
total_checks = 0
for i in range(n_actions):
action = random.choice(actions)
self.execute_action(action)
results = self.check_invariants()
for r in results:
total_checks += 1
if not r.passed:
bugs += 1
self.bugs_found.append({
"action_idx": i,
"action": action,
"test": r.test_name,
"detail": r.details
})
return {
"actions_executed": n_actions,
"checks_performed": total_checks,
"bugs_found": bugs,
"bug_rate": bugs / total_checks if total_checks > 0 else 0,
"bugs": self.bugs_found[:10] # 最多显示10个
}
# 使用示例
game = {
"player_pos": [50, 50],
"map_width": 100,
"map_height": 100,
"player_hp": 80,
"max_hp": 100,
"gold": 200
}
agent = GameTestAgent(game)
report = agent.fuzz_test(500)
print(f"=== 游戏测试报告 ===")
print(f"执行动作: {report['actions_executed']}")
print(f"检查次数: {report['checks_performed']}")
print(f"发现Bug: {report['bugs_found']}")
print(f"Bug率: {report['bug_rate']:.2%}")
if report['bugs']:
print("\nBug详情:")
for bug in report['bugs']:
print(f" [{bug['action']}] {bug['test']}: {bug['detail']}")
8. 玩家行为分析与个性化
from collections import defaultdict
from typing import Dict, List
import math
class PlayerProfiler:
"""玩家行为画像系统"""
def __init__(self, player_id: str):
self.player_id = player_id
self.action_counts: Dict[str, int] = defaultdict(int)
self.session_durations: List[float] = []
self.difficulty_choices: List[float] = []
self.death_locations: List[tuple] = []
self.achievement_times: Dict[str, float] = {}
self.total_playtime = 0.0
self.sessions = 0
def record_action(self, action_type: str, details: dict = None):
self.action_counts[action_type] += 1
def record_session(self, duration_minutes: float):
self.session_durations.append(duration_minutes)
self.total_playtime += duration_minutes
self.sessions += 1
def get_play_style(self) -> str:
"""分析玩家风格"""
total = sum(self.action_counts.values())
if total == 0:
return "unknown"
combat_ratio = (self.action_counts.get("attack", 0) +
self.action_counts.get("combat", 0)) / total
explore_ratio = (self.action_counts.get("explore", 0) +
self.action_counts.get("discover", 0)) / total
social_ratio = (self.action_counts.get("chat", 0) +
self.action_counts.get("trade", 0)) / total
craft_ratio = self.action_counts.get("craft", 0) / total
ratios = {
"combat": combat_ratio,
"explorer": explore_ratio,
"social": social_ratio,
"craftsman": craft_ratio
}
return max(ratios, key=ratios.get)
def get_engagement_score(self) -> float:
"""计算玩家参与度评分 0-100"""
if self.sessions == 0:
return 0
score = 0.0
# 频率分（每周session数）
avg_sessions = self.sessions # 简化
score += min(30, avg_sessions * 5)
# 时长分（平均session时长）
if self.session_durations:
avg_duration = sum(self.session_durations) / len(self.session_durations)
score += min(30, avg_duration / 2)
# 多样性分（使用了多少种动作类型）
diversity = len(self.action_counts) / 10.0
score += min(20, diversity * 20)
# 留存分（session间隔是否稳定）
if len(self.session_durations) > 2:
score += 20
return min(100, score)
def recommend_content(self) -> List[str]:
"""基于玩家画像推荐内容"""
style = self.get_play_style()
recommendations = {
"combat": ["新Boss副本", "竞技场赛季", "武器强化材料"],
"explorer": ["隐藏地图区域", "收集品成就", "世界Boss"],
"social": ["公会活动", "交易市场", "组队副本"],
"craftsman": ["新配方图纸", "稀有材料矿点", "制作成就"],
}
return recommendations.get(style, ["每日任务", "新手引导"])
def get_difficulty_adjustment(self) -> float:
"""动态难度调节建议"""
if not self.death_locations:
return 1.0
death_rate = len(self.death_locations) / max(1, self.total_playtime / 60)
if death_rate > 5:
return 0.7 # 降低难度
elif death_rate > 2:
return 0.85
elif death_rate < 0.5:
return 1.2 # 提高难度
return 1.0
# 使用示例
player = PlayerProfiler("player_001")
# 模拟玩家行为
for _ in range(50):
player.record_action("attack")
for _ in range(30):
player.record_action("explore")
for _ in range(20):
player.record_action("chat")
for _ in range(10):
player.record_action("craft")
for _ in range(5):
player.record_action("trade")
for _ in range(15):
player.record_action("discover")
player.record_session(45)
player.record_session(90)
player.record_session(30)
print("=== 玩家画像分析 ===")
print(f"玩家ID: {player.player_id}")
print(f"游戏风格: {player.get_play_style()}")
print(f"参与度: {player.get_engagement_score():.1f}/100")
print(f"难度调节: {player.get_difficulty_adjustment():.2f}x")
print(f"推荐内容: {player.recommend_content()}")
print(f"\n行为统计:")
for action, count in sorted(player.action_counts.items(), key=lambda x: -x[1]):
bar = "█" * (count // 2)
print(f" {action}: {count} {bar}")
9. 强化学习游戏AI训练
import numpy as np
import random
from collections import defaultdict
class SimpleGridWorld:
"""简单的网格世界环境"""
ACTIONS = ["up", "down", "left", "right"]
def __init__(self, width=5, height=5):
self.width = width
self.height = height
self.agent_pos = [0, 0]
self.goal_pos = [width-1, height-1]
self.walls = set()
self.rewards = defaultdict(float)
self.rewards[tuple(self.goal_pos)] = 100
self.max_steps = 100
self.steps = 0
def add_wall(self, x, y):
self.walls.add((x, y))
self.rewards[(x, y)] = -50
def reset(self):
self.agent_pos = [0, 0]
self.steps = 0
return self._get_state()
def step(self, action_idx):
self.steps += 1
action = self.ACTIONS[action_idx]
dx, dy = {"up": (0, -1), "down": (0, 1),
"left": (-1, 0), "right": (1, 0)}[action]
new_x = max(0, min(self.width-1, self.agent_pos[0] + dx))
new_y = max(0, min(self.height-1, self.agent_pos[1] + dy))
if (new_x, new_y) not in self.walls:
self.agent_pos = [new_x, new_y]
state = self._get_state()
pos = tuple(self.agent_pos)
reward = self.rewards.get(pos, -1) # 每步-1鼓励快速到达
done = (pos == tuple(self.goal_pos)) or (self.steps >= self.max_steps)
return state, reward, done
def _get_state(self):
return (self.agent_pos[0], self.agent_pos[1])
class QLearningAgent:
"""Q-Learning智能体"""
def __init__(self, n_actions=4, learning_rate=0.1,
discount_factor=0.95, epsilon=1.0, epsilon_decay=0.995):
self.q_table = defaultdict(lambda: np.zeros(n_actions))
self.n_actions = n_actions
self.lr = learning_rate
self.gamma = discount_factor
self.epsilon = epsilon
self.epsilon_decay = epsilon_decay
self.epsilon_min = 0.01
def choose_action(self, state):
if random.random() < self.epsilon:
return random.randint(0, self.n_actions - 1)
return int(np.argmax(self.q_table[state]))
def learn(self, state, action, reward, next_state, done):
current_q = self.q_table[state][action]
if done:
target = reward
else:
target = reward + self.gamma * np.max(self.q_table[next_state])
self.q_table[state][action] += self.lr * (target - current_q)
def decay_epsilon(self):
self.epsilon = max(self.epsilon_min, self.epsilon * self.epsilon_decay)
# 训练过程
env = SimpleGridWorld(6, 6)
env.add_wall(2, 1)
env.add_wall(2, 2)
env.add_wall(2, 3)
env.add_wall(4, 1)
env.add_wall(4, 2)
agent = QLearningAgent()
episodes = 1000
rewards_history = []
steps_history = []
for ep in range(episodes):
state = env.reset()
total_reward = 0
steps = 0
while True:
action = agent.choose_action(state)
next_state, reward, done = env.step(action)
agent.learn(state, action, reward, next_state, done)
state = next_state
total_reward += reward
steps += 1
if done:
break
agent.decay_epsilon()
rewards_history.append(total_reward)
steps_history.append(steps)
# 展示训练结果
window = 100
print("=== Q-Learning训练进度 ===")
for i in range(0, episodes, window):
avg_r = sum(rewards_history[i:i+window]) / window
avg_s = sum(steps_history[i:i+window]) / window
bar = "█" * max(0, int((avg_r + 50) / 3))
print(f"Episode {i:4d}: 平均奖励={avg_r:6.1f} 平均步数={avg_s:5.1f} {bar}")
# 展示学到的策略
print("\n学到的策略 (↑↓←→):")
for y in range(env.height):
row = ""
for x in range(env.width):
if (x, y) in env.walls:
row += " ██ "
elif (x, y) == tuple(env.goal_pos):
row += " 🎯 "
else:
state = (x, y)
best_action = np.argmax(agent.q_table[state])
symbols = [" ↑ ", " ↓ ", " ← ", " → "]
row += symbols[best_action]
print(row)
10. 实战案例：LLM驱动的开放世界NPC
将前面各节技术整合为一个完整的NPC系统。
from dataclasses import dataclass, field
from typing import Dict, List, Optional
import random
@dataclass
class WorldEvent:
name: str
description: str
affected_locations: List[str]
duration_ticks: int
impact: Dict[str, float] # 需求影响
class NPCPersonality:
def __init__(self, traits: Dict[str, float]):
self.traits = traits # trait_name -> 0.0~1.0
def get(self, trait: str, default: float = 0.5) -> float:
return self.traits.get(trait, default)
class Memory:
def __init__(self):
self.short_term: List[dict] = [] # 最近事件
self.long_term: List[dict] = [] # 重要事件
self.knowledge: Dict[str, any] = {} # 世界知识
def observe(self, event: dict, importance: float = 0.5):
self.short_term.append(event)
if importance > 0.7:
self.long_term.append(event)
if len(self.short_term) > 20:
self.short_term = self.short_term[-20:]
def recall_about(self, topic: str) -> List[dict]:
all_memories = self.short_term + self.long_term
return [m for m in all_memories if topic.lower() in str(m).lower()]
class OpenWorldNPC:
"""开放世界NPC，整合行为、对话和记忆"""
def __init__(self, name: str, role: str, location: str,
personality: NPCPersonality, knowledge: List[str]):
self.name = name
self.role = role
self.location = location
self.personality = personality
self.memory = Memory()
self.needs = {"hunger": 50, "energy": 70, "social": 40, "safety": 80}
self.mood = "neutral"
self.dialogue_history: List[dict] = []
self.daily_routine = self._generate_routine()
self.current_hour = 8
for k in knowledge:
self.memory.knowledge[k] = True
def _generate_routine(self) -> Dict[int, str]:
"""根据角色生成日程"""
routines = {
"blacksmith": {8: "open_shop", 12: "lunch", 13: "open_shop",
18: "tavern", 22: "sleep"},
"merchant": {7: "prepare_goods", 9: "market", 12: "lunch",
14: "market", 19: "home", 22: "sleep"},
"guard": {6: "patrol", 10: "break", 11: "patrol", 15: "training",
18: "patrol", 22: "guard_post"},
"farmer": {5: "farm", 8: "breakfast", 9: "farm", 12: "lunch",
13: "farm", 18: "market", 20: "home"},
}
template = routines.get(self.role, {8: "idle", 20: "sleep"})
return template
def tick(self, world_events: List[WorldEvent] = None):
"""每个游戏时tick更新"""
self.current_hour = (self.current_hour + 1) % 24
# 更新需求
self.needs["hunger"] = max(0, self.needs["hunger"] - 2)
self.needs["energy"] = max(0, self.needs["energy"] - 1)
# 执行日程行为
current_action = "idle"
for hour in sorted(self.daily_routine.keys(), reverse=True):
if self.current_hour >= hour:
current_action = self.daily_routine[hour]
break
# 行为对需求的影响
action_effects = {
"lunch": {"hunger": 30}, "breakfast": {"hunger": 25},
"sleep": {"energy": 40}, "break": {"energy": 10},
"tavern": {"social": 20, "hunger": 15},
"home": {"energy": 5, "safety": 10},
}
for need, delta in action_effects.get(current_action, {}).items():
self.needs[need] = min(100, self.needs[need] + delta)
# 处理世界事件
if world_events:
for event in world_events:
if self.location in event.affected_locations:
self.memory.observe({
"type": "world_event",
"event": event.name,
"description": event.description,
"hour": self.current_hour
}, importance=0.8)
for need, delta in event.impact.items():
if need in self.needs:
self.needs[need] = max(0, min(100, self.needs[need] + delta))
# 更新心情
avg_needs = sum(self.needs.values()) / len(self.needs)
if avg_needs > 70:
self.mood = "happy"
elif avg_needs > 40:
self.mood = "neutral"
else:
self.mood = "stressed"
def generate_dialogue_context(self, player_message: str) -> dict:
"""生成LLM对话上下文"""
relevant_memories = []
for word in player_message.split():
memories = self.memory.recall_about(word)
relevant_memories.extend(memories[:2])
return {
"system": {
"npc_name": self.name,
"role": self.role,
"location": self.location,
"personality": self.personality.traits,
"mood": self.mood,
"needs": self.needs,
"time": f"{self.current_hour}:00",
"knowledge": list(self.memory.knowledge.keys()),
"recent_memories": self.memory.short_term[-5:],
"relevant_memories": relevant_memories[:3]
},
"guidelines": [
"保持角色一致，不要跳出角色",
"根据心情调整语气",
"根据记忆中的事件自然地提及",
"回复简洁，不超过3句话"
]
}
def status(self) -> str:
lines = [f"【{self.name}】{self.role} @ {self.location}"]
lines.append(f" 时间:{self.current_hour}:00 心情:{self.mood}")
lines.append(f" 需求: " + " | ".join(
f"{k}:{v:.0f}" for k, v in self.needs.items()))
return "\n".join(lines)
# 模拟开放世界
print("=== 开放世界NPC模拟 ===\n")
# 创建NPC
npcs = [
OpenWorldNPC("汉斯", "blacksmith", "village",
NPCPersonality({"friendliness": 0.7, "courage": 0.6, "greed": 0.3}),
["龙脊山脉有好铁矿", "最近北方有怪物出没", "村长女儿失踪了"]),
OpenWorldNPC("玛利亚", "merchant", "market",
NPCPersonality({"friendliness": 0.8, "courage": 0.3, "greed": 0.7}),
["商队来自南方城市", "最近物价上涨了", "森林里有条秘密小路"]),
OpenWorldNPC("埃里克", "guard", "gate",
NPCPersonality({"friendliness": 0.4, "courage": 0.9, "greed": 0.2}),
["城门关禁时间是晚上10点", "可疑人物最近出没", "东边森林不安全"]),
]
# 模拟世界事件
events = [
WorldEvent("暴风雨", "一场猛烈的暴风雨袭击了村庄",
["village", "market"], 4, {"safety": -20, "energy": -10}),
WorldEvent("怪物袭击", "北方的怪物向南迁徙",
["gate", "village"], 8, {"safety": -30}),
]
# 模拟24小时
for hour in range(24):
for npc in npcs:
npc.tick(events if hour in [6, 14] else [])
if hour in [8, 12, 18]:
print(f"\n--- {hour}:00 ---")
for npc in npcs:
print(npc.status())
# 模拟对话上下文生成
print("\n=== 对话上下文示例 ===")
ctx = npcs[0].generate_dialogue_context("北方的怪物是真的吗？")
import json
print(json.dumps(ctx, ensure_ascii=False, indent=2))
11. 性能优化与实时约束
11.1 AI性能优化策略
import time
from functools import lru_cache
from collections import deque
import heapq
class AIPerformanceOptimizer:
"""AI性能优化工具集"""
@staticmethod
def time_sliced_update(entities: list, budget_ms: float,
update_fn, dt: float) -> dict:
"""时间切片：在帧预算内尽可能多地更新实体"""
start = time.perf_counter()
updated = 0
skipped = 0
for entity in entities:
elapsed = (time.perf_counter() - start) * 1000
if elapsed >= budget_ms:
skipped += 1
continue
update_fn(entity, dt)
updated += 1
return {
"updated": updated,
"skipped": skipped,
"total": len(entities),
"elapsed_ms": (time.perf_counter() - start) * 1000
}
@staticmethod
def lod_update(entities: list, player_pos: tuple, dt: float,
distances: dict = None) -> dict:
"""LOD（细节层次）更新：远处实体降低更新频率"""
updates = {"high": 0, "medium": 0, "low": 0, "skip": 0}
for entity in entities:
if distances and entity in distances:
dist = distances[entity]
else:
dist = 0
if dist < 20:
# 近处：每帧更新
updates["high"] += 1
elif dist < 50:
# 中等：每3帧更新
updates["medium"] += 1
elif dist < 100:
# 远处：每10帧更新
updates["low"] += 1
else:
# 极远：不更新
updates["skip"] += 1
return updates
@staticmethod
def spatial_hash(entities: list, cell_size: float = 10.0) -> dict:
"""空间哈希：快速邻近查询"""
grid = {}
for entity in entities:
pos = entity.get("pos", (0, 0))
cell = (int(pos[0] // cell_size), int(pos[1] // cell_size))
if cell not in grid:
grid[cell] = []
grid[cell].append(entity)
return grid
@staticmethod
def find_nearby(grid: dict, pos: tuple, cell_size: float = 10.0,
radius: float = 20.0) -> list:
"""在空间哈希中查找邻近实体"""
cx = int(pos[0] // cell_size)
cy = int(pos[1] // cell_size)
cells_to_check = radius / cell_size + 1
nearby = []
for dx in range(-int(cells_to_check), int(cells_to_check) + 1):
for dy in range(-int(cells_to_check), int(cells_to_check) + 1):
cell = (cx + dx, cy + dy)
if cell in grid:
for entity in grid[cell]:
ex, ey = entity.get("pos", (0, 0))
dist = ((pos[0] - ex)**2 + (pos[1] - ey)**2) ** 0.5
if dist <= radius:
nearby.append((entity, dist))
return sorted(nearby, key=lambda x: x[1])
class BehaviorTreeCache:
"""行为树结果缓存"""
def __init__(self, cache_ttl: int = 5):
self.cache = {}
self.cache_ttl = cache_ttl
self.tick_count = 0
def get_or_compute(self, key: str, compute_fn, *args):
self.tick_count += 1
if key in self.cache:
result, cached_tick = self.cache[key]
if self.tick_count - cached_tick < self.cache_ttl:
return result
result = compute_fn(*args)
self.cache[key] = (result, self.tick_count)
return result
def invalidate(self, key: str = None):
if key:
self.cache.pop(key, None)
else:
self.cache.clear()
# 性能基准测试
print("=== AI性能基准测试 ===\n")
# 创建大量实体
entities = [{"id": i, "pos": (i * 3 % 100, i * 7 % 100), "type": "npc"}
for i in range(1000)]
# 测试时间切片
def dummy_update(entity, dt):
pass # 模拟AI更新
result = AIPerformanceOptimizer.time_sliced_update(
entities, budget_ms=2.0, update_fn=dummy_update, dt=0.016)
print(f"时间切片 (2ms预算):")
print(f" 更新: {result['updated']}/{result['total']}, 跳过: {result['skipped']}")
print(f" 耗时: {result['elapsed_ms']:.2f}ms")
# 测试空间哈希
grid = AIPerformanceOptimizer.spatial_hash(entities, cell_size=10.0)
nearby = AIPerformanceOptimizer.find_nearby(grid, (50, 50), radius=15)
print(f"\n空间哈希邻近查询:")
print(f" 在(50,50)附近15单位内找到 {len(nearby)} 个实体")
# 测试LOD分布
lod = AIPerformanceOptimizer.lod_update(entities, (50, 50), 0.016)
print(f"\nLOD更新分布:")
for level, count in lod.items():
bar = "█" * (count // 10)
print(f" {level}: {count} {bar}")
# 测试行为树缓存
cache = BehaviorTreeCache(cache_ttl=3)
computation_count = 0
def expensive_computation(x):
global computation_count
computation_count += 1
return x * 2
for tick in range(20):
result = cache.get_or_compute("test_key", expensive_computation, 42)
print(f"\n行为树缓存 (20次tick, TTL=3):")
print(f" 实际计算次数: {computation_count} (预期7次)")
print(f" 缓存命中率: {1 - computation_count/20:.0%}")
11.2 总结
游戏AI开发是一个充满创造力的领域。从基础的有限状态机和行为树，到效用系统、LLM驱动的对话，再到强化学习训练的智能体——每种技术都有其适用场景。关键在于：
- 选择合适的复杂度：简单NPC用FSM即可，复杂角色才需要行为树或效用系统
- 性能与智能的平衡：用LOD、时间切片、缓存等技术确保AI不会拖垮帧率
- 数据驱动设计：通过玩家行为分析持续优化AI表现
- LLM是增强而非替代：用LLM增强对话和创意，但核心行为逻辑仍需传统AI架构保障确定性
随着LLM和强化学习技术的持续进化，游戏中的AI角色将变得越来越智能、越来越有"灵魂"。掌握这些技术，就能为玩家创造真正令人难忘的游戏体验。
