各位同仁，大家下午好！
今天，我们齐聚一堂，共同探讨一个在大规模对话系统中至关重要，却又极具挑战性的议题——Agent 人设一致性。随着大型语言模型（LLM）能力的飞速发展，我们构建的对话 Agent 越来越智能、越来越拟人化。然而，当这些 Agent 投入到真实世界，面对海量用户、漫长对话以及复杂多变的场景时，一个核心痛点便浮出水面：如何防止 Agent 的“人设崩塌”？
所谓“人设崩塌”，是指 Agent 在不同对话轮次、不同对话情境下，其身份、角色、语气、知识倾向甚至价值观出现矛盾或漂移，导致用户体验割裂，信任感降低。这不仅影响用户对Agent的感知，更可能损害品牌形象。为了解决这一问题，我今天将向大家介绍一个强大而优雅的解决方案：利用记忆锚点（Memory Anchors）来守护 Agent 的人设阵地。
一、Agent 人设：定义、重要性与挑战
在深入探讨记忆锚点之前，我们首先要明确什么是 Agent 人设，它为何如此重要，以及当前LLM面临的挑战。
1.1 Agent 人设的构成要素
一个Agent的人设，并非仅仅是它叫什么名字、扮演什么角色，它是一个多维度、动态的综合体，通常包含以下几个核心要素：
- 身份与角色 (Identity & Role): Agent的名称、性别（如果有），以及它在对话中扮演的具体角色（例如：客服、技术顾问、创意伙伴、虚拟助手等）。
- 语气与风格 (Tone & Style): Agent的表达方式，是正式严谨、幽默活泼、亲切友善，还是冷静客观？这包括词汇选择、句式结构、情感倾向等。
- 核心知识与专业领域 (Core Knowledge & Expertise): Agent所具备的、且在任何情况下都应坚持的专业知识边界和核心能力。
- 价值观与行为准则 (Values & Principles): Agent处理问题、提供建议时所依据的基本原则，例如：始终以用户利益为先、遵守法律法规、保持中立等。
- 长期记忆与偏好 (Long-term Memory & Preferences): 在长期互动中形成的用户特定偏好、历史交互记录中体现的Agent自身特点。
这些要素共同构建了一个立体、可信赖的Agent形象。
1.2 人设一致性的重要性
人设一致性是构建优秀对话体验的基石：
- 提升用户信任度： 一致的人设让用户感到Agent是可预测、可靠的，从而更容易建立信任。
- 优化用户体验： 避免了用户因Agent前后不一而产生的困惑、沮丧，提升了对话的流畅性和舒适度。
- 强化品牌形象： 对于企业而言，Agent是品牌在用户面前的直接代表。一致的人设有助于塑造和维护积极的品牌形象。
- 提高任务完成效率： 用户清楚Agent的能力和边界，能更高效地进行沟通，减少误解和重复提问。
1.3 LLM在人设一致性上面临的挑战
尽管LLM拥有强大的生成能力，但在维持人设一致性方面仍面临固有挑战：
- 上下文窗口限制 (Context Window Limitation): LLM的输入有最大令牌数限制。在长时间对话中，早期的对话历史和人设信息可能被截断，导致Agent“失忆”。
- 无状态性 (Statelessness): 每次API调用对于LLM而言都是一个全新的请求，它本身不保留任何长期记忆。所有状态都需要外部系统进行管理和维护。
- 上下文漂移 (Context Drift): 即使在上下文窗口内，随着对话的进行，LLM可能会过度关注最近的输入，而逐渐淡忘早期设定的人设指令，导致人设逐渐偏离。
- 信息冲突 (Information Contradiction): 在复杂场景下，用户输入、检索到的知识和Agent预设人设之间可能存在冲突，LLM需要一套机制来解决这些冲突并保持人设主线。
- 缺乏显式记忆管理机制 (Lack of Explicit Memory Management): LLM本身没有内置的长期记忆或优先级机制来区分哪些信息是人设核心、哪些是临时上下文。
为了克服这些挑战，我们需要一套智能的外部记忆管理系统，而“记忆锚点”正是这套系统的核心。
二、记忆锚点：核心概念与分类
记忆锚点，顾名思义，是对话Agent用于“锚定”其核心人设特征的持久化、高优先级的信息片段。它们如同船只的锚，在波涛汹涌的对话海洋中，确保Agent始终停留在预设的港湾。
2.1 记忆锚点的定义与工作原理
定义： 记忆锚点是Agent人设中那些最为关键、不应被遗忘、且应当在绝大多数对话中被优先考虑的结构化或非结构化信息。它们可以是Agent的名称、角色宣言、核心价值观、特定知识点，甚至是与用户建立的特定长期关系。
工作原理： 在每次与LLM交互之前，我们的系统会智能地选择并检索相关的记忆锚点，将其注入到LLM的上下文（Prompt）中。这样，无论对话进行到哪个阶段，LLM都能“看到”并参考这些核心人设信息，从而指导其生成符合人设的回复。
2.2 记忆锚点的分类
为了更好地管理和利用记忆锚点，我们可以将其划分为不同的类型：
2.2.1 静态锚点 (Static Anchors)
- 定义： 一旦设定，在Agent的生命周期内基本保持不变的锚点。它们是Agent人设的基础和核心。
- 示例：
- Agent的名称、角色。
- Agent的固定语气和风格指南。
- Agent的核心能力范围和禁区。
- Agent的基本行为准则（例如：绝不提供医疗建议、绝不发表政治评论）。
- 存储方式： 通常以配置文件（JSON, YAML）、数据库表的形式存储。
- 检索方式： 每次对话固定加载或按需加载全部。
2.2.2 动态锚点 (Dynamic Anchors)
动态锚点是根据对话进展、用户行为或系统反馈而实时生成、更新或调整的锚点。它们为人设增添了灵活性和适应性。
-
2.2.2.1 事件驱动锚点 (Event-Driven Anchors)
- 定义： 由特定事件触发而产生或更新的锚点。
- 示例：
- 用户明确声明的偏好（如：“我喜欢简洁明了的回答”）。
- Agent在对话中做出的承诺（如：“我已记录您的问题，稍后会为您跟进”）。
- 用户在特定模块完成的操作（如：购买了某件商品，Agent应记住并提供相关售后服务）。
- 存储方式： 通常存储在键值数据库（KV Store）或关系型数据库中，与用户ID或会话ID关联。
- 检索方式： 在对话开始时加载与当前用户/会话相关的锚点，或在特定模块中按需加载。
-
2.2.2.2 摘要锚点 (Summarization Anchors)
- 定义： 对长对话历史进行提炼和概括而形成的锚点，用于捕捉对话的核心要点、用户意图或Agent在对话中展现的特定行为。
- 示例：
- 对过去多轮对话的摘要，提炼出用户遇到的核心问题。
- Agent在处理复杂任务时，记录的阶段性进展和关键决策。
- 用户在本次会话中多次强调的某项需求。
- 存储方式： 可以是结构化的文本，也可以是向量嵌入（Vector Embeddings），存储在向量数据库（Vector DB）中，以便进行语义检索。
- 检索方式： 通过语义相似性搜索，从向量数据库中检索与当前用户查询或对话主题最相关的摘要锚点。
-
2.2.2.3 适应性锚点 (Adaptive Anchors)
- 定义： 基于用户反馈、Agent性能评估或A/B测试结果，对Agent人设进行微调和优化的锚点。
- 示例：
- 如果多数用户反馈Agent过于“生硬”，系统可以自动调整语气锚点，使其更“亲切”。
- 如果Agent在特定类型问题上表现不佳，可以添加或修改相关知识边界锚点。
- 存储方式： 通常与静态锚点类似，但需要有更新机制。
- 检索方式： 与静态锚点类似，但其内容是动态调整的。
通过对记忆锚点进行分类，我们可以更系统、更精细地管理Agent的人设，确保其在各种复杂场景下都能保持高度一致性。
三、基于记忆锚点的Agent架构设计
为了有效实现记忆锚点，我们需要一个精心设计的系统架构。下图展示了一个典型的包含记忆锚点的Agent系统结构。
+---------------------+ +---------------------+ +---------------------+
| | | | | |
| 用户输入 (User Input) |----->| 意图识别/实体抽取 |----->| 对话状态管理 |
| | | (Intent/Entity Ext.)| | (Dialogue State Mgt.)|
+---------------------+ +---------------------+ +---------------------+
|
v
+---------------------------------------+
| |
| 记忆锚点管理模块 |
| (Memory Anchor Management) |
| |
+------------------^--------------------+
|
+-----------------------------+-----------------------------+
| | |
v v v
+---------------------+ +---------------------+ +---------------------+
| | | | | |
| 静态锚点存储器 | | 动态锚点存储器 | | 向量记忆存储器 |
| (Static Anchor Store)| | (Dynamic Anchor Store)| | (Vector Memory Store)|
| (JSON/DB Table) | | (KV/Relational DB) | | (Vector Database) |
+---------------------+ +---------------------+ +---------------------+
^ ^ ^
| | |
+-----------------------------+-----------------------------+
|
v
+---------------------------------------+
| |
| 上下文构建模块 |
| (Context Builder) |
| (整合用户输入、历史、锚点、外部知识) |
+------------------^--------------------+
|
v
+---------------------------------------+
| |
| LLM交互模块 |
| (LLM Interaction) |
| |
+------------------^--------------------+
|
v
+---------------------------------------+
| |
| LLM 输出解析 |
| (LLM Output Parser) |
| |
+------------------^--------------------+
|
v
+---------------------------------------+
| |
| 人设监控与反馈 |
| (Persona Monitoring & Feedback) |
+---------------------------------------+
|
v
+---------------------+
| |
| Agent 回复 |
| (Agent Response) |
+---------------------+3.1 核心组件详解
- 意图识别/实体抽取 (Intent/Entity Extraction): 对用户输入进行初步理解，提取关键信息。
- 对话状态管理 (Dialogue State Management): 维护当前会话的上下文状态，包括当前任务、已收集槽位等。
- 记忆锚点管理模块 (Memory Anchor Management Module): 这是系统的核心。它负责：
- 根据当前对话状态和用户输入，决定需要加载哪些记忆锚点。
- 与各类记忆存储器交互，检索锚点。
- 在必要时，触发动态锚点的更新（例如，根据用户新的偏好更新锚点）。
- 记忆锚点存储器 (Memory Anchor Stores):
- 静态锚点存储器 (Static Anchor Store): 存储Agent的固有人设信息。可以是JSON文件、YAML文件，或简单的关系型数据库表。
- 动态锚点存储器 (Dynamic Anchor Store): 存储事件驱动的、用户特有的或会话特有的锚点。通常是键值数据库（如Redis）或 NoSQL 数据库，以支持快速读写。
- 向量记忆存储器 (Vector Memory Store): 存储摘要锚点或更复杂的语义记忆。使用向量数据库（如Faiss, Chroma, Pinecone, Milvus），支持通过语义相似性进行高效检索。
- 上下文构建模块 (Context Builder): 这是一个至关重要的环节。它负责将：
- 当前用户输入。
- 精简的对话历史。
- 从各类存储器中检索到的相关记忆锚点。
- （可选）从外部知识库检索到的信息。
- （可选）预设的系统提示词（System Prompt）。
整合、格式化成LLM能够理解并有效利用的完整上下文。
- LLM交互模块 (LLM Interaction Module): 负责将构建好的上下文发送给LLM API（如OpenAI GPT系列、Anthropic Claude等），并接收LLM的原始输出。
- LLM输出解析 (LLM Output Parser): 对LLM的原始文本输出进行结构化解析、后处理，确保其符合预期的格式和要求。
- 人设监控与反馈 (Persona Monitoring & Feedback): 持续监控Agent的回复，评估其人设一致性。可以通过人工标注、用户反馈、自动化指标（如情感分析、关键词匹配）等方式进行。发现人设偏离时，触发对记忆锚点的调整和优化。
- Agent回复 (Agent Response): 最终呈现给用户的Agent响应。
四、记忆锚点实施细节与代码示例
现在，让我们深入到代码层面，看看如何具体实现这些记忆锚点。我们将使用Python作为主要编程语言，结合一些常用的库。
4.1 定义 Agent Persona 的 Schema
首先，我们需要一个清晰、结构化的方式来定义Agent的人设。pydantic是一个非常适合这项任务的库。
from pydantic import BaseModel, Field
from typing import List, Dict, Optional
# 1. 定义 Agent 的核心人设模型
class AgentPersona(BaseModel):
"""
Agent 的核心人设定义，包含静态和动态的关键属性。
"""
id: str = Field(..., description="Agent 的唯一标识符")
name: str = Field(..., description="Agent 的名称")
role: str = Field(..., description="Agent 在对话中扮演的角色")
tone: str = Field(..., description="Agent 的语气和风格（如：'友好', '专业', '幽默'）")
core_principles: List[str] = Field(
default_factory=list,
description="Agent 必须遵守的核心原则和行为准则"
)
knowledge_domain: List[str] = Field(
default_factory=list,
description="Agent 擅长并应严格遵守的知识领域"
)
forbidden_topics: List[str] = Field(
default_factory=list,
description="Agent 绝不能讨论的话题"
)
# 静态锚点可以直接作为Persona的一部分
static_anchors_text: Dict[str, str] = Field(
default_factory=dict,
description="额外静态文本锚点，如介绍语、固定免责声明等"
)
def to_prompt_section(self) -> str:
"""
将核心人设转换为适合注入LLM Prompt的文本片段。
"""
prompt_parts = [
f"你是一位名为 {self.name} 的 {self.role}。",
f"你的语气应该保持 {self.tone}。"
]
if self.core_principles:
prompt_parts.append("你必须遵守以下核心原则：")
prompt_parts.extend([f"- {p}" for p in self.core_principles])
if self.knowledge_domain:
prompt_parts.append(f"你的专业领域是：{', '.join(self.knowledge_domain)}。")
if self.forbidden_topics:
prompt_parts.append(f"你绝不能讨论的话题包括：{', '.join(self.forbidden_topics)}。")
for key, value in self.static_anchors_text.items():
prompt_parts.append(f"重要信息({key}): {value}")
return "n".join(prompt_parts)
# 示例 Agent Persona 实例
CUSTOMER_SERVICE_AGENT_PERSONA = AgentPersona(
id="cs_agent_v1",
name="小A",
role="客户服务专家",
tone="亲切、专业、耐心",
core_principles=[
"始终以用户为中心，提供高效解决方案。",
"尊重用户隐私，不泄露任何个人信息。",
"保持中立，不发表个人政治或宗教观点。",
"在无法回答时，诚实告知并引导用户寻求其他帮助。"
],
knowledge_domain=["产品咨询", "订单管理", "售后服务", "技术支持（L1）"],
forbidden_topics=["医疗建议", "金融投资", "政治评论", "敏感个人信息查询"],
static_anchors_text={
"greeting_phrase": "您好！我是小A，很高兴为您服务。请问有什么可以帮助您的？",
"disclaimer_tech_support": "请注意，我只能提供L1技术支持，更复杂的问题可能需要转接人工客服。",
}
)
print("--- Agent 核心人设定义 ---")
print(CUSTOMER_SERVICE_AGENT_PERSONA.model_dump_json(indent=2))
print("n--- 转换为 Prompt 片段 ---")
print(CUSTOMER_SERVICE_AGENT_PERSONA.to_prompt_section())4.2 静态锚点存储与检索
静态锚点最简单的方式就是直接存储在配置文件或内存中，每次启动Agent时加载。
import json
class StaticAnchorStore:
"""
静态锚点存储器，通过JSON文件加载Agent的固有人设信息。
"""
def __init__(self, persona_file_path: str):
self.persona_file_path = persona_file_path
self._persona: Optional[AgentPersona] = None
self._load_persona()
def _load_persona(self):
"""从JSON文件加载Agent Persona。"""
try:
with open(self.persona_file_path, 'r', encoding='utf-8') as f:
persona_data = json.load(f)
self._persona = AgentPersona(**persona_data)
print(f"成功加载静态人设文件: {self.persona_file_path}")
except FileNotFoundError:
print(f"错误: 静态人设文件未找到在 {self.persona_file_path}")
self._persona = None
except Exception as e:
print(f"加载静态人设文件时发生错误: {e}")
self._persona = None
def get_persona(self) -> Optional[AgentPersona]:
"""获取Agent的完整静态人设。"""
return self._persona
def get_static_anchors_text(self) -> Dict[str, str]:
"""获取所有静态文本锚点。"""
if self._persona:
return self._persona.static_anchors_text
return {}
# 模拟保存 Agent Persona 到文件
persona_data_to_save = CUSTOMER_SERVICE_AGENT_PERSONA.model_dump()
with open("customer_service_persona.json", "w", encoding="utf-8") as f:
json.dump(persona_data_to_save, f, ensure_ascii=False, indent=2)
# 使用 StaticAnchorStore
static_store = StaticAnchorStore("customer_service_persona.json")
loaded_persona = static_store.get_persona()
if loaded_persona:
print("n--- 静态锚点检索示例 ---")
print(f"Agent 名称: {loaded_persona.name}")
print(f"Agent 角色: {loaded_persona.role}")
print(f"问候语: {loaded_persona.static_anchors_text.get('greeting_phrase')}")
4.3 动态锚点：事件驱动与摘要
动态锚点需要更复杂的存储和检索机制。
4.3.1 事件驱动锚点 (使用KV Store)
我们将使用一个简单的字典模拟键值存储，实际生产环境会用Redis等。
import time
class DynamicAnchorStore:
"""
动态锚点存储器，模拟键值数据库，存储会话或用户特定的动态信息。
"""
def __init__(self):
self._store: Dict[str, Dict[str, str]] = {} # {session_id: {anchor_key: anchor_value}}
def add_anchor(self, session_id: str, key: str, value: str):
"""
为特定会话添加或更新一个动态锚点。
"""
if session_id not in self._store:
self._store[session_id] = {}
self._store[session_id][key] = value
print(f"会话 {session_id} 更新动态锚点: {key}='{value}'")
def get_anchors_for_session(self, session_id: str) -> Dict[str, str]:
"""
获取特定会话的所有动态锚点。
"""
return self._store.get(session_id, {})
def remove_anchor(self, session_id: str, key: str):
"""
移除特定会话的某个动态锚点。
"""
if session_id in self._store and key in self._store[session_id]:
del self._store[session_id][key]
print(f"会话 {session_id} 移除动态锚点: {key}")
# 示例使用
dynamic_store = DynamicAnchorStore()
session_id_1 = "user_123_session_abc"
print("n--- 动态锚点（事件驱动）示例 ---")
# 用户在对话中声明了偏好
dynamic_store.add_anchor(session_id_1, "user_preference_style", "希望回答简短直接")
# Agent 做出承诺
dynamic_store.add_anchor(session_id_1, "agent_commitment_follow_up", "已记录您的问题，24小时内提供解决方案")
# 检索动态锚点
session_anchors = dynamic_store.get_anchors_for_session(session_id_1)
print(f"会话 {session_id_1} 的动态锚点: {session_anchors}")
# 模拟一段时间后，承诺已完成，移除锚点
# time.sleep(1) # 实际场景中会通过事件触发
# dynamic_store.remove_anchor(session_id_1, "agent_commitment_follow_up")
# print(f"会话 {session_id_1} 的动态锚点 (移除后): {dynamic_store.get_anchors_for_session(session_id_1)}")4.3.2 摘要锚点 (使用向量数据库进行语义检索)
这部分需要一个LLM来进行摘要，并一个向量数据库来存储和检索。我们将使用一个简化版本，用HuggingFace的SentenceTransformer模拟Embedding，用一个字典模拟向量数据库的存储和检索。
from sentence_transformers import SentenceTransformer
from sklearn.metrics.pairwise import cosine_similarity
import numpy as np
class VectorMemoryStore:
"""
向量记忆存储器，模拟向量数据库，存储摘要锚点并支持语义检索。
"""
def __init__(self, embedding_model_name: str = 'paraphrase-multilingual-MiniLM-L12-v2'):
self.model = SentenceTransformer(embedding_model_name)
self._memory_bank: List[Dict] = [] # 存储 {text: str, embedding: np.array, metadata: dict}
self._next_id = 0
print(f"加载 Embedding 模型: {embedding_model_name}")
def _get_embedding(self, text: str) -> np.array:
"""生成文本的嵌入向量。"""
return self.model.encode(text, convert_to_tensor=False)
def add_memory(self, text: str, metadata: Optional[Dict] = None):
"""
添加一段新的记忆（摘要锚点）到向量存储。
"""
embedding = self._get_embedding(text)
memory_item = {
"id": self._next_id,
"text": text,
"embedding": embedding,
"metadata": metadata if metadata is not None else {}
}
self._memory_bank.append(memory_item)
self._next_id += 1
print(f"添加记忆锚点 (ID: {memory_item['id']}): '{text}'")
return memory_item['id']
def retrieve_similar_memories(self, query_text: str, top_k: int = 3, threshold: float = 0.5) -> List[Dict]:
"""
根据查询文本，检索语义最相似的记忆锚点。
"""
if not self._memory_bank:
return []
query_embedding = self._get_embedding(query_text)
similarities = []
for item in self._memory_bank:
similarity = cosine_similarity([query_embedding], [item["embedding"]])[0][0]
similarities.append((similarity, item))
# 按照相似度降序排序
similarities.sort(key=lambda x: x[0], reverse=True)
results = []
for sim, item in similarities:
if sim >= threshold: # 过滤掉相似度低于阈值的
results.append({"score": sim, "text": item["text"], "metadata": item["metadata"]})
if len(results) >= top_k:
break
return results
# 模拟 LLM 摘要功能
def summarize_conversation_chunk(conversation_chunk: List[Dict]) -> str:
"""
模拟 LLM 对一段对话进行摘要。
conversation_chunk: [{"role": "user", "content": "..."}, {"role": "assistant", "content": "..."}]
"""
full_text = " ".join([msg["content"] for msg in conversation_chunk])
# 实际这里会调用LLM API进行摘要，例如：
# response = openai.ChatCompletion.create(
# model="gpt-3.5-turbo",
# messages=[
# {"role": "system", "content": "你是一个摘要助手，请精炼地总结以下对话。"},
# {"role": "user", "content": full_text}
# ]
# )
# return response.choices[0].message.content.strip()
# 简化模拟：直接截取或简单合并
if len(full_text) > 100:
return f"摘要: {full_text[:97]}..."
return f"摘要: {full_text}"
# 示例使用
vector_store = VectorMemoryStore()
session_id_2 = "user_456_session_xyz"
print("n--- 动态锚点（摘要与向量检索）示例 ---")
# 模拟一段对话
conversation_history_part1 = [
{"role": "user", "content": "我的订单S12345为什么还没发货？已经两天了。"},
{"role": "assistant", "content": "抱歉给您带来不便。我查询到您的订单S12345正在仓库打包，预计今天下午会发出。"},
{"role": "user", "content": "好的，谢谢。另外，你们的退货政策是怎样的？"},
{"role": "assistant", "content": "我们的退货政策是收到商品后7天内无理由退货，但需保持商品完好。"}
]
summary1 = summarize_conversation_chunk(conversation_history_part1)
vector_store.add_memory(summary1, {"session_id": session_id_2, "type": "conversation_summary", "timestamp": time.time()})
# 模拟另一段对话，可能发生在同一会话或不同会话
conversation_history_part2 = [
{"role": "user", "content": "我上次咨询的订单S12345发货了吗？"},
{"role": "assistant", "content": "是的，订单S12345已于昨天下午发出，预计明天送达。"},
{"role": "user", "content": "太好了！我还想问一下，如果我收到的商品有损坏怎么办？"}
]
summary2 = summarize_conversation_chunk(conversation_history_part2)
vector_store.add_memory(summary2, {"session_id": session_id_2, "type": "conversation_summary", "timestamp": time.time()})
# 用户发出新的查询
user_query = "关于订单S12345的最新状态是什么？或者退货流程。"
retrieved_memories = vector_store.retrieve_similar_memories(user_query, top_k=2, threshold=0.4)
print(f"n用户查询: '{user_query}'")
print("检索到的相关记忆锚点:")
for mem in retrieved_memories:
print(f" - 相似度: {mem['score']:.2f}, 内容: '{mem['text']}'")
4.4 上下文构建模块 (Context Builder)
这是将所有信息汇集到LLM Prompt中的关键环节。
class ContextBuilder:
"""
负责将所有相关信息（用户输入、对话历史、静态锚点、动态锚点、摘要锚点）
整合并格式化为 LLM 可以理解的上下文。
"""
def __init__(self,
static_store: StaticAnchorStore,
dynamic_store: DynamicAnchorStore,
vector_store: VectorMemoryStore,
max_history_tokens: int = 500):
self.static_store = static_store
self.dynamic_store = dynamic_store
self.vector_store = vector_store
self.max_history_tokens = max_history_tokens # 限制历史对话的令牌数
def build_context(self,
user_input: str,
session_id: str,
conversation_history: List[Dict],
llm_model_name: str = "gpt-3.5-turbo") -> List[Dict]:
"""
构建 LLM 的完整消息列表。
"""
messages = []
# 1. 注入系统角色和静态人设 (最高优先级)
persona = self.static_store.get_persona()
if persona:
messages.append({"role": "system", "content": persona.to_prompt_section()})
else:
messages.append({"role": "system", "content": "你是一个有帮助的AI助手。"})
# 2. 注入动态锚点 (事件驱动)
dynamic_anchors = self.dynamic_store.get_anchors_for_session(session_id)
if dynamic_anchors:
dynamic_prompt_parts = ["以下是当前会话的重要信息和用户偏好："]
for key, value in dynamic_anchors.items():
dynamic_prompt_parts.append(f"- {key}: {value}")
messages.append({"role": "system", "content": "n".join(dynamic_prompt_parts)})
# 3. 注入摘要锚点 (语义检索)
retrieved_summaries = self.vector_store.retrieve_similar_memories(user_input, top_k=2, threshold=0.4)
if retrieved_summaries:
summary_prompt_parts = ["以下是与当前问题相关的历史对话摘要："]
for i, summary in enumerate(retrieved_summaries):
summary_prompt_parts.append(f"{i+1}. (相似度: {summary['score']:.2f}) {summary['text']}")
messages.append({"role": "system", "content": "n".join(summary_prompt_parts)})
# 4. 注入精简的对话历史 (避免超出上下文窗口)
# 这是一个简化的令牌计数，实际需要使用LLM tokenizer
current_history_tokens = 0
history_to_add = []
for msg in reversed(conversation_history): # 从最新消息开始添加
# 简化：假设每个词1个token
msg_tokens = len(msg['content'].split()) + 5 # 加上角色和格式开销
if current_history_tokens + msg_tokens <= self.max_history_tokens:
history_to_add.insert(0, msg) # 保持原始顺序
current_history_tokens += msg_tokens
else:
break
messages.extend(history_to_add)
# 5. 注入当前用户输入
messages.append({"role": "user", "content": user_input})
return messages
# 模拟 LLM 交互
class LLMInteraction:
def __init__(self, api_key: str):
# 实际这里会初始化 OpenAI/Anthropic 等客户端
# self.client = OpenAI(api_key=api_key)
self.api_key = api_key # 仅为演示
def get_response(self, messages: List[Dict], model: str = "gpt-3.5-turbo") -> str:
"""
模拟调用 LLM API 获取回复。
"""
print(f"n--- LLM 输入消息 (模型: {model}) ---")
for msg in messages:
print(f"[{msg['role'].upper()}]: {msg['content']}")
# 模拟 LLM 回复
if "发货" in messages[-1]['content'] and "S12345" in messages[-1]['content']:
return "您的订单S12345已于昨天下午发出，预计明天送达。请问还有其他问题吗？"
elif "退货" in messages[-1]['content']:
return "我们的退货政策是收到商品后7天内无理由退货，但需保持商品完好。您可以在订单详情页申请退货。"
elif "希望回答简短直接" in messages[1]['content']: # 检查动态锚点是否生效
return "好的，我会尽量简洁地回答您的问题。请问您有什么问题？"
elif "小A" in messages[0]['content'] and "专业" in messages[0]['content']:
return "您好，我是小A，很高兴为您服务。请问有什么可以帮助您的？" # 模拟使用静态问候语
else:
return "感谢您的提问，我正在为您处理。请稍等。"
# 综合示例
print("n--- 综合 Agent 交互流程示例 ---")
# 初始化所有存储
static_store_instance = StaticAnchorStore("customer_service_persona.json")
dynamic_store_instance = DynamicAnchorStore()
vector_store_instance = VectorMemoryStore()
# 准备一些初始动态锚点和摘要锚点
session_id_complex = "user_789_session_complex"
dynamic_store_instance.add_anchor(session_id_complex, "user_language_preference", "中文")
dynamic_store_instance.add_anchor(session_id_complex, "user_tone_preference", "希望Agent回答语气亲切")
vector_store_instance.add_memory("用户曾咨询过订单S12345的发货情况，Agent已告知已发货。", {"session_id": session_id_complex})
vector_store_instance.add_memory("用户对退货政策有疑问，Agent已详细解释7天无理由退货。", {"session_id": session_complex})
# 初始化上下文构建器和LLM交互模块
context_builder = ContextBuilder(static_store_instance, dynamic_store_instance, vector_store_instance, max_history_tokens=200)
llm_agent = LLMInteraction(api_key="YOUR_OPENAI_API_KEY") # 替换为你的API Key
# 模拟对话历史
current_conversation_history = [
{"role": "user", "content": "你好，我想查一下我的订单S12345。"},
{"role": "assistant", "content": "您好！我是小A，很高兴为您服务。请问有什么可以帮助您的？"}
]
# 用户输入
user_query_1 = "我的订单S12345发货了吗？"
context_messages_1 = context_builder.build_context(user_query_1, session_id_complex, current_conversation_history)
agent_response_1 = llm_agent.get_response(context_messages_1)
print(f"nAgent 回复: {agent_response_1}")
current_conversation_history.append({"role": "user", "content": user_query_1})
current_conversation_history.append({"role": "assistant", "content": agent_response_1})
user_query_2 = "我想了解一下退货流程。"
context_messages_2 = context_builder.build_context(user_query_2, session_id_complex, current_conversation_history)
agent_response_2 = llm_agent.get_response(context_messages_2)
print(f"nAgent 回复: {agent_response_2}")
current_conversation_history.append({"role": "user", "content": user_query_2})
current_conversation_history.append({"role": "assistant", "content": agent_response_2})
# 再次提问，期望Agent能记住用户偏好和历史
user_query_3 = "我还有个问题，你们的增值服务有哪些？请亲切地回答我。"
context_messages_3 = context_builder.build_context(user_query_3, session_id_complex, current_conversation_history)
agent_response_3 = llm_agent.get_response(context_messages_3)
print(f"nAgent 回复: {agent_response_3}")
4.5 人设监控与反馈 (Adaptive Anchors)
这部分通常涉及更复杂的机器学习模型和人工标注流程。这里我们提供一个简化的反馈更新机制。
class PersonaFeedbackSystem:
"""
模拟人设监控与反馈系统，根据反馈调整Agent的人设锚点。
"""
def __init__(self, static_store: StaticAnchorStore):
self.static_store = static_store
def process_feedback(self, feedback: Dict):
"""
处理用户或系统反馈，尝试更新静态人设锚点。
feedback 示例: {"type": "tone_issue", "severity": "high", "description": "Agent语气过于生硬", "suggested_change": "更亲切友善"}
"""
print(f"n--- 处理反馈: {feedback} ---")
persona = self.static_store.get_persona()
if not persona:
print("错误: 无法获取Agent人设进行更新。")
return
if feedback.get("type") == "tone_issue" and feedback.get("severity") == "high":
current_tone = persona.tone
suggested_tone = feedback.get("suggested_change")
if suggested_tone and current_tone != suggested_tone:
print(f"检测到人设语气问题，尝试将语气从 '{current_tone}' 调整为 '{suggested_tone}'。")
persona.tone = suggested_tone
# 在实际系统中，这里会持久化更新到数据库或配置文件
# For demo, we just update the in-memory object
print(f"人设语气已更新为: {persona.tone}")
# 重新保存更新后的 persona
updated_persona_data = persona.model_dump()
with open(self.static_store.persona_file_path, "w", encoding="utf-8") as f:
json.dump(updated_persona_data, f, ensure_ascii=False, indent=2)
# 重新加载以确保一致性 (或者直接修改内存中的_persona对象)
self.static_store._load_persona()
else:
print("无需调整或建议的语气不明确。")
else:
print("当前反馈类型无需人设调整。")
# 示例使用
feedback_system = PersonaFeedbackSystem(static_store_instance)
# 模拟收到用户反馈
feedback_data = {
"type": "tone_issue",
"severity": "high",
"description": "Agent回答过于生硬，缺乏亲和力。",
"suggested_change": "亲切、友好、耐心"
}
feedback_system.process_feedback(feedback_data)
# 验证人设是否已更新
updated_persona = static_store_instance.get_persona()
if updated_persona:
print(f"更新后 Agent 语气: {updated_persona.tone}")
# 再次构建上下文，验证新语气是否生效
user_query_4 = "你好，请问你们公司都有哪些产品？"
context_messages_4 = context_builder.build_context(user_query_4, session_id_complex, current_conversation_history)
agent_response_4 = llm_agent.get_response(context_messages_4)
print(f"nAgent 回复 (语气调整后): {agent_response_4}")
五、高级策略与考量
5.1 锚点优先级与上下文窗口管理
- 优先级： 并非所有锚点都同等重要。静态锚点通常具有最高优先级，其次是与当前用户或会话强相关的动态锚点，最后是语义检索出的摘要锚点。在构建上下文时，应按照优先级顺序填充，并确保核心锚点始终被包含。
- 令牌预算： LLM的上下文窗口是有限的。我们需要为不同类型的锚点和对话历史设定令牌预算。当总令牌数接近上限时，需要策略性地截断低优先级信息或进行更激进的摘要。例如，可以采用滑动窗口机制来管理对话历史，只保留最近的N轮对话。
5.2 锚点刷新与衰减机制
- 防止信息过时： 动态锚点，特别是摘要锚点，可能会随着时间推移而变得不相关或过时。需要引入刷新（Refresh）和衰减（Decay）机制。
- 刷新： 定期对旧的摘要进行重新评估或与新信息合并，生成更精炼、更全面的摘要。
- 衰减： 为锚点设置“有效期”或“热度”分数。长时间未被使用或被新信息覆盖的锚点，其优先级或权重会降低，甚至被移除。
5.3 锚点冲突解决
当不同锚点之间，或锚点与当前用户输入之间出现信息冲突时，Agent需要一套明确的解决策略：
- 优先级规则： 预设优先级最高的锚点胜出。例如，核心原则锚点（“绝不提供医疗建议”）应优先于用户要求Agent提供医疗建议的输入。
- LLM决策： 在复杂冲突中，可以将冲突信息同时提供给LLM，并明确指示LLM根据预设规则进行判断和选择。例如：“以下信息可能存在冲突：[信息A] 和 [信息B]。请根据你的核心原则，选择并给出最合适的回答。”
- 人工介入： 对于严重或难以自动解决的冲突，应触发人工审核和干预流程，由人来更新或修正锚点。
5.4 安全性与隐私保护
- 敏感信息处理： 记忆锚点可能包含用户偏好、历史记录等敏感信息。在存储和使用这些锚点时，必须严格遵守数据隐私法规（如GDPR、CCPA）。
- 脱敏处理： 对个人身份信息（PII）进行脱敏或加密。
- 权限控制： 严格限制对锚点存储的访问权限。
- 生命周期管理： 设定锚点的存储期限，过期自动删除。
- 避免偏见： 动态锚点可能无意中捕捉到用户或Agent的偏见。需要定期审计锚点内容，确保其不会导致Agent产生不公平或歧视性的行为。
5.5 评估与迭代
如何衡量人设一致性是否真的得到了改善？
- 人工评估： 最直接有效的方法。通过人工标注团队对Agent的回复进行打分，评估其人设、语气、知识准确性等。
- 自动化指标：
- 关键词/短语匹配： 检查Agent回复中是否包含特定的人设关键词（如Agent的名字、核心原则中的短语）。
- 情感分析： 分析Agent回复的情感倾向是否与预设语气相符。
- 一致性评分模型： 训练一个专门的ML模型来评估Agent在不同对话轮次间的人设一致性。
- 用户满意度： 最终的衡量标准。通过用户反馈、满意度调查（CSAT）等指标来间接评估人设一致性的效果。
持续的评估和迭代是优化记忆锚点策略的关键。
六、展望：Agent 的“心智”构建
记忆锚点机制为我们构建了Agent的“长期记忆”和“核心价值观”，使其能够在一个庞大、动态的对话环境中保持其独特的“人格”。这不仅仅是简单的信息注入，它代表着我们向构建更智能、更稳定、更具情感连接的对话 Agent 迈出了坚实的一步。通过精细化地管理这些“记忆锚点”，我们赋予了 Agent 稳定的心智，使其在与用户的每一次互动中，都能以始终如一、可信赖的形象出现，这无疑将极大地提升用户体验，并为Agent的广泛应用奠定坚实基础。
感谢大家的聆听！