15.4.1 FastAPI 应用结构
赛博小镇的后端使用 FastAPI 框架构建，负责处理 Godot 前端的请求，调用 HelloAgents 的 NPC Agent，管理 NPC 状态和好感度，以及记录日志。一个清晰的应用结构能够让代码更易于维护和扩展。
我们的 FastAPI 应用采用模块化设计，将不同的功能分离到不同的文件中，如图 15.10 所示:
图 15.10 后端应用结构
让我们从main.py开始，这是 FastAPI 应用的入口文件：
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
from typing import Optional
import uvicorn
from agents import NPCAgentManager
from relationship_manager import RelationshipManager
from state_manager import StateManager
from logger import DialogueLogger
from config import settings
# 创建FastAPI应用
app = FastAPI(
title="赛博小镇后端服务",
description="基于HelloAgents的AI NPC对话系统",
version="1.0.0"
)
# 配置CORS,允许Godot前端访问
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # 生产环境应该限制具体域名
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# 初始化各个管理器
agent_manager = NPCAgentManager()
relationship_manager = RelationshipManager()
state_manager = StateManager()
dialogue_logger = DialogueLogger()
@app.on_event("startup")
async def startup_event():
"""应用启动时的初始化"""
print("=" * 60)
print("🎮 赛博小镇后端服务启动中...")
print("=" * 60)
# 初始化NPC Agents
agent_manager.initialize_npcs()
print("✅ NPC Agents已初始化")
# 初始化状态管理器
state_manager.initialize_npcs()
print("✅ 状态管理器已初始化")
@app.get("/")
async def root():
"""健康检查"""
return {
"status": "running",
"message": "赛博小镇后端服务正在运行",
"version": "1.0.0",
"npcs": state_manager.get_npc_count()
}
if __name__ == "__main__":
uvicorn.run(
app,
host=settings.HOST,
port=settings.PORT,
log_level="info"
)
这个主程序文件定义了 FastAPI 应用的基本结构，配置了 CORS 中间件以允许跨域请求，并在启动时初始化各个管理器。接下来我们将实现具体的 API 路由。
15.4.2 API 路由设计
赛博小镇的后端需要提供几个核心 API 端点，用于处理 Godot 前端的请求。我们将这些路由添加到main.py中。
获取 NPC 状态
这个 API 返回所有 NPC 的当前状态，包括位置、是否忙碌等信息：
from models import NPCStatusResponse
@app.get("/npcs/status", response_model=NPCStatusResponse)
async def get_npc_status():
"""获取所有NPC的状态"""
npcs = state_manager.get_all_npc_states()
return {"npcs": npcs}
@app.get("/npcs/{npc_id}/status")
async def get_single_npc_status(npc_id: str):
"""获取单个NPC的状态"""
npc = state_manager.get_npc_state(npc_id)
if not npc:
raise HTTPException(status_code=404, detail=f"NPC {npc_id} 不存在")
return npc
对话接口
这是最核心的 API，处理玩家与 NPC 的对话：
from models import DialogueRequest, DialogueResponse
@app.post("/dialogue", response_model=DialogueResponse)
async def dialogue(request: DialogueRequest):
"""处理玩家与NPC的对话"""
# 1. 验证NPC是否存在
if not agent_manager.has_npc(request.npc_id):
raise HTTPException(status_code=404, detail=f"NPC {request.npc_id} 不存在")
# 2. 检查NPC是否忙碌
if state_manager.is_npc_busy(request.npc_id):
raise HTTPException(status_code=409, detail=f"NPC {request.npc_id} 正在与其他玩家对话")
# 3. 标记NPC为忙碌状态
state_manager.set_npc_busy(request.npc_id, True)
try:
# 4. 获取当前好感度
affinity_info = relationship_manager.get_affinity(
request.npc_id,
request.player_name
)
# 5. 调用Agent生成回复
agent = agent_manager.get_agent(request.npc_id, affinity_info["level"])
reply = agent.run(request.player_message)
# 6. 更新好感度
new_affinity = relationship_manager.update_affinity(
request.npc_id,
request.player_name,
request.player_message,
reply
)
# 7. 记录日志
dialogue_logger.log_dialogue(
npc_id=request.npc_id,
player_name=request.player_name,
player_message=request.player_message,
npc_reply=reply,
affinity_info=new_affinity
)
# 8. 返回回复
return DialogueResponse(
npc_reply=reply,
affinity_level=new_affinity["level"],
affinity_score=new_affinity["score"]
)
except Exception as e:
dialogue_logger.log_error(f"对话处理失败: {str(e)}")
raise HTTPException(status_code=500, detail=f"对话处理失败: {str(e)}")
finally:
# 9. 释放NPC状态
state_manager.set_npc_busy(request.npc_id, False)
好感度查询
这个 API 允许查询玩家与 NPC 的好感度：
from models import AffinityInfo
@app.get("/affinity/{npc_id}/{player_name}", response_model=AffinityInfo)
async def get_affinity(npc_id: str, player_name: str):
"""获取玩家与NPC的好感度"""
if not agent_manager.has_npc(npc_id):
raise HTTPException(status_code=404, detail=f"NPC {npc_id} 不存在")
affinity = relationship_manager.get_affinity(npc_id, player_name)
return affinity
API 路由的调用流程如图 15.11 所示：
图 15.11 API 调用流程
15.4.3 状态管理与日志系统
状态管理器
状态管理器负责跟踪每个 NPC 的当前状态，包括位置、是否忙碌、当前动作等。这对于防止并发问题很重要,比如避免一个 NPC 同时与多个玩家对话。
# state_manager.py
from typing import Dict, List, Optional
from datetime import datetime
class StateManager:
"""NPC状态管理器"""
def __init__(self):
self.npc_states: Dict[str, dict] = {}
def initialize_npcs(self):
"""初始化NPC状态"""
npcs = [
{
"npc_id": "zhang_san",
"name": "张三",
"role": "Python工程师",
"position": {"x": 300, "y": 200}
},
{
"npc_id": "li_si",
"name": "李四",
"role": "产品经理",
"position": {"x": 500, "y": 200}
},
{
"npc_id": "wang_wu",
"name": "王五",
"role": "UI设计师",
"position": {"x": 700, "y": 200}
}
]
for npc in npcs:
self.npc_states[npc["npc_id"]] = {
**npc,
"is_busy": False,
"current_action": "idle",
"last_interaction": None
}
def get_npc_state(self, npc_id: str) -> Optional[dict]:
"""获取NPC状态"""
return self.npc_states.get(npc_id)
def get_all_npc_states(self) -> List[dict]:
"""获取所有NPC状态"""
return list(self.npc_states.values())
def is_npc_busy(self, npc_id: str) -> bool:
"""检查NPC是否忙碌"""
npc = self.npc_states.get(npc_id)
return npc["is_busy"] if npc else False
def set_npc_busy(self, npc_id: str, busy: bool):
"""设置NPC忙碌状态"""
if npc_id in self.npc_states:
self.npc_states[npc_id]["is_busy"] = busy
if busy:
self.npc_states[npc_id]["last_interaction"] = datetime.now().isoformat()
def get_npc_count(self) -> int:
"""获取NPC数量"""
return len(self.npc_states)
日志系统
日志系统实现了双输出：控制台和文件。这样既方便实时查看，又能保存历史记录。
# logger.py
import logging
from datetime import datetime
from pathlib import Path
class DialogueLogger:
"""对话日志记录器"""
def __init__(self, log_dir: str = "logs"):
self.log_dir = Path(log_dir)
self.log_dir.mkdir(exist_ok=True)
# 创建日志文件名(按日期)
today = datetime.now().strftime("%Y-%m-%d")
log_file = self.log_dir / f"dialogue_{today}.log"
# 配置日志
self.logger = logging.getLogger("DialogueLogger")
self.logger.setLevel(logging.INFO)
# 控制台处理器
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)
console_formatter = logging.Formatter(
'%(asctime)s - %(levelname)s - %(message)s',
datefmt='%H:%M:%S'
)
console_handler.setFormatter(console_formatter)
# 文件处理器
file_handler = logging.FileHandler(log_file, encoding='utf-8')
file_handler.setLevel(logging.INFO)
file_formatter = logging.Formatter(
'%(asctime)s - %(levelname)s - %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
file_handler.setFormatter(file_formatter)
# 添加处理器
self.logger.addHandler(console_handler)
self.logger.addHandler(file_handler)
def log_dialogue(self, npc_id: str, player_name: str,
player_message: str, npc_reply: str,
affinity_info: dict):
"""记录对话"""
log_message = f"""
{'='*60}
NPC: {npc_id}
玩家: {player_name}
玩家消息: {player_message}
NPC回复: {npc_reply}
好感度: {affinity_info['level']} ({affinity_info['score']}/100)
互动次数: {affinity_info['interaction_count']}
{'='*60}
"""
self.logger.info(log_message)
def log_error(self, error_message: str):
"""记录错误"""
self.logger.error(error_message)
这个日志系统会在控制台实时显示对话内容，同时保存到文件中。每天的日志会保存在单独的文件中,方便后续分析。
15.4.4 理解 Godot 的场景系统
在开始构建游戏场景之前，我们需要先理解 Godot 的核心概念——场景(Scene)和节点(Node)。这是 Godot 与其他游戏引擎最大的不同之处，也是它最强大的特性之一。
什么是节点?
节点是 Godot 中最基本的构建块。你可以把节点想象成乐高积木，每个节点都有特定的功能。比如，Sprite2D 节点用于显示图片，AudioStreamPlayer 节点用于播放音频，CharacterBody2D 节点用于处理角色的物理移动。Godot 提供了上百种不同类型的节点，每种节点都专注于做好一件事。
节点之间可以形成父子关系，构成一个树状结构。父节点可以影响子节点，比如移动父节点会同时移动所有子节点，隐藏父节点会同时隐藏所有子节点。这种层级关系让我们可以轻松地组织和管理复杂的游戏对象。
什么是场景?
场景是一组节点的集合，保存在一个.tscn 文件中。你可以把场景理解为一个"预制件"。比如，我们可以创建一个"玩家"场景，包含角色的精灵、碰撞体、音效等所有相关节点。然后在游戏中多次使用这个场景，每次使用都会创建一个独立的实例。
场景的强大之处在于它的可复用性和模块化。我们可以在一个场景中实例化另一个场景，形成嵌套结构。比如，主场景可以包含玩家场景、多个 NPC 场景和 UI 场景。修改 NPC 场景会自动影响所有 NPC 实例，这大大简化了开发和维护。
一个简单的例子
让我们用一个简单的例子来理解场景和节点。假设我们要创建一个"玩家"场景：
Player (CharacterBody2D) ← 根节点,负责物理移动
├─ AnimatedSprite2D ← 子节点,显示角色动画
├─ CollisionShape2D ← 子节点,定义碰撞形状
└─ Camera2D ← 子节点,摄像机跟随玩家
这个场景包含 4 个节点，形成树状结构。CharacterBody2D 是根节点，其他三个是它的子节点。我们可以给每个节点添加脚本来控制它的行为，也可以给根节点添加脚本来协调所有子节点。
当我们在主场景中实例化这个 Player 场景时，Godot 会创建这整个节点树的一个副本。我们可以创建多个玩家实例，每个实例都是独立的，有自己的位置、状态和行为。
场景实例化的优势
在赛博小镇中，我们有三个 NPC：张三、李四和王五。如果不使用场景系统，我们需要为每个 NPC 分别创建节点、设置属性、编写脚本，这会导致大量重复工作。而使用场景系统，我们只需要创建一个通用的 NPC 场景，然后实例化三次，通过脚本参数设置不同的名称和角色信息即可。
这种设计的好处是：如果我们想给所有 NPC 添加一个新功能(比如头顶显示对话气泡)，只需要修改 NPC 场景，所有实例都会自动获得这个功能。