您可以使用记忆库生成结构化个人资料,这些个人资料是使用 LLM 填充和更新的静态架构的数据结构。通过定义固定架构,您可以确保智能体能够立即、低延迟地访问不断变化的信息,而无需在会话期间执行昂贵的搜索操作。
如需完成本指南中演示的步骤,您必须先按照设置记忆库中的步骤操作。
概览
在代理中使用结构化个人资料,可确保提取的信息(例如用户的技术栈或偏好设置)能够快速、简洁且以一致的格式提供。例如,您可以检索包含以下内容的用户个人资料:
MemoryProfile(
profile={
"technical_stacks": "ADK, Python",
"preferred_language": "Python",
"tone_preference": "Succinct"
},
schema_id="user-profile"
)
结构化个人资料针对低延迟检索进行了优化,因为整理信息的繁重工作是在生成时完成的。它们非常适合用于初始化代理与用户之间的互动。
了解结构化内存分析器
结构化个人资料是通过您用于自然语言记忆的相同方法(GenerateMemories 和 IngestEvents)生成的。定义架构后,Memory Bank 会自动尝试使用提供的数据源生成与架构一致的配置文件。
使用结构化个人资料时,记忆库会在记忆生成过程中执行以下操作:
- 提取:从数据源中提取符合架构的信息和上下文。系统只会提取与架构一致的信息。您可以使用记忆修订版本检查提取的信息和上下文。
- 整合:更新(如有必要)个人资料中的现有字段。LLM 将根据新提取的信息和上下文来判断如何更新现有内容。如果该字段在个人资料中尚不存在,系统将跳过合并,并直接使用提取的信息更新该字段。
系统会根据您在将数据提取到记忆库时提供的 scope(例如 {"user_id": "123"})来隔离个人资料。对于每个架构和范围,记忆库都会维护一个单一的配置文件作为可信来源。生成的配置文件由一个或多个 Memory 实例组成。每个 Memory 实例都代表个人资料中的一个字段,您可以使用该实例来检查字段的元数据和修订历史记录,例如:
Memory(
create_time=datetime.datetime(...),
memory_type=<MemoryType.STRUCTURED_PROFILE: 'STRUCTURED_PROFILE'>,
name='projects/.../locations/.../reasoningEngines/.../memories/...',
scope={
'user_id': '123'
},
structured_content=MemoryStructuredContent(
data={
'language_preference': 'Java'
},
schema_id='user-profile'
),
update_time=datetime.datetime(...),
expire_time=datetime.datetime(...),
metadata={...}
)
架构定义
记忆库生成的配置文件与记忆库实例中定义的架构保持一致。您可以定义多个独立的配置文件架构;每个架构都必须具有唯一 ID。请注意,添加或更新架构不会对历史记忆中的个人资料进行回填。
控制台
如需在 Google Cloud 控制台中的独立 Memory Bank 实例上添加或修改配置文件架构(需要 aiplatform.reasoningEngines.update 才能保存更改),请执行以下操作:
在 Google Cloud 控制台中,前往 Agent Platform 的记忆库页面,然后点击您的记忆库实例。
点击操作栏中的架构(或配置标签页上的生成部分)。
点击添加架构以创建具有唯一 ID 的架构,或点击现有架构卡片标题将其展开。
使用简单或 JSON 视图定义或修改架构字段,然后点击保存。
Agent Platform SDK
您可以使用 pydantic 模型来定义希望 Memory Bank 提取和维护的字段。例如:
from pydantic import BaseModel, Field
from typing import Literal
class UserProfile(BaseModel):
name: str = Field(
description="Name of the user.")
technical_stack: str = Field(
description="Comma-separated list tools or languages used by the user.")
primary_goal: str = Field(
description="The main objective the user is pursuing.")
expertise_level: str = Field(
description="Current skill level (e.g., Junior, Senior).")
job_status: Literal['unemployed', 'part_time', 'full_time', 'student'] = Field(
description="The job status of the individual")
在创建或更新记忆库实例时,将架构上传到记忆库:
schema_config = {
"id": "user-profile",
"memory_schema": UserProfile.model_json_schema()
}
memory_bank = client.memory_banks.create(
managed_semantic_memory_config={
"structured_memory_configs": [
{
"schema_configs": [schema_config]
}
]
}
)
默认情况下,记忆库会提取自然语言记忆。 如果您只想让记忆库生成个人资料,可以关闭自然语言记忆生成功能:
memory_bank = client.memory_banks.create(
managed_semantic_memory_config={
"structured_memory_configs": [
{
"schema_configs": [schema_config]
}
],
# Optional: Turn off natural language memories.
"unstructured_memory_configs": [
{"disable_natural_language_memories": True}
]
}
)
生成配置文件
您可以通过向 GenerateMemories 或 IngestEvents 方法提供对话历史记录(事件)来触发结构化记忆的生成。与自然语言记忆一样,记忆库使用 LLM 从数据源中提取有意义的信息,并将其与现有记忆整合。
示例
以下示例演示了如何使用之前定义的架构生成记忆个人资料。
在发送到记忆库的第一组事件中,用户表示他们使用 ADK 智能体,范围为 {"user_id": "123"}:
client.memory_banks.memories.generate(
name=memory_bank.name,
scope={"user_id": "123"},
direct_contents_source={
"events": [
{"content": {
"parts": [{
"text": "Can you help me build an ADK agent that organizes my daily tasks?"}]}}]
}
)
记忆库会提取“ADK”作为架构中 technical_stack 字段的值。由于提取的事件不包含架构其余部分的相关信息,因此不会填充其他字段。由于这是此范围内的首次互动,因此系统会跳过合并,并使用此初始值初始化配置文件。
result = client.memory_banks.memories.retrieve_profiles(
name=memory_bank.name,
scope={"user_id": "123"},
)
"""
Returns:
RetrieveProfilesResponse(
profiles={
'user-profile': MemoryProfile(
profile={
'technical_stack': 'ADK'
},
schema_id='user-profile'
)
}
)
"""
在发送到记忆库的下一组事件中,用户表明自己是一名学生,主要使用 Python 编写代码:
client.memory_banks.memories.generate(
name=memory_bank.name,
scope={"user_id": "123"},
direct_contents_source={
"events": [
{"content": {
"parts": [
{"text": "Do you have any career recommendations for students that specialize in Python?"}]}}]
}
)
记忆库会从互动中提取“Python”和“学生”身份。technical_stack 片段已合并,将“Python”附加到现有的“ADK”条目。系统会使用“student”枚举填充之前为空的 job_status 字段,并跳过合并步骤。
result = client.memory_banks.memories.retrieve_profiles(
name=memory_bank.name,
scope=scope
)
"""
Returns:
RetrieveProfilesResponse(
profiles={
'user-profile': MemoryProfile(
profile={
'technical_stack': 'ADK, Python',
'job_status': 'student'
},
schema_id='user-profile'
)
}
)
"""
个人资料检索
生成后,您可以使用 RetrieveProfiles 方法检索特定范围的合并配置文件。这将返回映射到您的架构的最新数据。
result = client.memory_banks.memories.retrieve_profiles(
name=memory_bank.name,
scope={"user_id": "123"},
)
# Accessing the data
for profile in result.profiles.values():
print(profile)
# Output: {'technical_stack': 'ADK, Python', 'job_status': 'student', ...}
商家资料检查
在后台,一个个人资料由 STRUCTURED_PROFILE 类型的各个记忆组成。架构中的每个字段都映射到每个范围内的单个记忆,从而提供精细的可观测性。
虽然 RetrieveProfiles 是在生产环境中检索个人资料的主要方法,但您可以在Google Cloud 控制台中或使用 Agent Platform SDK 检查单个记忆,以审核个人资料的演变:
- 字段级元数据:查看和更新配置文件中各个字段的特定存留时间 (TTL) 和
Memory.metadata。 修订历史记录:跟踪字段的沿袭,查看历史值以及触发每次更改的特定对话上下文。
控制台
在 Google Cloud 控制台中,在记忆库页面上打开独立实例,或在部署页面上打开已部署的代理,然后选择记忆标签页。
选择结构化,然后为目标范围内的所有键选择值(或在非结构化中应用范围过滤条件,然后点击搜索结构化记忆)。匹配的配置文件行会显示架构、类型、内容、创建时间和更新时间。
点击个人资料行中的查看,打开记忆详情面板,其中会显示范围条状标签、指向架构的链接、各个记忆事实(每个事实都有查看修订版本选项),以及合并后的个人资料 JSON。
Agent Platform SDK
使用 RetrieveMemories 检索包含用户个人资料片段的记忆。默认情况下,RetrieveMemories 仅检索自然语言记忆,因此请明确请求 STRUCTURED_PROFILE 记忆:
client.memory_banks.memories.retrieve(
name="...",
scope={"user_id": "123"},
config={
"memory_types": ["STRUCTURED_PROFILE"]
}
)
"""
Returns:
[RetrieveMemoriesResponseRetrievedMemory(
memory=Memory(
create_time=datetime.datetime(...),
memory_type=<MemoryType.STRUCTURED_PROFILE: 'STRUCTURED_PROFILE'>,
name='projects/.../locations/.../reasoningEngines/.../memories/...',
scope={
'user_id': '123'
},
structured_content=MemoryStructuredContent(
data={
'technical_stack': 'ADK, Python'
},
schema_id='user'
),
update_time=datetime.datetime(...)
)
)]
"""
然后,检索结构化个人资料片段的修订历史记录,以检查个人资料字段随时间的变化情况以及每次更改的相关背景信息:
for retrieved_memory in list(results):
list(client.memory_banks.memories.revisions.list(
name=retrieved_memory.memory.name
))
"""
Returns:
[MemoryRevision(
create_time=datetime.datetime(...),
expire_time=datetime.datetime(...),
extracted_memories=[
IntermediateExtractedMemory(
context='The user indicated that they have expertise in Python when asking about career options.',
structured_data={
'technical_stack': 'Python'
}
),
],
name='projects/.../locations/.../reasoningEngines/.../memories/.../revisions/...',
structured_data={
'technical_stack': 'ADK, Python'
}
),
MemoryRevision(
create_time=datetime.datetime(...),
expire_time=datetime.datetime(...),
extracted_memories=[
IntermediateExtractedMemory(
context='The user indicated that they need help building an ADK agent',
structured_data={
'technical_stack': 'ADK'
}
),
],
name='projects/.../locations/.../reasoningEngines/.../memories/.../revisions/...',
structured_data={
'technical_stack': 'ADK'
}
)]
"""