GUMem Python SDK

GUMem API Key 只能放在服务端。不要把 API Key 暴露给浏览器、移动端应用、客户端脚本或公开仓库。

Before you start

你需要：

Python 3.9 或更高版本。
一个可用的 GUMem API Key。
一个 Python 服务端运行环境，例如 FastAPI、Django、Flask、后台 worker、队列任务或 Agent 服务。

如果只是本地验证，可以先设置环境变量：

bash

export GUMEM_API_KEY="<your-gumem-api-key>"

Install

bash

pip install steamory-agent-kit-gumem

SDK 使用 httpx 发起请求，并提供同步客户端 GUMemClient 与异步客户端 AsyncGUMemClient。请求类型使用 TypedDict 描述，因此你既可以获得类型提示，也可以直接传入普通 Python dict。

Quickstart

下面的示例完成最短接入路径：初始化同步客户端、创建 Session、写入几条对话消息，然后召回相关 Memory。

python

import os

from gumem import GUMemClient

gumem = GUMemClient(api_key=os.environ["GUMEM_API_KEY"])

session = gumem.sessions.create({
    "user_id": "user_123",
    "title": "Team scheduling session",
})

session.add_messages([
    {
        "role": "user",
        "content": (
            "For team scheduling, use Berlin when I mention Europe "
            "and Toronto when I mention the Americas."
        ),
    },
    {
        "role": "assistant",
        "content": (
            "Got it. I will use Berlin for Europe scheduling "
            "and Toronto for Americas scheduling."
        ),
    },
])

memory = session.get_memory({
    "query": "which city should be used for Europe team scheduling",
    "details": True,
})

print(memory.get("data"))

Checkpoint

这段代码运行后，你应该能拿到一个 Session 对象，并从 memory.get("data") 中读取 GUMem 返回的 Memory 结果。session.id 是后续继续写入消息和召回 Memory 的稳定标识。

生产环境中建议在应用关闭时调用 gumem.close()，或者使用 context manager 自动关闭 SDK 持有的 httpx.Client。

python

import os

from gumem import GUMemClient

with GUMemClient(api_key=os.environ["GUMEM_API_KEY"]) as gumem:
    health = gumem.health()
    print(health)

Async usage

异步服务可以使用 AsyncGUMemClient。它暴露的资源与同步客户端一致，但网络方法需要 await。

python

import os

from gumem import AsyncGUMemClient


async def main() -> None:
    async with AsyncGUMemClient(api_key=os.environ["GUMEM_API_KEY"]) as gumem:
        session = await gumem.sessions.create({
            "user_id": "user_123",
            "title": "Async support session",
        })

        await session.add_message({
            "role": "user",
            "content": "For Europe planning, keep Berlin as the default city.",
        })

        memory = await session.get_memory({
            "query": "default city for Europe planning",
            "details": True,
        })

        print(memory.get("data"))

同步和异步 API 的主要对应关系：

Sync	Async	Description
`GUMemClient`	`AsyncGUMemClient`	客户端入口。
`Session`	`AsyncSession`	本地 Session handle。
`gumem.health()`	`await gumem.health()`	检查服务健康状态。
`gumem.sessions.create()`	`await gumem.sessions.create()`	创建 Session。
`gumem.sessions.from_id()`	`gumem.sessions.from_id()`	从已有 Session ID 恢复本地 handle，不发起网络请求。
`session.add_message()`	`await session.add_message()`	写入单条消息。
`session.add_messages()`	`await session.add_messages()`	写入多条消息。
`session.get_memory()`	`await session.get_memory()`	召回 Memory。
`gumem.user_actions.create()`	`await gumem.user_actions.create()`	记录用户 Action。

Use GUMem in an assistant turn

在真实 Agent 或助手系统中，推荐的顺序是：

用户输入到达 Python 服务端。
使用当前 session.id 召回相关 Memory。
将 Memory 放入模型上下文。
得到模型回复后，把用户消息和助手回复写回 GUMem。

python

import os
from typing import Any, Dict

from gumem import GUMemClient

gumem = GUMemClient(api_key=os.environ["GUMEM_API_KEY"])


def generate_assistant_reply(input: Dict[str, Any]) -> str:
    # Replace this function with your OpenAI, Anthropic, Gemini, or local model call.
    return f"I will use the recalled memory for: {input['user_content']}"


def assistant_turn(session_id: str, user_content: str) -> Dict[str, Any]:
    session = gumem.sessions.from_id(session_id)

    memory = session.get_memory({
        "query": user_content,
        "details": True,
    })

    assistant_reply = generate_assistant_reply({
        "user_content": user_content,
        "recalled_memory": memory.get("data"),
    })

    session.add_messages([
        {"role": "user", "content": user_content},
        {"role": "assistant", "content": assistant_reply},
    ])

    return {
        "reply": assistant_reply,
        "memory": memory.get("data"),
    }

Checkpoint

生产环境中通常在创建新对话时保存 session.id。后续请求只需要用 gumem.sessions.from_id(session_id) 恢复本地 Session handle；这个操作不会发起网络请求。

Client configuration

默认情况下只需要传入 api_key。如果你使用 GUMem 默认服务，不需要配置 host。

python

import os

from gumem import GUMemClient

gumem = GUMemClient(api_key=os.environ["GUMEM_API_KEY"])

需要连接自定义部署、调整超时、注入默认 headers 或复用自定义 httpx.Client 时，再传入其他选项：

python

import httpx

from gumem import GUMemClient

client = httpx.Client(transport=httpx.HTTPTransport(retries=2))

gumem = GUMemClient(
    api_key="gumem_api_key",
    host="gumem.asix.inc",
    timeout_ms=30_000,
    headers={"X-Service": "assistant-api"},
    client=client,
)

Option	Type	Default	Description
`api_key`	`str`	必填	GUMem API Key。SDK 会发送 `Authorization: Api-Key <api_key>`。如果传入值已经带 `Api-Key` 前缀，SDK 不会重复添加。
`host`	`str \| None`	`gumem.asix.inc`	GUMem 服务 host。普通 host 会补全为 HTTPS，显式 `http://` 或 `https://` 会保留，末尾 `/` 会被移除。
`timeout_ms`	`int \| None`	`30000`	默认请求超时时间，单位毫秒。设置为 `0` 可以关闭 SDK timeout。
`headers`	`Mapping[str, str] \| None`	`None`	默认 headers。SDK 会把它们合并进每次请求。
`client`	`httpx.Client \| None`	`None`	同步 HTTP client，可用于代理、测试、自定义 transport 或观测封装。异步客户端对应 `httpx.AsyncClient`。

单次请求可以通过 options 覆盖 timeout 和 headers：

python

memory = session.get_memory(
    {"query": "Europe planning"},
    options={
        "timeout_ms": 10_000,
        "headers": {"X-Trace-Id": "trace_123"},
    },
)

Name	Type	Required	Description
`options.timeout_ms`	`int \| None`	否	覆盖本次请求的 timeout。
`options.headers`	`Mapping[str, str] \| None`	否	追加或覆盖本次请求 headers。

Health check

gumem.health(options=None) 检查 GUMem 服务健康状态，适合在启动检查、部署验证或监控探针中使用。

python

health = gumem.health()

print(health)

异步客户端中使用：

python

health = await gumem.health()

参数：

Name	Type	Required	Description
`options`	`RequestOptions`	否	单次请求选项。可以覆盖 timeout 和 headers。

Sessions

Session 是 GUMem 中承载一段用户对话和 Memory 召回上下文的对象。优先使用 Session handle 上的方法；只有在你更适合传递原始 Session ID 时，再使用 lower-level resource 方法。

Create a Session

gumem.sessions.create(input, options=None) 创建 Session，并返回 Session 对象。user_id 是当前 GUMem API 的必填字段。

python

session = gumem.sessions.create({
    "user_id": "user_123",
    "title": "Support chat",
    "metadata": {
        "source": "web-chat",
        "locale": "en-US",
    },
})

print(session.id)
print(session.raw_response)

参数：

Name	Type	Required	Description
`input`	`SessionCreateRequest`	是	创建 Session 的请求体。
`input.user_id`	`str`	是	业务系统中的用户 ID。GUMem 使用它把 Session 和用户 Memory 关联起来。
`input.title`	`str \| None`	否	Session 标题，适合保存会话主题或来源说明。
`input.metadata`	`dict[str, Any] \| None`	否	自定义元数据，例如渠道、语言、产品模块或业务 trace 信息。
`options`	`RequestOptions`	否	单次请求选项。

session.raw_response 保留创建 Session 时 GUMem 返回的原始 response envelope。如果 GUMem 成功响应但没有返回 data.session_id，SDK 会抛出 ValueError，消息为 GUMem API did not return data.session_id。

Restore a Session handle

gumem.sessions.from_id(session_id) 从已有 Session ID 创建本地 Session handle，不会发起网络请求。

python

session = gumem.sessions.from_id("session_123")

session.add_message({
    "role": "user",
    "content": "Continue with the city preference from earlier.",
})

参数：

Name	Type	Required	Description
`session_id`	`str`	是	已经由 GUMem 创建并保存在业务系统中的 Session ID。

Add messages

session.add_message(message, options=None) 写入单条消息。

python

session.add_message({
    "role": "user",
    "content": "For Europe planning, keep Berlin as the default city.",
    "metadata": {
        "channel": "chat",
    },
})

参数：

Name	Type	Required	Description
`message`	`Message`	是	要写入当前 Session 的单条消息。
`message.role`	`str`	是	消息角色，例如 `user`、`assistant`、`system` 或你的业务自定义角色。
`message.content`	`str`	是	消息正文。GUMem 会基于它沉淀和召回 Memory。
`message.id`	`str \| None`	否	业务侧消息 ID。用于和你的消息表、日志或 trace 关联。
`message.metadata`	`dict[str, Any]`	否	消息级元数据，例如渠道、模型名、语言或业务标签。
`message.timestamp`	`str \| datetime \| None`	否	消息发生时间。传入 `datetime` 时 SDK 会序列化为 ISO string。
`message.created_at`	`str \| datetime \| None`	否	消息创建时间。适合和服务端写入时间区分。
`message.status`	`pending \| chunked \| processed \| failed`	否	消息处理状态。
`options`	`RequestOptions`	否	单次请求选项。

session.add_messages(input, options=None) 写入多条消息。input 可以是消息列表，也可以是包含 messages 字段的对象。

python

session.add_messages([
    {"role": "user", "content": "I prefer Berlin for Europe meetings."},
    {"role": "assistant", "content": "I will remember Berlin for Europe meetings."},
])

session.add_messages({
    "messages": [
        {"role": "user", "content": "Toronto works best for Americas meetings."},
    ],
    "user_id": "user_123",
})

参数：

Name	Type	Required	Description
`input`	`list[Message] \| AddMessagesRequest`	是	要写入的消息列表，或包含 `messages` 字段的请求体。
`input.messages`	`list[Message]`	是	批量写入的消息数组。
`input.user_id`	`str \| None`	否	可选用户 ID。适合后端接口需要显式传递用户归属时使用。
`options`	`RequestOptions`	否	单次请求选项。

如果你更适合传递原始 Session ID，可以使用 lower-level 方法：

python

gumem.sessions.add_messages("session_123", {
    "messages": [
        {"role": "user", "content": "hello"},
    ],
})

Get Memory

session.get_memory(params=None, options=None) 召回当前 Session 的相关 Memory。

python

memory = session.get_memory({
    "query": "which city should be used for Europe scheduling",
    "details": True,
})

print(memory.get("data"))

参数：

Name	Type	Required	Description
`params`	`GetSessionMemoryParams`	否	Memory 召回参数。
`params.query`	`str`	否	当前用户问题或业务查询。SDK 会把它作为 query string 发送。
`params.details`	`bool`	否	是否请求更详细的召回结果。SDK 会把它作为 query string 发送。
`params.recall_config`	`RecallConfig \| None`	否	召回配置。只要请求体中出现 `recall_config` 字段，SDK 就会使用 POST context endpoint。
`options`	`RequestOptions`	否	单次请求选项。

传入 recall_config 可以调整近期 Message 数量，并按 metadata 精确过滤召回范围：

python

memory = session.get_memory({
    "query": "which city should be used for the user request",
    "details": True,
    "recall_config": {
        "MessageRecentLimit": 20,
        "MetadataFilters": {
            "ping": "pong",
        },
    },
})

常用 RecallConfig 字段：

字段名以当前 SDK/API 为准；说明中使用 GUMem 的公开术语。

Name	Type	Description
`MessageRecentLimit`	`int`	近期 Message 召回数量。
`MetadataFilters`	`dict`	简单的 metadata KV 字典，用于精确过滤召回范围。key 必须是非空字符串，value 必须是字符串、数字或布尔值。

lower-level 方式：

python

memory = gumem.sessions.get_memory("session_123", {
    "query": "Europe scheduling",
    "details": True,
})

用户 Action 适合记录点击、页面访问、业务操作、偏好修改等非对话事件，让 GUMem 能从用户行为中沉淀 Memory。

python

from datetime import datetime, timezone

gumem.user_actions.create({
    "user_id": "user_123",
    "timestamp": datetime(2026, 4, 22, 1, 2, 3, tzinfo=timezone.utc),
    "content": "User opened the Europe team scheduling page",
    "session_id": "session_123",
    "event_type": "page_view",
    "page": "team_scheduling",
    "anchors": {"region": "Europe", "city": "Berlin"},
    "metadata": {"source": "assistant-api"},
})

参数：

Name	Type	Required	Description
`input`	`ActionLogInput`	是	用户 Action 请求体。
`input.user_id`	`str`	是	业务系统中的用户 ID。
`input.timestamp`	`str \| datetime`	是	事件发生时间。传入 `datetime` 时 SDK 会序列化为 ISO string。
`input.content`	`str`	是	事件内容描述。应写成对 Memory 有意义的自然语言。
`input.log_id`	`str \| None`	否	业务侧事件 ID。
`input.session_id`	`str \| None`	否	关联的 GUMem Session ID。
`input.device_id`	`str \| None`	否	设备 ID。
`input.app`	`str \| None`	否	应用名或产品名。
`input.platform`	`str \| None`	否	平台，例如 `web`、`ios`、`android` 或 `backend`。
`input.event_type`	`str \| None`	否	事件类型，例如 `page_view`、`click`、`purchase`、`preference_update`。
`input.page`	`str \| None`	否	页面或业务模块。
`input.anchors`	`dict[str, str] \| None`	否	结构化锚点，例如地区、城市、文档 ID 或商品 ID。
`input.metadata`	`dict[str, Any] \| None`	否	自定义元数据。
`input.entities`	`list[str] \| None`	否	与事件相关的实体列表。
`options`	`RequestOptions`	否	单次请求选项。

异步客户端中使用：

python

await gumem.user_actions.create({
    "user_id": "user_123",
    "timestamp": "2026-04-22T01:02:03Z",
    "content": "User opened the Europe team scheduling page",
})

Error handling

SDK 把 GUMem API 错误、网络错误和超时错误区分为不同异常。

python

from gumem import GUMemApiError, GUMemConnectionError, GUMemTimeoutError

try:
    gumem.sessions.create({"user_id": "user_123"})
except GUMemApiError as error:
    print(error.status_code, error.status_text, error.detail, error.body)
except GUMemTimeoutError as error:
    print(f"Timed out after {error.timeout_ms}ms")
except GUMemConnectionError as error:
    print("Network failure", error.cause)

Error	When it is raised	Useful fields
`GUMemApiError`	GUMem API 返回非 2xx 响应。	`status_code`、`status`、`status_text`、`headers`、`body`、`detail`。
`GUMemTimeoutError`	请求在配置的 timeout 内未完成。	`timeout_ms`、`cause`。
`GUMemConnectionError`	请求在 GUMem 返回响应前失败，例如 DNS、连接或 transport 错误。	`cause`。
`GUMemError`	SDK 错误基类。	可用于统一捕获 SDK 异常。

Public types / exported surface

Python SDK 从 gumem 包根导出常用客户端、异常和类型。业务代码通常只需要导入客户端与少量类型。

python

from gumem import (
    ActionLogInput,
    AsyncGUMemClient,
    GetSessionMemoryParams,
    GUMemApiError,
    GUMemClient,
    Message,
    RecallConfig,
    SessionCreateRequest,
)

Export	Description
`GUMemClient`	同步客户端入口。
`AsyncGUMemClient`	异步客户端入口。
`Session`	同步 Session handle，由 `gumem.sessions.create()` 或 `gumem.sessions.from_id()` 返回。
`AsyncSession`	异步 Session handle，由 `AsyncGUMemClient` 的 sessions resource 返回。
`SessionCreateRequest`	创建 Session 的请求类型。
`Message`	写入 Session 的消息类型。
`AddMessagesRequest`	批量写入消息的请求类型。
`GetSessionMemoryParams`	召回 Memory 的参数类型。
`RecallConfig`	Memory 召回配置类型。
`ActionLogInput`	写入用户 Action 的请求类型。
`GUMemEnvelope`	GUMem API response envelope，类型为 `dict[str, Any]`。
`RequestOptions`	单次请求选项类型。
`GUMemApiError`、`GUMemConnectionError`、`GUMemTimeoutError`、`GUMemError`	SDK 异常类型。

Python SDK v1 的公开业务表面刻意保持收敛：不暴露 threads、get_context 或 user_actions.query。如果你需要继续操作已有 Session，请使用 gumem.sessions.from_id(session_id)。

GUMem Python SDK ​

Before you start ​

Install ​

Quickstart ​

Checkpoint ​

Async usage ​

Use GUMem in an assistant turn ​

Checkpoint ​

Client configuration ​

Health check ​

Sessions ​

Create a Session ​

Restore a Session handle ​

Add messages ​

Get Memory ​

User actions ​

Error handling ​

Public types / exported surface ​

GUMem Python SDK

Before you start

Install

Quickstart

Checkpoint

Async usage

Use GUMem in an assistant turn

Checkpoint

Client configuration

Health check

Sessions

Create a Session

Restore a Session handle

Add messages

Get Memory

User actions

Error handling

Public types / exported surface