Sitemap
Welcome Offer
Access to everything. Now 30% off.
Upgrade now
Upgrade now
Google Cloud - Community

A collection of technical articles and blogs published or curated by Google Cloud Developer Advocates. The views expressed are those of the authors and don't necessarily reflect those of Google.

Create an Agent that Remembers with Agent Platform Memory Bank利用 Agent Platform Memory Bank 创建具备记忆功能的智能体

Bhandari Haren
Bhandari HarenBhandari Haren
12 min read阅读需 12 分钟3 days ago

Imagine having a conversation with an expert financial advisor, personal trainer, or research partner. You discuss your goals, share your background, critique ideas, and lay out long-term plans. Now imagine that the next morning, you call them back-and they have no idea who you are, what you talked about, or what decisions you reached together.试想一下,你正在与一位专业的财务顾问、私人教练或研究伙伴进行对话。你们讨论目标、分享背景、推敲想法,并制定长期计划。现在,想象一下第二天早上你再联系他们时,他们却完全不知道你是谁,你们讨论过什么,或者你们共同达成了哪些决定。

LLMs are completely stateless so they do not have a built-in memory that keeps the record of the past interactions. While foundational models have grown exponentially in raw intelligence and context window capacity, raw context size does not equal memory. True intelligence requires continuity. To transition from transactional chatbots to autonomous, personalized digital colleagues, AI agents need robust, production-grade memory systems.大语言模型(LLM)完全是无状态的,因此它们没有内置的记忆功能来记录过去的交互。虽然基础模型在原始智能和上下文窗口容量方面呈指数级增长,但原始上下文大小并不等同于记忆。真正的智能需要连续性。要从事务型聊天机器人转型为自主、个性化的数字同事,AI 智能体需要健壮的、生产级的记忆系统。

In this guide, we will explore Agent Platform Memory Bank — a fully managed, cognitive memory infrastructure on Google Cloud. We will cover the mechanics of agent harnesses, dive into context engineering, dissect short-term vs. long-term memory architectures, and walk through building a persistent, memory-enabled agent from scratch using Google’s Agent Development Kit (ADK) and the Agents CLI (google-agents-cli).在本指南中,我们将探索 Agent Platform Memory Bank——这是 Google Cloud 上的一种全托管认知记忆基础设施。我们将介绍智能体架构(Agent Harness)的机制,深入探讨上下文工程,剖析短期与长期记忆架构,并演示如何使用 Google 的智能体开发套件(ADK)和 Agents CLI (google-agents-cli) 从零开始构建一个具备持久化记忆功能的智能体。

Introduction to Agents and the Agent Harness智能体与智能体架构简介

Before tackling memory, we must establish clear definitions for modern AI agents and the infrastructure that hosts them.在处理记忆问题之前,我们必须为现代 AI 智能体及其托管基础设施建立明确的定义。

What Is an Agent?什么是智能体?

  • Perception: Interpreting user input and environmental state.感知:解读用户输入和环境状态。
  • Reasoning & Planning: Decomposing complex objectives into actionable steps (e.g., ReAct, Chain-of-Thought, or Plan-and-Execute loops).推理与规划:将复杂目标分解为可执行的步骤(例如 ReAct、思维链或规划与执行循环)。
  • Tool Use : Interacting with external APIs, databases, calculators, and enterprise software.工具使用:与外部 API、数据库、计算器和企业软件进行交互。
  • Execution & Reflection: Iterating toward an objective, observing tool execution results, and self-correcting when errors occur.执行与反思:朝着目标进行迭代,观察工具执行结果,并在出现错误时进行自我修正。
Press enter or click to view image in full size按下回车键或点击以查看完整尺寸图片

What Is an Agent Harness?什么是智能体架构(Agent Harness)?

A raw LLM cannot execute shell commands, manage state across API calls, or enforce enterprise security boundaries on its own. It is simply a text-in, text-out predictive engine.原始的 LLM 无法自行执行 shell 命令、管理跨 API 调用的状态或强制执行企业安全边界。它仅仅是一个输入文本、输出文本的预测引擎。

Press enter or click to view image in full size

The Agent Harness is responsible for:智能体架构负责以下工作:

  1. Lifecycle Management: Instantiating agent sessions, preserving conversational turns, and handling graceful terminations.生命周期管理:实例化智能体会话、保留对话轮次并处理优雅终止。
  2. Tool Sandboxing: Resolving model function calls against executable tools, validating arguments, enforcing rate limits, and securing credentials.工具沙箱:将模型函数调用解析为可执行工具,验证参数,强制执行速率限制并保护凭据。
  3. Guardrails & Safety: Intercepting inputs and outputs to enforce safety policies, compliance checks, and prompt-injection mitigations.护栏与安全:拦截输入和输出,以执行安全策略、合规性检查和提示词注入缓解措施。
  4. Observability & Auditability: Emitting OpenTelemetry traces, latency breakdowns, token consumption metrics, and step-by-step reasoning logs to Cloud Logging and Cloud Trace.可观测性与可审计性:向 Cloud Logging 和 Cloud Trace 发送 OpenTelemetry 追踪、延迟明细、Token 消耗指标以及分步推理日志。
  5. Memory Orchestration: Intercepting turns to retrieve relevant historical context before the prompt reaches the LLM, and consolidating facts asynchronously after the turn concludes.记忆编排:在提示词到达 LLM 之前拦截对话轮次以检索相关的历史上下文,并在对话结束后异步整合事实。

Why Agent Memory Is Crucial为什么智能体记忆至关重要

A common misconception in the era of 1M+ and 2M+ token context windows is: “If our context window is virtually limitless, why do we need dedicated memory systems? Why not just pass the entire chat history into every request?”在拥有 100 万甚至 200 万以上 Token 上下文窗口的时代,一个常见的误区是:“如果我们的上下文窗口几乎是无限的,为什么还需要专门的记忆系统?为什么不直接把整个聊天记录传给每一个请求?”

The Failure Modes of Naive Context Dumping简单粗暴地堆砌上下文的失败模式

Passing unbounded conversational history directly into the context window triggers severe operational and cognitive bottlenecks:将无限的对话历史直接传入上下文窗口会引发严重的运营和认知瓶颈:

  1. Context Rot & “Lost in the Middle”: Even frontier models suffer from degraded recall and reasoning when buried under hundreds of thousands of tokens of raw conversational transcript. Crucial instructions or user preferences expressed 50 turns ago get diluted among irrelevant pleasantries and intermediate tool outputs.上下文腐烂与“迷失在中间”:即使是前沿模型,在被数十万个原始对话记录 Token 淹没时,其回忆和推理能力也会下降。50 轮对话前表达的关键指令或用户偏好,会被淹没在无关的寒暄和中间工具输出中。
  2. Quadratic Latency & Throughput Degradation: Time to First Token (TTFT) scales with prompt token length. Feeding 100,000 tokens of chat history on every turn results in latency spikes, frustrating real-time interactive users.二次方延迟与吞吐量下降:首字延迟(TTFT)随提示词长度的增加而增加。在每一轮对话中输入 10 万个 Token 的聊天记录会导致延迟飙升,令实时交互用户感到沮丧。
  3. Escalating Token Economics: Re-sending entire histories on every interaction consumes massive token volumes on both prompt caching misses and API billing.不断升级的 Token 成本:在每次交互中重新发送完整历史记录,会在提示词缓存未命中和 API 计费方面消耗海量 Token。
  4. Session Boundaries: When a user closes their browser or starts a new session a week later, the in-memory context window resets completely.会话边界:当用户关闭浏览器或一周后开始新会话时,内存中的上下文窗口会完全重置。
Press enter or click to view image in full size

Context Engineering上下文工程

Context Engineering is the deliberate art and science of curating the most information-dense, high-signal, and task-relevant context for the model at each turn.上下文工程是一门经过深思熟虑的艺术与科学,旨在为模型在每一轮交互中策划信息密度最高、信号最强且与任务最相关的上下文。

  • The Context Window is the working registers and L1 cache.上下文窗口相当于工作寄存器和 L1 缓存。
  • The Session Buffer is RAM (fast, temporary, session-scoped).会话缓冲区相当于 RAM(快速、临时、会话范围内有效)。
  • The Memory Bank is NVMe persistent storage (durable, indexed, searchable, personalized across sessions).Memory Bank 相当于 NVMe 持久化存储(持久、可索引、可搜索,且跨会话个性化)。

Memory systems power context engineering by dynamically extracting core semantic facts, updating user profiles, discarding transient conversational scaffolding, and serving precise contextual knowledge just in time.记忆系统通过动态提取核心语义事实、更新用户画像、丢弃瞬时对话脚手架,并及时提供精确的上下文知识,从而赋能上下文工程。

Memory Architectures in Google Cloud Agent PlatformGoogle Cloud Agent Platform 中的记忆架构

Press enter or click to view image in full size

Short-Term Memory (Session Memory)短期记忆(会话记忆)

  • Scope: Bound to a single conversational session ID.范围:绑定到单个对话会话 ID。
  • Contents: The immediate sliding window of user-agent turns, system prompts, intermediate thought traces, and tool execution call/response pairs.内容:用户与智能体交互的即时滑动窗口、系统提示词、中间思维轨迹以及工具执行的调用/响应对。
  • Storage: Ephemeral in-memory caches, Agent Runtime Session Services, or fast key-value stores (e.g., Cloud Memorystore / Redis / Firestore).存储:瞬时内存缓存、Agent Runtime 会话服务或快速键值存储(如 Cloud Memorystore / Redis / Firestore)。
  • Lifecycle: Destroyed or archived when the session expires or the user resets the conversation.生命周期:在会话过期或用户重置对话时销毁或归档。

Long-Term Memory (Agent Platform Agent Runtime Memory Bank)长期记忆(Agent Platform Agent Runtime Memory Bank)

Scope: Bound to an entity (typically a user_id or an enterprise account), persisting across days, months, and distinct sessions.范围:绑定到实体(通常是 user_id 或企业账户),跨越数天、数月及不同会话持久存在。

Contents: Extracted facts, entity relationships, declared preferences, past goals, and behavioral patterns.内容:提取的事实、实体关系、声明的偏好、过去的目标和行为模式。

Become a Medium member

Mechanism:机制:

  • Autonomous Fact Extraction: When a conversation concludes or periodically throughout a session, an asynchronous background pipeline powered by Gemini reviews the transcript.自主事实提取:当对话结束时,或在会话期间定期,由 Gemini 驱动的异步后台流水线会审查对话记录。
  • Memory Reconciliation: The pipeline evaluates new statements against existing memories. If a user says “I recently moved from Austin to Seattle” , the system updates the previous residence fact rather than storing conflicting truths.记忆调和:流水线会根据现有记忆评估新陈述。如果用户说“我最近从奥斯汀搬到了西雅图”,系统会更新之前的居住地事实,而不是存储相互冲突的信息。
  • Vector Indexing & Semantic Search: Extracted memories are embedded and stored in managed vector indexes.向量索引与语义搜索:提取的记忆会被嵌入并存储在托管向量索引中。
  • Scoped Isolation: Memory spaces are strictly scoped per user and tenant, ensuring GDPR/CCPA compliance and preventing multi-tenant data leaks.范围隔离:记忆空间严格按用户和租户进行划分,确保符合 GDPR/CCPA 合规性并防止多租户数据泄露。

4. Setting Up Memory Bank for Long-Term Memory4. 为长期记忆设置 Memory Bank

Core Prerequisites核心先决条件

To utilize Agent Platform Agent Runtime and Memory Bank, ensure the following Google Cloud APIs are enabled in your project:要使用 Agent Platform Agent Runtime 和 Memory Bank,请确保在您的项目中启用了以下 Google Cloud API:

# Set your active GCP project
gcloud config set project YOUR_PROJECT_ID

# Enable the required APIs
gcloud services enable \
aiplatform.googleapis.com \
compute.googleapis.com \
run.googleapis.com \
artifactregistry.googleapis.com

You can run the above commands from Google Cloud Shell or authenticated local environment. gcloud您可以在 Google Cloud Shell 或已认证的本地环境中运行上述命令。gcloud

Architecture of the Memory Bank Service in ADKADK 中 Memory Bank 服务的架构

Google’s Agent Development Kit (ADK) exposes the VertexAiMemoryBankService (and its native client interfaces), providing an integrated bridge between your agent code and the cloud memory infrastructure:Google 的智能体开发套件(ADK)提供了 VertexAiMemoryBankService(及其原生客户端接口),为您的智能体代码与云记忆基础设施之间提供了集成的桥梁:

Press enter or click to view image in full size

Memory Extraction Customization记忆提取自定义

Memory Bank allows you to define Extraction Scopes and Custom Topics. For instance, an enterprise sales agent might focus memory extraction on:Memory Bank 允许您定义提取范围和自定义主题。例如,企业销售智能体可能会将记忆提取重点放在:

  • Budget预算
  • Decision Makers决策者
  • Procurement采购
  • Pain Points痛点

Meanwhile, a personal assistant agent would focus extraction on:与此同时,个人助理智能体会将提取重点放在:

  • Dietary Restrictions饮食限制
  • Frequent Flyer Numbers常旅客号码
  • Family Members家庭成员
  • Working Hours工作时间

5. Building an Agent with Agents CLI5. 使用 Agents CLI 构建智能体

Google provides Agents CLI (google-agents-cli), a developer tool and coding assistant skill package built to streamline the entire lifecycle of ADK agents: from scaffolding to local execution, evaluation, and deployment onto Google Cloud Agent Runtime.Google 提供了 Agents CLI (google-agents-cli),这是一个开发工具和编码助手技能包,旨在简化 ADK 智能体的整个生命周期:从脚手架搭建到本地执行、评估以及部署到 Google Cloud Agent Runtime。

Step 1: Install and Initialize Agents CLI第一步:安装并初始化 Agents CLI

The recommended way to run agents-cli is via uvx (from the high-performance Python package manager uv):推荐通过 uvx(来自高性能 Python 包管理器 uv)来运行 agents-cli:

# Setup the Agents CLI and install required developer skills
uvx google-agents-cli setup

Step 2: Scaffold a New ADK Project第二步:搭建新的 ADK 项目脚手架

The setup command verifies your Google Cloud authentication (gcloud auth application-default login), prepares Python virtual environments, and installs the specialized Agent Development Kit skills into your coding agent environment.setup 命令会验证您的 Google Cloud 身份认证 (gcloud auth application-default login),准备 Python 虚拟环境,并将专门的智能体开发套件技能安装到您的编码智能体环境中。

agents-cli scaffold create concierge-agent --prototype --yes

Let’s inspect the generated directory structure:让我们检查生成的目录结构:

concierge-agent/
├── .adk/ # ADK metadata and environment state
├── pyproject.toml # Project dependencies (google-adk, google-cloud-aiplatform)
├── README.md # Quickstart instructions
├── tests/ # Unit and integration test suites
│ └── test_agent.py
└── src/
└── concierge_agent/
├── __init__.py
├── agent.py # Core agent definition, prompt, and tools
└── tools.py # Custom tool implementations

Step 3: Implement the Persistent Memory Agent Code第三步:实现持久化记忆智能体代码

In your project, dependencies should include google-adk and the Agent Platform libraries. Let’s write src/concierge_agent/agent.py to leverage the Agent Platform Memory Bank Service.在您的项目中,依赖项应包含 google-adk 和 Agent Platform 库。让我们编写 src/concierge_agent/agent.py 来利用 Agent Platform Memory Bank 服务。

As you can see we are adding 4 different kind of long term memory topics,如您所见,我们正在添加 4 种不同类型的长期记忆主题:

-user_preferences-user_preferences(用户偏好)

-dietary_restrictions-dietary_restrictions(饮食限制)

-travel_habits-travel_habits(旅行习惯)

-personal_projects-personal_projects(个人项目)



import os
from google.adk.agents import Agent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.memory import VertexAiMemoryBankService
from google.genai import types

PROJECT_ID = os.getenv("GOOGLE_CLOUD_PROJECT", "your-gcp-project-id")
LOCATION = os.getenv("GOOGLE_CLOUD_REGION", "us-central1")

PROJECT_ID = os.getenv("GOOGLE_CLOUD_PROJECT", default_project_id)
LOCATION = os.getenv("GOOGLE_CLOUD_LOCATION", "global")
AGENT_ENGINE_ID = os.getenv("AGENT_ENGINE_ID") or os.getenv(
"GOOGLE_CLOUD_AGENT_ENGINE_ID"
)

# 1. Initialize the Long-Term Memory Service
# VertexAiMemoryBankService connects directly to the Agent Runtime backend when configured
if AGENT_ENGINE_ID:
memory_service = VertexAiMemoryBankService(
project=PROJECT_ID,
location=LOCATION,
agent_engine_id=AGENT_ENGINE_ID,
)
else:
memory_service = InMemoryMemoryService()


async def generate_memories_callback(callback_context: CallbackContext) -> None:
"""Sends the session's events to Memory Bank for automatic memory generation."""
try:
await callback_context.add_session_to_memory()
except ValueError:
# Memory service is not attached to this runner (e.g. in basic tests)
pass
return None


# 2. Define the Agent Instructions and Tools
agent = Agent(
name="concierge_agent",
model="gemini-2.5-pro",
instruction="""
You are an intelligent, highly personalized executive concierge assistant.
Your goal is to assist the user with recommendations, scheduling, and planning.

CRITICAL INSTRUCTIONS FOR MEMORY:
1. Always consult your recalled memories about the user to personalize your answers.
2. If the user mentions a personal preference, allergy, favorite place, or goal, acknowledge it naturally.
3. Never contradict established user memories unless the user explicitly updates them.
"
"",
# The agent preloads memories into context and saves new memories after turns
tools=[PreloadMemoryTool()],
after_agent_callback=generate_memories_callback,
)

root_agent = agent
app = App(name="concierge_agent", root_agent=agent)

# 3. Configure the Runner (Agent Harness)
# The Runner orchestrates Session state, Long-Term Memory, and Model Invocation
runner = Runner(
app=app,
session_service=InMemorySessionService(), # Short-term memory buffer
memory_service=memory_service, # Long-term persistent Memory Bank
)


def handle_user_message(user_id: str, session_id: str, message_text: str) -> str:
"""Process an incoming turn through the memory-augmented harness.

Args:
user_id: The unique identifier for the user (memory scope).
session_id: The identifier for the current conversational session.
message_text: The user's input prompt.
"
""
# Execute the turn through the runner.
# The runner automatically:
# 1. Retrieves relevant facts from Memory Bank scoped to `user_id`.
# 2. Injects them into the system context.
# 3. Executes the agent reasoning loop.
# 4. Triggers background fact consolidation into the Memory Bank.
events = runner.run(
user_id=user_id,
session_id=session_id,
new_message=types.Content(
role="user",
parts=[types.Part.from_text(text=message_text)],
),
)
texts = []
for event in events:
if event.content and event.content.parts:
for part in event.content.parts:
if part.text:
texts.append(part.text)
return "".join(texts)

Step 4: Enhancing the Project with Deployment Scaffolding第四步:通过部署脚手架增强项目

cd concierge-agent
agents-cli scaffold enhance --deployment-target agent_runtime

Step 5: Deploy the Agent第五步:部署智能体

To deploy the agent to agent runtime, use the following command:要将智能体部署到 Agent Runtime,请使用以下命令:

cd concierge-agent
agents-cli scaffold enhance --deployment-target agent_runtime

Testing the Memory Lifecycle Across Sessions测试跨会话的记忆生命周期

To appreciate the power of Google Agent Runtime Memory Bank, observe how an agent behaves across two distinct sessions with cold-start isolation.为了体验 Google Agent Runtime Memory Bank 的强大功能,请观察智能体如何在两个具有冷启动隔离的不同会话中表现。

Session 1: Imparting Knowledge (Monday Morning)会话 1:传授知识(周一上午)

Once you deploy the Agent in Agent Runtime, run the remote agent using the following code.一旦您在 Agent Runtime 中部署了智能体,请使用以下代码运行远程智能体。

import argparse
import asyncio
import json
import os
import re
from collections.abc import AsyncGenerator
from pathlib import Path
from typing import Any

import agentplatform


def sanitize_session_id(session_id: str) -> str:
"""

Agent Runtime requires session_id to contain only lowercase letters, digits,
and hyphens, starting and ending with an alphanumeric character.
"
""
sanitized = session_id.lower().replace("_", "-")
sanitized = re.sub(r"[^a-z0-9-]+", "-", sanitized)
return sanitized.strip("-")


class DeployedAgentClient:

def __init__(
self,
remote_agent_runtime_id: str | None = None,
location: str | None = None,
metadata_path: str = "deployment_metadata.json",
):
if not remote_agent_runtime_id:
metadata_file = Path(metadata_path)
if not metadata_file.exists():
raise FileNotFoundError(
f"Deployment metadata file not found at {metadata_path}. "
"Please specify remote_agent_runtime_id explicitly."
)
with open(metadata_file) as f:
metadata = json.load(f)
remote_agent_runtime_id = metadata["remote_agent_runtime_id"]

self.remote_agent_runtime_id = remote_agent_runtime_id

if not location:
# Parse location from resource name:
# projects/{project}/locations/{location}/reasoningEngines/{engine_id}
parts = remote_agent_runtime_id.split("/")
if len(parts) >= 4 and parts[2] == "locations":
location = parts[3]
else:
location = os.getenv("GOOGLE_CLOUD_LOCATION", "us-central1")

self.location = location
self.platform_client = agentplatform.Client(location=self.location)
self.agent = self.platform_client.agent_engines.get(
name=self.remote_agent_runtime_id
)

async def stream_query(
self,
message: str,
user_id: str,
session_id: str | None = None,
) -> AsyncGenerator[dict[str, Any], None]:
"""Streams events from the deployed agent."""
valid_session_id = sanitize_session_id(session_id) if session_id else None
async for event in self.agent.async_stream_query(
message=message,
user_id=user_id,
session_id=valid_session_id,
):
yield event

async def query(
self,
message: str,
user_id: str,
session_id: str | None = None,
) -> str:
"""Sends a query to the agent and returns the accumulated text response."""
full_text: list[str] = []
async for event in self.stream_query(
message=message,
user_id=user_id,
session_id=session_id,
):
if isinstance(event, dict) and event.get("errorCode"):
raise RuntimeError(f"Error from agent: {event.get('errorMessage')}")

content = event.get("content", {})
parts = content.get("parts", [])
for part in parts:
text = part.get("text")
if text:
full_text.append(text)

return "".join(full_text)

async def add_session_to_memory(
self,
user_id: str,
session_id: str,
) -> None:

valid_session_id = sanitize_session_id(session_id)
session = await self.agent.async_get_session(
user_id=user_id,
session_id=valid_session_id,
)
if not session:
raise ValueError(f"Session {session_id} not found for user {user_id}")
await self.agent.async_add_session_to_memory(session=session)

async def search_memory(
self,
user_id: str,
query: str = "*",
) -> list[str]:
"""Searches persistent Memory Bank for extracted facts/memories."""
response = await self.agent.async_search_memory(user_id=user_id, query=query)
memories = response.get("memories", [])
facts = []
for m in memories:
content = m.get("content", {})
for part in content.get("parts", []):
text = part.get("text")
if text:
facts.append(text)
return facts


async def main():
parser = argparse.ArgumentParser(description="Call deployed Agent Runtime agent.")
parser.add_argument(
"--user-id",
default="alice_99",
help="User ID for the session context",
)
parser.add_argument(
"--session-id",
default="sess_001",
help="Session ID for the conversation",
)
parser.add_argument(
"--message",
default=(
"Hi, I'm Alice. I'm training for the Boston Marathon this spring, \n"
" and I adhere to a strict gluten-free pescatarian diet."
),
help="User message to send to the agent",
)
parser.add_argument(
"--record-to-memory",
action="store_true",
default=True,
help="Trigger Memory Bank extraction for this session after the turn",
)
parser.add_argument(
"--search-memory",
action="store_true",
help="Search and print memories for the user instead of querying",
)
args = parser.parse_args()

client = DeployedAgentClient()

if args.search_memory:
print(f"Searching Memory Bank for user {args.user_id}...")
facts = await client.search_memory(user_id=args.user_id)
if not facts:
print("No memories found.")
else:
print("Recorded memories in Memory Bank:")
for fact in facts:
print(f" • {fact}")
return

print("Initializing client...")
print(f"Connected to agent: {client.remote_agent_runtime_id}")
print(f"Location: {client.location}")
print(f"User ID: {args.user_id}")
print(f"Session ID: {args.session_id}")
print(f"Message: {args.message.strip()}\n")
print("--- Streaming Agent Response ---")

async for event in client.stream_query(
message=args.message,
user_id=args.user_id,
session_id=args.session_id,
):
if isinstance(event, dict) and event.get("errorCode"):
print(f"\n[Error]: {event.get('errorMessage')}")
continue

content = event.get("content", {})
parts = content.get("parts", [])
for part in parts:
text = part.get("text")
if text:
print(text, end="", flush=True)

print("\n\n--- End of Response ---")

if args.record_to_memory:
print("\nTriggering Memory Bank consolidation...")
await client.add_session_to_memory(
user_id=args.user_id, session_id=args.session_id
)
print("Memory extraction request submitted successfully.")


if __name__ == "__main__":
asyncio.run(main())

At the end of Session 1, the Agent Harness sends the turn transcripts to the Memory Bank background extraction pipeline. The Gemini extraction engine extracts structured facts:在会话 1 结束时,智能体架构会将对话记录发送到 Memory Bank 后台提取流水线。Gemini 提取引擎会提取结构化事实:

Press enter or click to view image in full size

Session 2: Cold Start Retrieval (Friday Evening)会话 2:冷启动检索(周五晚上)

Context: user_id = “alice_99”, session_id = “sess_999” (Completely new session, clean memory cache)上下文:user_id = “alice_99”, session_id = “sess_999”(全新的会话,干净的内存缓存)

Notice that the user provides no context about their diet or athletic training:请注意,用户没有提供关于其饮食或运动训练的任何上下文:

Press enter or click to view image in full size

What Happened Behind the Scenes?幕后发生了什么?

Press enter or click to view image in full size
  1. The client started sess_999 with zero conversational turns in the session buffer.客户端启动了 sess_999,会话缓冲区中没有任何对话轮次。
  2. The Agent Harness extracted the intent ( dining , recovery run , Back Bay ) and queried the Memory Bank for alice_99 .智能体架构提取了意图(餐饮、恢复跑、Back Bay)并查询了 alice_99 的 Memory Bank。
  3. Memory Bank performed a semantic search over Alice’s memory vectors and surfaced her marathon training and dietary requirements.Memory Bank 对 Alice 的记忆向量执行了语义搜索,并浮现出她的马拉松训练和饮食要求。
  4. Harness injected these synthesized facts directly into the system context.架构将这些合成的事实直接注入到系统上下文中。
  5. Gemini generated a fully personalized response that felt natural, supportive, and continuous.Gemini 生成了一个完全个性化的响应,感觉自然、支持且具有连续性。

You can also check this in the Google Cloud Console. Go Agent Platform in google cloud and go to Agents->Deployments. Click on your agent and click on Memories. You will see saved memories like below.您也可以在 Google Cloud Console 中查看。在 Google Cloud 中进入 Agent Platform,然后转到 Agents->Deployments。点击您的智能体,然后点击 Memories。您将看到如下保存的记忆。

Press enter or click to view image in full size

Conclusion结论

As AI agents evolve from novelty demonstrations to mission-critical enterprise workflows, the systems we build around them must mature accordingly.随着 AI 智能体从新奇的演示演变为关键任务的企业工作流,我们围绕它们构建的系统也必须相应地成熟。

  • Short-Term Session Memory for conversational agility and multi-step reasoning scratchpads.用于对话敏捷性和多步推理草稿纸的短期会话记忆。
  • Long-Term Memory Bank for cross-session continuity, personalization, and evolving knowledge retention.用于跨会话连续性、个性化和不断演进的知识保留的长期 Memory Bank。

Useful Resources有用资源

Originally published at https://bufferof.com.最初发布于 https://bufferof.com。

Bhandari Haren
Bhandari Haren

Written by Bhandari Haren

I am a Senior Developer Advocate, AIML at Google. My interest are Generative AI and AI Agents. Linkedin: https://bit.ly/3J90h3k X: @HarenB16221

Google Cloud - Community
Google Cloud - Community

Published in Google Cloud - Community

A collection of technical articles and blogs published or curated by Google Cloud Developer Advocates. The views expressed are those of the authors and don't necessarily reflect those of Google.