15.2 NPC 智能体系统 · CatPtainSchool

15.2.1 基于 HelloAgents 的 SimpleAgent

在赛博小镇中，每个 NPC 都是一个独立的智能体。我们使用 HelloAgents 框架中的 SimpleAgent 来实现 NPC 的智能。SimpleAgent 是一个轻量级的智能体实现，它封装了 LLM 调用、消息管理和工具调用等核心功能。

回顾一下第七章中我们学习的 SimpleAgent，它的核心是一个简单的对话循环：接收用户消息，调用 LLM 生成回复，返回结果。在赛博小镇中，我们需要为每个 NPC 创建一个 SimpleAgent 实例，并为其配置独特的系统提示词，让每个 NPC 拥有不同的性格和角色设定。

让我们看看如何创建一个 NPC Agent。首先，我们需要定义 NPC 的基本信息，包括 ID、名称、职业和性格。然后，我们根据这些信息构建系统提示词，让 LLM 扮演这个 NPC 的角色。最后，我们创建 SimpleAgent 实例，并配置记忆系统。

from hello_agents import SimpleAgent, HelloAgentsLLM
from hello_agents.memory import MemoryManager, WorkingMemory, EpisodicMemory

def create_npc_agent(npc_id: str, name: str, role: str, personality: str):
    """创建NPC Agent"""
    # 构建系统提示词
    system_prompt = f"""你是{name},一位{role}。
你的性格特点:{personality}

你在Datawhale办公室工作,与同事们一起推动开源社区的发展。
请根据你的角色和性格,自然地与玩家对话。
记住你们之前的对话内容,保持对话的连贯性。
"""

    # 创建LLM实例
    llm = HelloAgentsLLM()

    # 创建记忆管理器
    memory_manager = MemoryManager(
        working_memory=WorkingMemory(capacity=10, ttl_minutes=120),
        episodic_memory=EpisodicMemory(
            db_path=f"memory_data/{npc_id}_episodic.db",
            collection_name=f"{npc_id}_memories"
        )
    )

    # 创建Agent
    agent = SimpleAgent(
        name=name,
        llm=llm,
        system_prompt=system_prompt,
        memory_manager=memory_manager
    )

    return agent

这段代码展示了如何创建一个 NPC Agent。系统提示词定义了 NPC 的身份和性格，记忆管理器让 NPC 能够记住与玩家的对话历史。WorkingMemory 是短期记忆，容量为 10 条消息，保留时间为 120 分钟。EpisodicMemory 是长期记忆，使用 SQLite 数据库和 Qdrant 向量数据库存储，可以检索相关的历史对话。

NPC Agent 的工作流程如图 15.5 所示：

图 15.5 NPC Agent 工作流程

15.2.2 NPC 角色设定与 Prompt 设计

一个好的 NPC 需要有鲜明的性格和角色设定。在赛博小镇中，我们设计了三个 NPC，分别代表不同的职业和性格。

张三 - Python 工程师

张三是一位资深的 Python 工程师，负责 HelloAgents 框架的核心开发。他性格严谨，说话直接，喜欢用技术术语。他对代码质量有很高的要求，经常会分享一些编程技巧和最佳实践。

npc_zhang = {
    "npc_id": "zhang_san",
    "name": "张三",
    "role": "Python工程师",
    "personality": "严谨、专业、喜欢分享技术知识。说话直接,注重代码质量。"
}

李四 - 产品经理

李四是一位经验丰富的产品经理，负责 HelloAgents 框架的产品规划和用户体验设计。他性格外向，善于沟通，总是能从用户的角度思考问题。他喜欢讨论产品设计和用户需求，经常会问"为什么"。

npc_li = {
    "npc_id": "li_si",
    "name": "李四",
    "role": "产品经理",
    "personality": "外向、善于沟通、注重用户体验。喜欢从用户角度思考问题。"
}

王五 - UI 设计师

王五是一位富有创意的 UI 设计师，负责 HelloAgents 框架的界面设计和视觉呈现。他性格温和，审美独特，对色彩和布局有敏锐的感知。他喜欢讨论设计理念和美学，经常会分享一些设计灵感。

npc_wang = {
    "npc_id": "wang_wu",
    "name": "王五",
    "role": "UI设计师",
    "personality": "温和、富有创意、审美独特。注重视觉呈现和用户体验。"
}

这三个 NPC 的设定各有特色，玩家可以根据自己的兴趣选择与不同的 NPC 互动。张三可以教你编程技巧，李四可以和你讨论产品设计，王五可以分享设计灵感。

15.2.3 记忆系统集成

记忆系统是 NPC 智能的关键。一个能够记住过去对话的 NPC，会让玩家感觉更加真实和有趣。我们采用 helloagents 的WorkingMemory和EpisodicMemory构造短期记忆和长期记忆。

短期记忆存储最近的对话内容，容量有限，会随着时间自动清理。它的作用是保持对话的连贯性，让 NPC 能够理解上下文。比如，当玩家说"它是什么颜色的?"时，NPC 需要从短期记忆中找到"它"指的是什么。

长期记忆存储所有的对话历史，使用向量数据库进行语义检索。当玩家提到某个话题时，NPC 可以从长期记忆中检索相关的历史对话，回忆起之前讨论过的内容。比如，当玩家说"还记得我们上次讨论的那个项目吗?"，NPC 可以从长期记忆中找到相关的对话记录。

记忆系统的架构如图 15.6 所示：

图 15.6 记忆系统架构

在实际使用中，Agent 会先从短期记忆中获取最近的对话，然后从长期记忆中检索相关的历史对话，将这些信息一起发送给 LLM，生成更加准确和个性化的回复。

# Agent处理对话的流程
def process_dialogue(agent, player_message):
    # 1. 从短期记忆获取最近对话
    recent_messages = agent.memory_manager.working_memory.get_recent_messages(5)

    # 2. 从长期记忆检索相关历史
    relevant_memories = agent.memory_manager.episodic_memory.search(
        query=player_message,
        top_k=3
    )

    # 3. 构建上下文
    context = {
        "recent": recent_messages,
        "relevant": relevant_memories
    }

    # 4. 调用Agent生成回复
    reply = agent.run(player_message, context=context)

    # 5. 保存到记忆系统
    agent.memory_manager.add_interaction(player_message, reply)

    return reply

这个流程确保了 NPC 能够记住与玩家的互动历史，并在对话中体现出来。

15.2.4 批量对话生成：轻负载模式

在实际运行中，很快就会发现了一个问题：当多个玩家同时与不同的 NPC 对话时，后端需要并发处理多个 LLM 请求。每个请求都需要调用 API，这不仅增加了成本，还可能因为并发限制导致请求失败或延迟。

为了解决这个问题，我们设计了一个批量对话生成系统。核心思想是：将多个 NPC 的对话请求合并成一次 LLM 调用，让 LLM 一次性生成所有 NPC 的回复。这就像餐厅的"预制菜"一样，提前批量准备好，需要时直接使用，大大降低了成本和延迟。

批量生成的工作流程如图 15.7 所示：

图 15.7 批量生成 vs 传统模式

批量生成器的实现非常巧妙。我们构建一个特殊的提示词，要求 LLM 一次性生成所有 NPC 的对话，并以 JSON 格式返回。这样，一次 API 调用就能获得所有 NPC 的回复，成本降低到原来的 1/3，延迟也大幅减少。

class NPCBatchGenerator：
    """批量生成NPC对话的生成器"""

    def __init__(self):
        self.llm = HelloAgentsLLM()
        self.npc_configs = NPC_ROLES  # 所有NPC的配置

    def generate_batch_dialogues(self, context: Optional[str] = None) -> Dict[str, str]:
        """批量生成所有NPC的对话

        Args:
            context: 场景上下文(如"上午工作时间"、"午餐时间"等)

        Returns:
            Dict[str, str]: NPC名称到对话内容的映射
        """
        # 构建批量生成提示词
        prompt = self._build_batch_prompt(context)

        # 一次LLM调用生成所有对话
        response = self.llm.invoke([
            {"role": "system", "content": "你是一个游戏NPC对话生成器,擅长创作自然真实的办公室对话。"},
            {"role": "user", "content": prompt}
        ])

        # 解析JSON响应
        dialogues = json.loads(response)
        # 返回格式: {"张三": "...", "李四": "...", "王五": "..."}

        return dialogues

    def _build_batch_prompt(self, context: Optional[str] = None) -> str:
        """构建批量生成提示词"""
        # 根据时间自动推断场景
        if context is None:
            context = self._get_current_context()

        # 构建NPC描述
        npc_descriptions = []
        for name, cfg in self.npc_configs.items():
            desc = f"- {name}({cfg['title']}): 在{cfg['location']}{cfg['activity']},性格{cfg['personality']}"
            npc_descriptions.append(desc)

        npc_desc_text = "\n".join(npc_descriptions)

        prompt = f"""请为Datawhale办公室的3个NPC生成当前的对话或行为描述。

【场景】{context}

【NPC信息】
{npc_desc_text}

【生成要求】
1. 每个NPC生成1句话(20-40字)
2. 内容要符合角色设定、当前活动和场景氛围
3. 可以是自言自语、工作状态描述、或简单的思考
4. 要自然真实,像真实的办公室同事
5. **必须严格按照JSON格式返回**

【输出格式】(严格遵守)
{{"张三": "...", "李四": "...", "王五": "..."}}

【示例输出】
{{"张三": "这个bug真是见鬼了,已经调试两小时了...", "李四": "嗯,这个功能的优先级需要重新评估一下。", "王五": "这杯咖啡的拉花真不错,灵感来了!"}}

请生成(只返回JSON,不要其他内容):
"""
        return prompt

这个设计的关键在于提示词的构建。我们明确要求 LLM 返回 JSON 格式，并提供了示例输出。LLM 会严格按照这个格式生成回复，我们只需要解析 JSON 就能获得所有 NPC 的对话。

批量生成还有一个额外的好处：所有 NPC 的对话是在同一个上下文中生成的，因此它们之间会有一定的关联性。比如，如果张三在调试 bug，李四可能会提到要帮忙看看;如果王五在设计界面，张三可能会说等会儿去看看设计稿。这让整个办公室的氛围更加真实和连贯。

当然，批量生成也有一些限制。它更适合生成 NPC 的"背景对话"或"自言自语"，而不是与玩家的直接互动。对于玩家发起的对话，我们仍然使用单独的 Agent 来处理，以保证回复的个性化和准确性。批量生成主要用于以下场景：

NPC 背景对话：玩家进入场景时，NPC 正在做什么、说什么
定时更新：每隔一段时间更新 NPC 的状态和对话
场景氛围：根据时间(早上、中午、晚上)生成不同的对话
降低成本：在高并发场景下，使用批量生成降低 API 调用次数

混合模式：批量生成+即时响应

在实际实现中，我们采用了一种混合模式，将批量生成和即时响应结合起来。这个设计非常巧妙，既保证了效率，又保证了交互的质量。

具体来说，系统会在后台定期运行批量生成，为所有 NPC 生成当前场景下的"背景对话"。这些对话会被缓存起来，当玩家靠近 NPC 但还没有发起交互时，NPC 会显示这些背景对话，比如"正在调试代码..."、"在看产品文档..."等。这让 NPC 看起来是"活着的"，而不是静止的模型。

但是，当玩家按下 E 键发起交互时，系统会立即切换到即时响应模式。此时，后端会调用该 NPC 的专属 Agent，根据玩家的具体消息、历史记忆和好感度，生成个性化的回复。这个过程是实时的，确保 NPC 的回复与玩家的输入高度相关。

# 在main.py中的混合模式实现
@app.post("/dialogue")
async def dialogue(request: DialogueRequest):
    """处理玩家与NPC的对话(即时响应模式)"""
    npc_id = request.npc_id
    player_message = request.player_message
    player_name = request.player_name

    # 获取NPC Agent(每个NPC有独立的Agent)
    agent = npc_agents.get(npc_id)
    if not agent:
        raise HTTPException(status_code=404, detail="NPC not found")

    # 即时生成个性化回复
    # 这里不使用批量生成,而是调用Agent的run方法
    reply = agent.run(player_message)

    # 更新好感度
    affinity_change = relationship_manager.update_affinity(
        npc_id, player_name, player_message, reply
    )

    return {
        "npc_reply": reply,
        "affinity_score": affinity_change["score"],
        "affinity_level": affinity_change["level"]
    }

# 后台任务:定期批量生成背景对话
async def background_dialogue_update():
    """后台任务:每5分钟更新一次NPC背景对话"""
    while True:
        try:
            # 使用批量生成器生成所有NPC的背景对话
            batch_generator = get_batch_generator()
            dialogues = batch_generator.generate_batch_dialogues()

            # 更新到状态管理器
            for npc_name, dialogue in dialogues.items():
                state_manager.update_npc_background_dialogue(npc_name, dialogue)

            print(f"✅ 背景对话更新完成: {len(dialogues)}个NPC")
        except Exception as e:
            print(f"❌ 背景对话更新失败: {e}")

        # 等待5分钟
        await asyncio.sleep(300)

这种混合模式的优势非常明显：

降低成本：背景对话使用批量生成，一次调用生成所有 NPC 的对话，成本低
保证质量：玩家交互使用即时响应，每个回复都是个性化的，质量高
提升体验：NPC 始终有"背景对话"，看起来很生动;玩家交互时回复准确，体验好
灵活调整：可以根据服务器负载动态调整批量生成的频率

通过批量生成和即时响应的结合，我们实现了一个既高效又智能的 NPC 系统。在正常情况下，玩家感受不到任何差异，但后端的成本和性能得到了显著优化。这个设计思路也可以应用到其他需要大量 AI 调用的场景中。