概览 - AgentKit 文档

短期记忆保存会话级上下文，让智能体在多轮交互中记住之前说过的话。它本质上就是发送给模型的对话上下文——系统提示词加历史消息——VeADK 用 session_id 来标识：复用同一个 session_id，智能体就能记住之前的轮次。当用户开启对话时，VeADK 会自动创建一个 Session 对象，全程跟踪并管理该会话的所有内容。

短期记忆基于 Google ADK 的 Session 机制，更多背景见 Google ADK Session。

统一入口：ShortTermMemory

无论使用哪种后端，均通过统一的 veadk.memory.short_term_memory.ShortTermMemory 接入。它根据 backend（或 db_url）选择存储方式，并提供统一的会话管理能力。

from veadk.memory.short_term_memory import ShortTermMemory

# 通过 backend 选择实现
stm = ShortTermMemory(backend="sqlite", local_database_path="./stm.db")

参数

参数	类型	默认值	说明
`backend`	`"local" \| "sqlite" \| "mysql" \| "postgresql" \| "database"`	`"local"`	选择后端实现。`database` 已废弃，等价于 `sqlite`。
`db_url`	`str`	`""`	直接给出数据库连接串（如 `sqlite:///./test.db`）。一旦设置，将忽略 `backend`，并采用 `db_url` 与 `db_kwargs`。
`backend_configs`	`dict`	`{}`	后端专用配置，例如通过 `mysql_config` 或 `postgresql_config` 覆盖数据库连接配置。
`db_kwargs`	`dict`	`{}`	数据库连接或连接池的额外参数。
`local_database_path`	`str`	`/tmp/veadk_local_database.db`	仅 `sqlite` 使用的本地数据库文件路径。
`after_load_memory_callback`	`Callable \| None`	`None`	读取会话后触发的回调，入参为加载到的 `Session`。
`after_create_session_callback`	`Callable \| None`	`None`	新会话创建成功后触发的同步或异步回调，入参为新建的 `Session`；复用已有会话时不触发。

当连接串里的用户名或密码包含 @、: 等特殊字符时，请先用 urllib.parse.quote_plus 编码（例如 p@ssword → p%40ssword），否则解析会出错。

后端行为

所有后端提供相同的会话管理能力，区别在于数据保存位置和连接方式：

local 只在当前进程内保存会话；
sqlite 把会话写入本地文件；
mysql 与 postgresql 把会话写入外部数据库，适合多实例共享。

选择后端

后端	是否持久化	依赖外部服务	适用场景	文档
`local`	否（仅内存）	无	本地调试、临时会话	本地内存
`sqlite`	是（本地文件）	无	单机持久化	SQLite
`mysql`	是	MySQL	分布式持久化	MySQL
`postgresql`	是	PostgreSQL	分布式持久化	PostgreSQL

与 Runner 协作

短期记忆通常传给 Runner，由 Runner 自动创建或恢复会话。运行时复用相同的 session_id 即可延续上下文。

import asyncio

from veadk import Agent, Runner
from veadk.memory.short_term_memory import ShortTermMemory

stm = ShortTermMemory(backend="sqlite", local_database_path="./stm.db")
agent = Agent(name="memory_agent", instruction="记住用户告诉你的信息。")
runner = Runner(agent=agent, short_term_memory=stm, app_name="memory_demo")

async def main():
    sid = "user-42-chat"
    print(await runner.run(messages="我叫小明，最喜欢蓝色。", session_id=sid))
    print(await runner.run(messages="我叫什么？喜欢什么颜色？", session_id=sid))

asyncio.run(main())

若既没给 Runner 传 short_term_memory 也没传 session_service，Runner 会自动创建一个 local（内存）短期记忆兜底。

会话管理接口

你通常无需直接创建或管理 Session，而是通过 session_service 管理整个会话生命周期：

启动新会话 create_session()：用户发起交互时创建新的 Session。
恢复已有会话 get_session()：通过 session_id 检索特定 Session，接续之前的进度。
保存进度 append_event()：把新的交互（Event）追加到会话历史。
列出会话 list_sessions()：查询某用户与应用下的活跃会话。
清理会话 delete_session()：删除 Session 及其关联数据。

会话创建回调用于在会话首次创建后执行资源准备、审计记录或外部系统同步。注册 after_create_session_callback：

import asyncio

async def after_create_session(session):
    await provision_external_resources(session.id)

stm = ShortTermMemory(
    after_create_session_callback=after_create_session,
)

回调只在真正创建新会话时执行；复用已有会话时不触发。回调可以是同步函数或异步函数（async def）。如果回调抛出异常，异常会传递给调用方，智能体不会继续执行。

数据库后端下，ShortTermMemory.create_session() 会先检查并复用同 session_id 的已有会话，避免重复创建。回调仅在首次创建时运行。

上下文压缩

随着会话进行，历史会不断增长，导致模型处理的数据变多、响应变慢。上下文压缩用滑动窗口汇总历史：当会话历史超过设定阈值时，自动压缩较早的事件。

配置上下文压缩

配置后，Runner 会在每次达到间隔时自动压缩会话历史。

from google.adk.apps.app import App, EventsCompactionConfig

from veadk import Agent

root_agent = Agent(
    name="my_agent",
    instruction="你是一个智能助手，擅长用中文礼貌地回复用户问题。",
)

app = App(
    name="my_agent",
    root_agent=root_agent,
    events_compaction_config=EventsCompactionConfig(
        compaction_interval=3,  # 每 3 次新调用触发一次压缩
        overlap_size=1,         # 与上一个窗口的最后一个事件重叠
    ),
)

自定义压缩器

用 LlmEventSummarizer 指定压缩使用的模型和提示词模板。模型的 API Key 与 API Base 通过环境变量提供，这里从 os.environ 读取，不要硬编码：

import os

from google.adk.apps.app import App, EventsCompactionConfig
from google.adk.apps.llm_event_summarizer import LlmEventSummarizer
from google.adk.models.lite_llm import LiteLlm

from veadk import Agent

root_agent = Agent(
    name="my_agent",
    instruction="你是一个智能助手，擅长用中文礼貌地回复用户问题。",
)

summarization_llm = LiteLlm(
    model="volcengine/doubao-seed-2-1-pro-260628",
    api_key=os.environ["MODEL_AGENT_API_KEY"],
    api_base=os.environ.get(
        "MODEL_AGENT_API_BASE", "https://ark.cn-beijing.volces.com/api/v3/"
    ),
)

my_compactor = LlmEventSummarizer(
    llm=summarization_llm,
    prompt_template="""请总结这段对话，压缩要求：
1. 保留关键实体、数据点和时间线；
2. 突出讨论过的核心问题与解决方案；
3. 保持逻辑连贯与上下文相关；
4. 去除重复表达和冗余细节。""",
)

app = App(
    name="my_agent",
    root_agent=root_agent,
    events_compaction_config=EventsCompactionConfig(
        compactor=my_compactor,
        compaction_interval=5,
        overlap_size=1,
    ),
)

概览

快速开始

核心组件

可观测

扩展

前端

安全

部署

概览

统一入口：ShortTermMemory

参数

后端行为

选择后端

与 Runner 协作

会话管理接口

上下文压缩

配置上下文压缩

自定义压缩器

​统一入口：ShortTermMemory

​参数

​后端行为

​选择后端

​与 Runner 协作

​会话管理接口

​上下文压缩

​配置上下文压缩

​自定义压缩器

统一入口：ShortTermMemory

参数

后端行为

选择后端

与 Runner 协作

会话管理接口

上下文压缩

配置上下文压缩

自定义压缩器