jianmu.memory

适用对象：Agent 应用开发者 / 记忆与上下文扩展开发者 是否必读：按需 相关模块：jianmu.message, jianmu.skill, jianmu.rag

1. 模块职责

jianmu.memory 负责工作记忆、长期记忆、checkpoint、上下文构建和记忆工具。

它的核心价值不是单纯“存数据”，而是把消息、状态、历史和工具描述整理成可供模型消费的上下文。

2. 适合查什么

• 协议：WorkingMemoryProtocol、LongTermMemoryProtocol、CheckpointerProtocol
• 上下文构建：ContextBuilder、ContextBuilderProtocol
• token 预算：TokenCounterProtocol、TokenBudgetFilter
• checkpoint：InMemoryCheckpointer、FileCheckpointer、NullCheckpointer
• 内置记忆工具：MemoryAddTool、MemorySearchTool

3. 使用建议

• 想先把聊天历史、状态和工具说明拼成 prompt 时，优先看上下文构建相关对象
• 想给 runner 增加恢复能力时，先接 FileCheckpointer 或 InMemoryCheckpointer
• 想给 agent 提供轻量长期记忆工具时，再使用 MemoryAddTool / MemorySearchTool
• 想接入能力型长期记忆后端时，优先实现或使用 LongTermMemoryProtocol
• 需要更准确的 token 预算控制时，不要只停留在 SimpleTokenCounter，优先考虑 TiktokenTokenCounter 或 HFTokenCounter

4. 注意事项

• LongTermMemoryProtocol 是长期记忆的公开主抽象
• SimpleStore 是一个轻量内置长期记忆后端，而公开抽象边界仍然是 LongTermMemoryProtocol
• 这里的长期记忆层与 RAG 向量库不是同一层抽象
• 需要检索增强时，再结合 jianmu.rag
• ContextBuilder 属于 prompt 上下文构建层，不等同于长期记忆或知识库

5. 最小示例

from jianmu.memory import FileCheckpointer, Mem0LongTermMemory, SimpleStore, create_memory_tools

store = SimpleStore()
checkpointer = FileCheckpointer()
memory_tools = create_memory_tools(long_term_memory=store)
long_term_memory = Mem0LongTermMemory()

6. 常见入口

• 想持久化 runner 状态：看 FileCheckpointer / InMemoryCheckpointer
• 想组 prompt 上下文：看 ContextBuilder
• 想限制上下文 token 预算：看 TokenCounterProtocol、SimpleTokenCounter、TiktokenTokenCounter、HFTokenCounter、TokenBudgetFilter
• 想给 agent 一个轻量 memory read/write 能力：看 MemoryAddTool、MemorySearchTool
• 想接入 mem0 这类长期记忆系统：看 LongTermMemoryProtocol、Mem0LongTermMemory

7. API 参考

memory

jianmu memory module – context, checkpoint, and long-term memory primitives.

WorkingMemoryProtocol

Bases: Protocol

Abstract interface for filtering or summarizing state messages.

build_view

build_view(messages: Sequence[Message]) -> List[Message]

Build the working-memory view from raw history messages.

参数：

名称	类型	描述	默认
messages	Sequence[Message]	Source message history before truncation or summarization.	必需

返回：

类型	描述
List[Message]	Messages that should remain visible in the model context window.

源代码位于： jianmu/memory/base.py

def build_view(self, messages: Sequence[Message]) -> List[Message]:
    """Build the working-memory view from raw history messages.

    Args:
        messages: Source message history before truncation or summarization.

    Returns:
        Messages that should remain visible in the model context window.
    """
    ...

CheckpointerProtocol

Bases: Protocol

Abstract interface for thread state checkpointing.

save_checkpoint

save_checkpoint(
    thread_id: str,
    step: int,
    state_dump: dict,
    tree_state: Optional[dict] = None,
    runtime_metadata: Optional[dict] = None,
) -> None

Persist a checkpoint for a logical thread.

参数：

名称	类型	描述	默认
thread_id	str	Stable identifier for the checkpoint stream.	必需
step	int	Monotonic checkpoint step.	必需
state_dump	dict	Serializable runtime state payload.	必需
tree_state	Optional[dict]	Optional serialized tree-status payload.	None
runtime_metadata	Optional[dict]	Optional runner-internal metadata payload.	None

源代码位于： jianmu/memory/base.py

def save_checkpoint(
    self,
    thread_id: str,
    step: int,
    state_dump: dict,
    tree_state: Optional[dict] = None,
    runtime_metadata: Optional[dict] = None,
) -> None:
    """Persist a checkpoint for a logical thread.

    Args:
        thread_id: Stable identifier for the checkpoint stream.
        step: Monotonic checkpoint step.
        state_dump: Serializable runtime state payload.
        tree_state: Optional serialized tree-status payload.
        runtime_metadata: Optional runner-internal metadata payload.
    """
    ...

get_checkpoint

get_checkpoint(thread_id: str) -> Optional[dict]

Load the latest checkpoint for a logical thread.

参数：

名称	类型	描述	默认
thread_id	str	Stable identifier for the checkpoint stream.	必需

返回：

类型	描述
Optional[dict]	Latest checkpoint payload, or None if no checkpoint exists.

源代码位于： jianmu/memory/base.py

def get_checkpoint(self, thread_id: str) -> Optional[dict]:
    """Load the latest checkpoint for a logical thread.

    Args:
        thread_id: Stable identifier for the checkpoint stream.

    Returns:
        Latest checkpoint payload, or ``None`` if no checkpoint exists.
    """
    ...

ContextBuilderProtocol

Bases: Protocol

Build a context (messages/prompt) from state and runtime context.

build

build(
    local_state: Any,
    global_state: Any = None,
    ctx: Any | None = None,
    **kwargs: Any,
) -> Sequence[Message]

Build a message context from local state and runtime context.

参数：

名称	类型	描述	默认
local_state	Any	State owned by the current agent or node.	必需
global_state	Any	Optional shared state visible across agents or trees.	None
ctx	Any | None	Optional runtime context object.	None
**kwargs	Any	Builder-specific extension arguments.	{}

返回：

类型	描述
Sequence[Message]	Ordered messages ready to be sent to the model.

源代码位于： jianmu/memory/context/base.py

def build(
    self,
    local_state: Any,
    global_state: Any = None,
    ctx: Any | None = None,
    **kwargs: Any,
) -> Sequence[Message]:
    """Build a message context from local state and runtime context.

    Args:
        local_state: State owned by the current agent or node.
        global_state: Optional shared state visible across agents or trees.
        ctx: Optional runtime context object.
        **kwargs: Builder-specific extension arguments.

    Returns:
        Ordered messages ready to be sent to the model.
    """
    ...

transform

transform(
    messages: Sequence[Message],
    *,
    local_state: Any = None,
    global_state: Any = None,
    ctx: Any | None = None,
    **kwargs: Any,
) -> Sequence[Message]

Transform an existing message list.

参数：

名称	类型	描述	默认
messages	Sequence[Message]	Existing message list to post-process.	必需
local_state	Any	State owned by the current agent or node.	None
global_state	Any	Optional shared state visible across agents or trees.	None
ctx	Any | None	Optional runtime context object.	None
**kwargs	Any	Builder-specific extension arguments.	{}

返回：

类型	描述
Sequence[Message]	Transformed message sequence.

源代码位于： jianmu/memory/context/base.py

def transform(
    self,
    messages: Sequence[Message],
    *,
    local_state: Any = None,
    global_state: Any = None,
    ctx: Any | None = None,
    **kwargs: Any,
) -> Sequence[Message]:
    """Transform an existing message list.

    Args:
        messages: Existing message list to post-process.
        local_state: State owned by the current agent or node.
        global_state: Optional shared state visible across agents or trees.
        ctx: Optional runtime context object.
        **kwargs: Builder-specific extension arguments.

    Returns:
        Transformed message sequence.
    """
    ...

MessageProvider

Bases: Protocol

Produce a list of messages from state and runtime context.

get_messages

get_messages(
    local_state: Any,
    global_state: Any = None,
    ctx: Any | None = None,
) -> Sequence[Message]

Produce messages from the provided state and runtime context.

参数：

名称	类型	描述	默认
local_state	Any	State owned by the current agent or node.	必需
global_state	Any	Optional shared state visible across agents or trees.	None
ctx	Any | None	Optional runtime context object.	None

返回：

类型	描述
Sequence[Message]	Message sequence contributed by this provider.

源代码位于： jianmu/memory/context/base.py

def get_messages(self, local_state: Any, global_state: Any = None, ctx: Any | None = None) -> Sequence[Message]:
    """Produce messages from the provided state and runtime context.

    Args:
        local_state: State owned by the current agent or node.
        global_state: Optional shared state visible across agents or trees.
        ctx: Optional runtime context object.

    Returns:
        Message sequence contributed by this provider.
    """
    ...

ContextFilter

Bases: Protocol

Post-process aggregated messages.

apply

apply(
    messages: Sequence[Message],
    local_state: Any = None,
    global_state: Any = None,
    ctx: Any | None = None,
    **kwargs: Any,
) -> Sequence[Message]

Post-process an aggregated message list.

参数：

名称	类型	描述	默认
messages	Sequence[Message]	Message list produced by providers or prior filters.	必需
local_state	Any	State owned by the current agent or node.	None
global_state	Any	Optional shared state visible across agents or trees.	None
ctx	Any | None	Optional runtime context object.	None
**kwargs	Any	Filter-specific extension arguments.	{}

返回：

类型	描述
Sequence[Message]	Filtered or augmented message sequence.

源代码位于： jianmu/memory/context/base.py

def apply(
    self,
    messages: Sequence[Message],
    local_state: Any = None,
    global_state: Any = None,
    ctx: Any | None = None,
    **kwargs: Any,
) -> Sequence[Message]:
    """Post-process an aggregated message list.

    Args:
        messages: Message list produced by providers or prior filters.
        local_state: State owned by the current agent or node.
        global_state: Optional shared state visible across agents or trees.
        ctx: Optional runtime context object.
        **kwargs: Filter-specific extension arguments.

    Returns:
        Filtered or augmented message sequence.
    """
    ...

ContextBuilder

ContextBuilder(
    providers: Sequence[MessageProvider],
    filters: Sequence[ContextFilter] | None = None,
)

Bases: ContextBuilderProtocol

Composable context builder: providers first, then filters.

Use this when prompt assembly needs to combine multiple sources such as chat history, tool descriptions, state summaries, and post-processing filters like token budgets.

属性：

名称	类型	描述
providers		Ordered providers contributing base messages.
filters		Ordered filters applied after message aggregation.

Create a context builder.

参数：

名称	类型	描述	默认
providers	Sequence[MessageProvider]	Ordered providers that contribute base messages.	必需
filters	Sequence[ContextFilter] | None	Optional ordered filters that post-process aggregated messages.	None

源代码位于： jianmu/memory/context/builder.py

def __init__(self, providers: Sequence[MessageProvider], filters: Sequence[ContextFilter] | None = None):
    """Create a context builder.

    Args:
        providers: Ordered providers that contribute base messages.
        filters: Optional ordered filters that post-process aggregated messages.
    """
    self.providers = list(providers)
    self.filters = list(filters or [])

for_chat classmethod

for_chat(
    system_prompt: str | None = None,
    working_memory: Any | None = None,
    max_messages: int | None = None,
    max_tokens: int | None = None,
    *,
    runtime_prompt: "PromptRuntimeContext | None" = None,
) -> "ContextBuilder"

Build a chat-oriented context builder.

This preset assembles persona prompt, optional bootstrap-file context, and conversation history. It does not inject tool or skill guidance.

参数：

名称	类型	描述	默认
system_prompt	str | None	Optional persona prompt. When omitted, the configured default persona prompt is used.	None
working_memory	Any | None	Optional working-memory adapter applied to history.	None
max_messages	int | None	Optional maximum number of messages retained after filtering.	None
max_tokens	int | None	Optional token budget applied after aggregation.	None
runtime_prompt	'PromptRuntimeContext | None'	Optional execution-scoped prompt overrides. for_chat only consumes generic fields such as bootstrap file settings and ignores skill-specific fields.	None

返回：

类型	描述
'ContextBuilder'	A configured ContextBuilder instance for plain chat prompting.

源代码位于： jianmu/memory/context/builder.py

@classmethod
def for_chat(
    cls,
    system_prompt: str | None = None,
    working_memory: Any | None = None,
    max_messages: int | None = None,
    max_tokens: int | None = None,
    *,
    runtime_prompt: "PromptRuntimeContext | None" = None,
) -> "ContextBuilder":
    """Build a chat-oriented context builder.

    This preset assembles persona prompt, optional bootstrap-file context,
    and conversation history. It does not inject tool or skill guidance.

    Args:
        system_prompt: Optional persona prompt. When omitted, the
            configured default persona prompt is used.
        working_memory: Optional working-memory adapter applied to history.
        max_messages: Optional maximum number of messages retained after
            filtering.
        max_tokens: Optional token budget applied after aggregation.
        runtime_prompt: Optional execution-scoped prompt overrides.
            ``for_chat`` only consumes generic fields such as bootstrap
            file settings and ignores skill-specific fields.

    Returns:
        A configured ``ContextBuilder`` instance for plain chat prompting.
    """
    from jianmu.memory.context.providers import (
        BootstrapFilesProvider,
        StaticPromptProvider,
        StateHistoryProvider,
    )

    _, _, _, include_bootstrap_files, workspace_dir = _resolve_runtime_prompt_settings(runtime_prompt=runtime_prompt)

    providers: list[MessageProvider] = []
    persona_prompt = system_prompt if system_prompt is not None else get_persona_default_prompt()
    if persona_prompt:
        providers.append(StaticPromptProvider(persona_prompt))
    if include_bootstrap_files:
        providers.append(BootstrapFilesProvider(workspace_dir=workspace_dir))
    providers.append(StateHistoryProvider(working_memory=working_memory))
    return cls(
        providers=providers,
        filters=_build_default_filters(
            max_messages=max_messages,
            max_tokens=max_tokens,
        ),
    )

for_react classmethod

for_react(
    system_prompt: str | None = None,
    tools_desc: str | None = None,
    working_memory: Any | None = None,
    max_messages: int | None = None,
    max_tokens: int | None = None,
    *,
    runtime_prompt: "PromptRuntimeContext | None" = None,
    bootstrap_files: Sequence[str] | None = None,
) -> "ContextBuilder"

Build a ReAct-oriented context builder.

This preset assembles persona prompt, ReAct protocol, tool descriptions, optional bootstrap-file context, and message history. It does not inject skill summaries or active skill instructions.

参数：

名称	类型	描述	默认
system_prompt	str | None	Optional persona prompt. When omitted, the configured default persona prompt is used.	None
tools_desc	str | None	Optional preformatted tool description block.	None
working_memory	Any | None	Optional working-memory adapter applied to history.	None
max_messages	int | None	Optional maximum number of messages retained after filtering.	None
max_tokens	int | None	Optional token budget applied after aggregation.	None
runtime_prompt	'PromptRuntimeContext | None'	Optional execution-scoped prompt overrides. for_react consumes generic runtime prompt fields such as bootstrap-file settings but intentionally ignores skill-specific fields.	None
bootstrap_files	Sequence[str] | None	Optional explicit bootstrap filenames to load when bootstrap prompting is enabled.	None

返回：

类型	描述
'ContextBuilder'	A configured ContextBuilder instance for ReAct prompting.

源代码位于： jianmu/memory/context/builder.py

@classmethod
def for_react(
    cls,
    system_prompt: str | None = None,
    tools_desc: str | None = None,
    working_memory: Any | None = None,
    max_messages: int | None = None,
    max_tokens: int | None = None,
    *,
    runtime_prompt: "PromptRuntimeContext | None" = None,
    bootstrap_files: Sequence[str] | None = None,
) -> "ContextBuilder":
    """Build a ReAct-oriented context builder.

    This preset assembles persona prompt, ReAct protocol, tool
    descriptions, optional bootstrap-file context, and message history. It
    does not inject skill summaries or active skill instructions.

    Args:
        system_prompt: Optional persona prompt. When omitted, the
            configured default persona prompt is used.
        tools_desc: Optional preformatted tool description block.
        working_memory: Optional working-memory adapter applied to history.
        max_messages: Optional maximum number of messages retained after
            filtering.
        max_tokens: Optional token budget applied after aggregation.
        runtime_prompt: Optional execution-scoped prompt overrides.
            ``for_react`` consumes generic runtime prompt fields such as
            bootstrap-file settings but intentionally ignores skill-specific
            fields.
        bootstrap_files: Optional explicit bootstrap filenames to load when
            bootstrap prompting is enabled.

    Returns:
        A configured ``ContextBuilder`` instance for ReAct prompting.
    """
    from jianmu.memory.context.providers import (
        BootstrapFilesProvider,
        StaticPromptProvider,
        StateHistoryProvider,
        ToolsDescProvider,
    )

    _, _, _, include_bootstrap_files, workspace_dir = _resolve_runtime_prompt_settings(runtime_prompt=runtime_prompt)
    resolved_bootstrap_files = tuple(bootstrap_files) if bootstrap_files is not None else tuple(get_bootstrap_files())

    providers: list[MessageProvider] = []
    persona_prompt = system_prompt if system_prompt is not None else get_persona_default_prompt()
    if persona_prompt:
        providers.append(StaticPromptProvider(persona_prompt))
    react_protocol = get_react_protocol_prompt()
    if react_protocol:
        providers.append(StaticPromptProvider(react_protocol))
    if tools_desc:
        providers.append(ToolsDescProvider(tools_desc))
    if include_bootstrap_files:
        providers.append(
            BootstrapFilesProvider(
                workspace_dir=workspace_dir,
                files=resolved_bootstrap_files,
            )
        )
    providers.append(StateHistoryProvider(working_memory=working_memory))
    return cls(
        providers=providers,
        filters=_build_default_filters(
            max_messages=max_messages,
            max_tokens=max_tokens,
        ),
    )

for_skill classmethod

for_skill(
    *,
    system_prompt: str | None = None,
    skill_set: Any | None = None,
    tools: Sequence[Any] | None = None,
    mode: str = "summary",
    include_resources: bool = True,
    include_react_protocol: bool = False,
    max_messages: int | None = None,
    max_tokens: int | None = None,
    runtime_prompt: "PromptRuntimeContext | None" = None,
    bootstrap_files: Sequence[str] | None = None,
    include_unavailable_skills: bool = False,
) -> "ContextBuilder"

Build a skill-aware context builder.

This preset can operate in two modes:

1. Explicit mode, where skill_set already contains the selected skills for the run.
2. Runtime-resolved mode, where runtime_prompt supplies a skill catalog or skill directory and the builder derives active skills and/or summaries from it.

参数：

名称	类型	描述	默认
system_prompt	str | None	Optional persona prompt. When omitted, the configured default persona prompt is used.	None
skill_set	Any | None	Optional explicitly selected skill set for the run.	None
tools	Sequence[Any] | None	Optional tools available alongside the selected skills.	None
mode	str	Prompt rendering mode for explicit skills, typically "summary" or "full".	'summary'
include_resources	bool	Whether to include runtime-visible skill resource sections for explicit skills.	True
include_react_protocol	bool	Whether to prepend the ReAct protocol contract used by ReAct-backed skill execution.	False
max_messages	int | None	Optional maximum number of messages retained after filtering.	None
max_tokens	int | None	Optional token budget applied after aggregation.	None
runtime_prompt	'PromptRuntimeContext | None'	Optional execution-scoped prompt overrides. for_skill consumes the full skill-related portion of the runtime prompt, including skill catalogs, skill summaries, active-skill toggles, and bootstrap-file settings.	None
bootstrap_files	Sequence[str] | None	Optional explicit bootstrap filenames to load when bootstrap prompting is enabled.	None
include_unavailable_skills	bool	Whether unavailable catalog skills may be included when deriving prompt records from runtime prompt state.	False

返回：

类型	描述
'ContextBuilder'	A configured ContextBuilder instance for skill-aware prompting.

源代码位于： jianmu/memory/context/builder.py

@classmethod
def for_skill(
    cls,
    *,
    system_prompt: str | None = None,
    skill_set: Any | None = None,
    tools: Sequence[Any] | None = None,
    mode: str = "summary",
    include_resources: bool = True,
    include_react_protocol: bool = False,
    max_messages: int | None = None,
    max_tokens: int | None = None,
    runtime_prompt: "PromptRuntimeContext | None" = None,
    bootstrap_files: Sequence[str] | None = None,
    include_unavailable_skills: bool = False,
) -> "ContextBuilder":
    """Build a skill-aware context builder.

    This preset can operate in two modes:

    1. Explicit mode, where ``skill_set`` already contains the selected
       skills for the run.
    2. Runtime-resolved mode, where ``runtime_prompt`` supplies a skill
       catalog or skill directory and the builder derives active skills
       and/or summaries from it.

    Args:
        system_prompt: Optional persona prompt. When omitted, the
            configured default persona prompt is used.
        skill_set: Optional explicitly selected skill set for the run.
        tools: Optional tools available alongside the selected skills.
        mode: Prompt rendering mode for explicit skills, typically
            ``"summary"`` or ``"full"``.
        include_resources: Whether to include runtime-visible skill
            resource sections for explicit skills.
        include_react_protocol: Whether to prepend the ReAct protocol
            contract used by ReAct-backed skill execution.
        max_messages: Optional maximum number of messages retained after
            filtering.
        max_tokens: Optional token budget applied after aggregation.
        runtime_prompt: Optional execution-scoped prompt overrides.
            ``for_skill`` consumes the full skill-related portion of the
            runtime prompt, including skill catalogs, skill summaries,
            active-skill toggles, and bootstrap-file settings.
        bootstrap_files: Optional explicit bootstrap filenames to load when
            bootstrap prompting is enabled.
        include_unavailable_skills: Whether unavailable catalog skills may
            be included when deriving prompt records from runtime prompt
            state.

    Returns:
        A configured ``ContextBuilder`` instance for skill-aware prompting.
    """
    from jianmu.memory.context.providers import (
        BootstrapFilesProvider,
        SkillSetPromptProvider,
        StaticPromptProvider,
        StateHistoryProvider,
    )
    from jianmu.skill import SkillSet

    (
        skills_catalog,
        include_skills_summary,
        include_active_skills,
        include_bootstrap_files,
        workspace_dir,
    ) = _resolve_runtime_prompt_settings(runtime_prompt=runtime_prompt)
    resolved_bootstrap_files = tuple(bootstrap_files) if bootstrap_files is not None else tuple(get_bootstrap_files())

    providers: list[MessageProvider] = []
    persona_prompt = system_prompt if system_prompt is not None else get_persona_default_prompt()
    if persona_prompt:
        providers.append(StaticPromptProvider(persona_prompt))
    if include_react_protocol:
        providers.append(StaticPromptProvider(get_react_protocol_prompt()))
    if include_bootstrap_files:
        providers.append(BootstrapFilesProvider(workspace_dir=workspace_dir, files=resolved_bootstrap_files))

    effective_skill_set = skill_set
    if effective_skill_set is None and skills_catalog is not None:
        effective_skill_set = SkillSet([], catalog=skills_catalog)

    if effective_skill_set is not None:
        if getattr(effective_skill_set, "skills", None):
            providers.append(
                SkillSetPromptProvider(
                    effective_skill_set,
                    mode=mode,
                    tools=tools,
                    include_resources=include_resources,
                )
            )
        elif skills_catalog is not None and (include_active_skills or include_skills_summary):
            records = skills_catalog.select_records_for_context(
                include_unavailable_skills=include_unavailable_skills,
            )
            if include_active_skills:
                providers.append(
                    SkillSetPromptProvider(
                        effective_skill_set,
                        records=records,
                        mode="full",
                        include_resources=False,
                        always_only=True,
                    )
                )
            if include_skills_summary:
                providers.append(
                    SkillSetPromptProvider(
                        effective_skill_set,
                        records=records,
                        mode="summary",
                        include_resources=False,
                    )
                )
    providers.append(StateHistoryProvider())
    return cls(
        providers=providers,
        filters=_build_default_filters(
            max_messages=max_messages,
            max_tokens=max_tokens,
        ),
    )

resolve_for_llm classmethod

resolve_for_llm(
    *,
    explicit_builder: ContextBuilderProtocol | None = None,
    runtime_prompt: "PromptRuntimeContext | None" = None,
    system_prompt: str | None = None,
    tools_desc: str | None = None,
    prefer_react: bool = False,
    allow_skill_context: bool = False,
    skill_set: Any | None = None,
    skill_tools: Sequence[Any] | None = None,
    skill_mode: str = "summary",
    include_skill_resources: bool = True,
    include_react_protocol: bool = False,
    include_unavailable_skills: bool = False,
) -> ContextBuilderProtocol

Resolve the effective builder for one LLM-style node invocation.

Resolution order is:

1. Explicit builder passed directly to the node.
2. Runtime-provided builder from PromptRuntimeContext.
3. for_skill(...) when the caller explicitly enables skill-aware context and the runtime requests skill-aware prompting.
4. for_react(...) when the caller prefers ReAct semantics or tool descriptions are present.
5. for_chat(...) otherwise.

参数：

名称	类型	描述	默认
explicit_builder	ContextBuilderProtocol | None	Optional builder provided directly by the caller.	None
runtime_prompt	'PromptRuntimeContext | None'	Optional execution-scoped prompt overrides.	None
system_prompt	str | None	Optional base system prompt forwarded to the resolved preset builder.	None
tools_desc	str | None	Optional preformatted tool description block. Its presence selects the ReAct preset.	None
prefer_react	bool	Whether this call site should resolve the ReAct preset even when no tool descriptions are present yet.	False
allow_skill_context	bool	Whether this call site allows skill-aware prompt assembly.	False
skill_set	Any | None	Optional explicitly selected skill set for the run.	None
skill_tools	Sequence[Any] | None	Optional tools visible alongside the selected skills.	None
skill_mode	str	Prompt rendering mode for skill-aware prompting.	'summary'
include_skill_resources	bool	Whether runtime-visible skill resources should be rendered when skill-aware prompting is selected.	True
include_react_protocol	bool	Whether ReAct protocol semantics should be prepended when skill-aware prompting is selected.	False
include_unavailable_skills	bool	Whether unavailable catalog skills may be included when deriving skill context from runtime prompt.	False

返回：

类型	描述
ContextBuilderProtocol	The effective ContextBuilderProtocol for this invocation.

源代码位于： jianmu/memory/context/builder.py

@classmethod
def resolve_for_llm(
    cls,
    *,
    explicit_builder: ContextBuilderProtocol | None = None,
    runtime_prompt: "PromptRuntimeContext | None" = None,
    system_prompt: str | None = None,
    tools_desc: str | None = None,
    prefer_react: bool = False,
    allow_skill_context: bool = False,
    skill_set: Any | None = None,
    skill_tools: Sequence[Any] | None = None,
    skill_mode: str = "summary",
    include_skill_resources: bool = True,
    include_react_protocol: bool = False,
    include_unavailable_skills: bool = False,
) -> ContextBuilderProtocol:
    """Resolve the effective builder for one LLM-style node invocation.

    Resolution order is:

    1. Explicit builder passed directly to the node.
    2. Runtime-provided builder from ``PromptRuntimeContext``.
    3. ``for_skill(...)`` when the caller explicitly enables skill-aware
       context and the runtime requests skill-aware prompting.
    4. ``for_react(...)`` when the caller prefers ReAct semantics or tool
       descriptions are present.
    5. ``for_chat(...)`` otherwise.

    Args:
        explicit_builder: Optional builder provided directly by the caller.
        runtime_prompt: Optional execution-scoped prompt overrides.
        system_prompt: Optional base system prompt forwarded to the
            resolved preset builder.
        tools_desc: Optional preformatted tool description block. Its
            presence selects the ReAct preset.
        prefer_react: Whether this call site should resolve the ReAct
            preset even when no tool descriptions are present yet.
        allow_skill_context: Whether this call site allows skill-aware
            prompt assembly.
        skill_set: Optional explicitly selected skill set for the run.
        skill_tools: Optional tools visible alongside the selected skills.
        skill_mode: Prompt rendering mode for skill-aware prompting.
        include_skill_resources: Whether runtime-visible skill resources
            should be rendered when skill-aware prompting is selected.
        include_react_protocol: Whether ReAct protocol semantics should be
            prepended when skill-aware prompting is selected.
        include_unavailable_skills: Whether unavailable catalog skills may
            be included when deriving skill context from runtime prompt.

    Returns:
        The effective ``ContextBuilderProtocol`` for this invocation.
    """
    if explicit_builder is not None:
        return explicit_builder

    runtime_builder = runtime_prompt.context_builder if runtime_prompt is not None else None
    if runtime_builder is not None:
        return runtime_builder

    if allow_skill_context and (skill_set is not None or _requests_skill_context(runtime_prompt)):
        return cls.for_skill(
            system_prompt=system_prompt,
            skill_set=skill_set,
            tools=skill_tools,
            mode=skill_mode,
            include_resources=include_skill_resources,
            include_react_protocol=include_react_protocol,
            runtime_prompt=runtime_prompt,
            include_unavailable_skills=include_unavailable_skills,
        )

    if prefer_react or tools_desc:
        return cls.for_react(
            system_prompt=system_prompt,
            tools_desc=tools_desc,
            runtime_prompt=runtime_prompt,
        )

    return cls.for_chat(
        system_prompt=system_prompt,
        runtime_prompt=runtime_prompt,
    )

build

build(
    local_state: Any,
    global_state: Any = None,
    ctx: Any | None = None,
    **kwargs: Any,
) -> Sequence[Message]

Build context messages by running providers and then filters.

参数：

名称	类型	描述	默认
local_state	Any	State owned by the current agent or node.	必需
global_state	Any	Optional shared state visible across agents or trees.	None
ctx	Any | None	Optional runtime context object.	None
**kwargs	Any	Reserved for future pipeline extensions.	{}

返回：

类型	描述
Sequence[Message]	Ordered message sequence produced by the builder.

源代码位于： jianmu/memory/context/builder.py

def build(
    self,
    local_state: Any,
    global_state: Any = None,
    ctx: Any | None = None,
    **kwargs: Any,
) -> Sequence[Message]:
    """Build context messages by running providers and then filters.

    Args:
        local_state: State owned by the current agent or node.
        global_state: Optional shared state visible across agents or trees.
        ctx: Optional runtime context object.
        **kwargs: Reserved for future pipeline extensions.

    Returns:
        Ordered message sequence produced by the builder.
    """
    messages: List[Message] = []
    for provider in self.providers:
        try:
            messages.extend(provider.get_messages(local_state, global_state, ctx))
        except Exception as e:
            logger.debug("Context provider {} failed: {}", type(provider).__name__, e)

    for filt in self.filters:
        try:
            messages = list(
                filt.apply(
                    messages,
                    local_state=local_state,
                    global_state=global_state,
                    ctx=ctx,
                )
            )
        except Exception as e:
            logger.debug("Context filter {} failed: {}", type(filt).__name__, e)

    return messages

transform

transform(
    messages: Sequence[Message],
    *,
    local_state: Any = None,
    global_state: Any = None,
    ctx: Any | None = None,
    **kwargs: Any,
) -> Sequence[Message]

Apply only the filter phase to an existing message sequence.

参数：

名称	类型	描述	默认
messages	Sequence[Message]	Existing message sequence to transform.	必需
local_state	Any	State owned by the current agent or node.	None
global_state	Any	Optional shared state visible across agents or trees.	None
ctx	Any | None	Optional runtime context object.	None
**kwargs	Any	Reserved for future pipeline extensions.	{}

返回：

类型	描述
Sequence[Message]	Filtered message sequence.

源代码位于： jianmu/memory/context/builder.py

def transform(
    self,
    messages: Sequence[Message],
    *,
    local_state: Any = None,
    global_state: Any = None,
    ctx: Any | None = None,
    **kwargs: Any,
) -> Sequence[Message]:
    """Apply only the filter phase to an existing message sequence.

    Args:
        messages: Existing message sequence to transform.
        local_state: State owned by the current agent or node.
        global_state: Optional shared state visible across agents or trees.
        ctx: Optional runtime context object.
        **kwargs: Reserved for future pipeline extensions.

    Returns:
        Filtered message sequence.
    """
    out = list(messages)
    for filt in self.filters:
        try:
            out = list(
                filt.apply(
                    out,
                    local_state=local_state,
                    global_state=global_state,
                    ctx=ctx,
                )
            )
        except Exception as e:
            logger.debug("Context filter {} failed: {}", type(filt).__name__, e)
    return out

TokenCounterProtocol

Bases: Protocol

Estimate token counts for individual messages and message sequences.

count_message

count_message(message: Message) -> int

Estimate token count for one message.

参数：

名称	类型	描述	默认
message	Message	The message to count.	必需

返回：

类型	描述
int	Estimated token count for the message.

源代码位于： jianmu/memory/context/base.py

def count_message(self, message: Message) -> int:
    """Estimate token count for one message.

    Args:
        message: The message to count.

    Returns:
        Estimated token count for the message.
    """
    ...

count_messages

count_messages(messages: Iterable[Message]) -> int

Estimate token count for a sequence of messages.

参数：

名称	类型	描述	默认
messages	Iterable[Message]	Iterable of messages to count.	必需

返回：

类型	描述
int	Estimated total token count for the sequence.

源代码位于： jianmu/memory/context/base.py

def count_messages(self, messages: Iterable[Message]) -> int:
    """Estimate token count for a sequence of messages.

    Args:
        messages: Iterable of messages to count.

    Returns:
        Estimated total token count for the sequence.
    """
    ...

SimpleTokenCounter

SimpleTokenCounter(chars_per_token: int = 4)

Bases: TokenCounterProtocol

Approximate token counter based on character length.

属性：

名称	类型	描述
chars_per_token		Approximate character-to-token ratio used for estimates.

Configure the approximate characters-per-token ratio.

源代码位于： jianmu/memory/context/filters.py

def __init__(self, chars_per_token: int = 4):
    """Configure the approximate characters-per-token ratio."""
    self.chars_per_token = max(1, int(chars_per_token))

count_message

count_message(message: Message) -> int

Estimate token count for one message.

参数：

名称	类型	描述	默认
message	Message	The Message instance to count.	必需

返回：

类型	描述
int	The estimated token count.

源代码位于： jianmu/memory/context/filters.py

def count_message(self, message: Message) -> int:
    """Estimate token count for one message.

    Args:
        message: The Message instance to count.

    Returns:
        The estimated token count.
    """
    text = message.to_text()
    return max(1, len(text) // self.chars_per_token)

count_messages

count_messages(messages: Iterable[Message]) -> int

Estimate token count for a sequence of messages.

参数：

名称	类型	描述	默认
messages	Iterable[Message]	Iterable of Message instances.	必需

返回：

类型	描述
int	Sum of estimated token counts.

源代码位于： jianmu/memory/context/filters.py

def count_messages(self, messages: Iterable[Message]) -> int:
    """Estimate token count for a sequence of messages.

    Args:
        messages: Iterable of Message instances.

    Returns:
        Sum of estimated token counts.
    """
    return sum(self.count_message(m) for m in messages)

TiktokenTokenCounter

TiktokenTokenCounter(
    *,
    model: str | None = None,
    encoding_name: str | None = None,
)

Bases: TokenCounterProtocol

Token counter backed by the optional tiktoken package.

属性：

名称	类型	描述
_encoding		tiktoken encoding object used for counting.

Configure a tiktoken-based token counter.

参数：

名称	类型	描述	默认
model	str | None	Optional model name used to resolve the encoding.	None
encoding_name	str | None	Optional explicit encoding name. Takes precedence over model.	None

引发：

类型	描述
RuntimeError	If tiktoken is not installed.
ValueError	If neither model nor encoding_name resolves to a valid encoding.

源代码位于： jianmu/memory/context/filters.py

def __init__(
    self,
    *,
    model: str | None = None,
    encoding_name: str | None = None,
):
    """Configure a ``tiktoken``-based token counter.

    Args:
        model: Optional model name used to resolve the encoding.
        encoding_name: Optional explicit encoding name. Takes precedence over model.

    Raises:
        RuntimeError: If ``tiktoken`` is not installed.
        ValueError: If neither ``model`` nor ``encoding_name`` resolves to a valid encoding.
    """
    try:
        import tiktoken
    except ImportError as exc:
        raise RuntimeError("tiktoken is not installed. Run: pip install tiktoken") from exc

    try:
        if encoding_name:
            self._encoding = tiktoken.get_encoding(encoding_name)
        elif model:
            self._encoding = tiktoken.encoding_for_model(model)
        else:
            self._encoding = tiktoken.get_encoding("cl100k_base")
    except Exception as exc:
        target = encoding_name or model or "cl100k_base"
        raise RuntimeError(
            "Failed to initialize tiktoken encoding "
            f"'{target}'. Ensure the encoding is valid and cached locally, "
            "or run once with network access."
        ) from exc

count_message

count_message(message: Message) -> int

Estimate token count for one message using tiktoken.

参数：

名称	类型	描述	默认
message	Message	The message to count.	必需

返回：

类型	描述
int	Estimated token count for the message.

源代码位于： jianmu/memory/context/filters.py

def count_message(self, message: Message) -> int:
    """Estimate token count for one message using ``tiktoken``.

    Args:
        message: The message to count.

    Returns:
        Estimated token count for the message.
    """
    text = message.to_text()
    if not text:
        return 1
    return max(1, len(self._encoding.encode(text)))

count_messages

count_messages(messages: Iterable[Message]) -> int

Estimate token count for a message sequence using tiktoken.

参数：

名称	类型	描述	默认
messages	Iterable[Message]	Messages to process.	必需

返回：

类型	描述
int	Estimated total token count for the sequence.

源代码位于： jianmu/memory/context/filters.py

def count_messages(self, messages: Iterable[Message]) -> int:
    """Estimate token count for a message sequence using ``tiktoken``.

    Args:
        messages: Messages to process.

    Returns:
        Estimated total token count for the sequence.
    """
    return sum(self.count_message(message) for message in messages)

HFTokenCounter

HFTokenCounter(
    model_name: str | None = None,
    *,
    tokenizer: Any | None = None,
    use_fast: bool = True,
)

Bases: TokenCounterProtocol

Token counter backed by an optional HuggingFace tokenizer.

属性：

名称	类型	描述
_tokenizer		HuggingFace tokenizer instance used for counting.

Configure a HuggingFace tokenizer-based counter.

参数：

名称	类型	描述	默认
model_name	str | None	Model or tokenizer identifier for AutoTokenizer loading.	None
tokenizer	Any | None	Optional prebuilt tokenizer instance. If provided, model_name is not required.	None
use_fast	bool	Whether to prefer fast tokenizers when loading from model name.	True

引发：

类型	描述
RuntimeError	If transformers is not installed.
ValueError	If neither tokenizer nor model_name is provided.

源代码位于： jianmu/memory/context/filters.py

def __init__(
    self,
    model_name: str | None = None,
    *,
    tokenizer: Any | None = None,
    use_fast: bool = True,
):
    """Configure a HuggingFace tokenizer-based counter.

    Args:
        model_name: Model or tokenizer identifier for ``AutoTokenizer`` loading.
        tokenizer: Optional prebuilt tokenizer instance. If provided, ``model_name`` is not required.
        use_fast: Whether to prefer fast tokenizers when loading from model name.

    Raises:
        RuntimeError: If ``transformers`` is not installed.
        ValueError: If neither ``tokenizer`` nor ``model_name`` is provided.
    """
    if tokenizer is not None:
        self._tokenizer = tokenizer
        return
    if not model_name:
        raise ValueError("model_name is required when tokenizer is not provided")
    try:
        from transformers import AutoTokenizer
    except ImportError as exc:
        raise RuntimeError("transformers is not installed. Run: pip install transformers") from exc
    try:
        self._tokenizer = AutoTokenizer.from_pretrained(model_name, use_fast=use_fast)
    except Exception as exc:
        raise RuntimeError(
            "Failed to initialize HuggingFace tokenizer "
            f"'{model_name}'. Ensure the tokenizer exists locally or run once "
            "with network access to cache it."
        ) from exc

count_message

count_message(message: Message) -> int

Estimate token count for one message using a HuggingFace tokenizer.

参数：

名称	类型	描述	默认
message	Message	The message to count.	必需

返回：

类型	描述
int	Estimated token count for the message.

源代码位于： jianmu/memory/context/filters.py

def count_message(self, message: Message) -> int:
    """Estimate token count for one message using a HuggingFace tokenizer.

    Args:
        message: The message to count.

    Returns:
        Estimated token count for the message.
    """
    text = message.to_text()
    if not text:
        return 1
    encoded = self._tokenizer.encode(text, add_special_tokens=False)
    return max(1, len(encoded))

count_messages

count_messages(messages: Iterable[Message]) -> int

Estimate token count for a message sequence using a HuggingFace tokenizer.

参数：

名称	类型	描述	默认
messages	Iterable[Message]	Messages to process.	必需

返回：

类型	描述
int	Estimated total token count for the sequence.

源代码位于： jianmu/memory/context/filters.py

def count_messages(self, messages: Iterable[Message]) -> int:
    """Estimate token count for a message sequence using a HuggingFace tokenizer.

    Args:
        messages: Messages to process.

    Returns:
        Estimated total token count for the sequence.
    """
    return sum(self.count_message(message) for message in messages)

MaxMessagesFilter

MaxMessagesFilter(
    max_messages: int,
    pinned_roles: Sequence[str] = ("system",),
)

Bases: ContextFilter

Limit message count while keeping leading pinned roles.

属性：

名称	类型	描述
max_messages		Maximum number of messages preserved after filtering.
pinned_roles		Leading roles that remain pinned at the front of context.

Configure a message-count limit with pinned leading roles.

源代码位于： jianmu/memory/context/filters.py

def __init__(
    self, 
    max_messages: int, 
    pinned_roles: Sequence[str] = ("system",)
):
    """Configure a message-count limit with pinned leading roles."""
    self.max_messages = max_messages
    self.pinned_roles = tuple(pinned_roles)

apply

apply(
    messages: Sequence[Message],
    local_state: Any = None,
    global_state: Any = None,
    ctx: Any | None = None,
    **kwargs: Any,
) -> Sequence[Message]

Trim message count while preserving the pinned prefix.

参数：

名称	类型	描述	默认
messages	Sequence[Message]	Incoming message sequence.	必需
local_state	Any	Local state object.	None
global_state	Any	Global state object.	None
ctx	Any | None	Execution context.	None
**kwargs	Any	Extra parameters.	{}

返回：

类型	描述
Sequence[Message]	Trimmed sequence of messages.

源代码位于： jianmu/memory/context/filters.py

def apply(
    self,
    messages: Sequence[Message],
    local_state: Any = None,
    global_state: Any = None,
    ctx: Any | None = None,
    **kwargs: Any,
) -> Sequence[Message]:
    """Trim message count while preserving the pinned prefix.

    Args:
        messages: Incoming message sequence.
        local_state: Local state object.
        global_state: Global state object.
        ctx: Execution context.
        **kwargs: Extra parameters.

    Returns:
        Trimmed sequence of messages.
    """
    if self.max_messages is None or len(messages) <= self.max_messages:
        return list(messages)

    pinned_count = _pinned_prefix_count(messages, self.pinned_roles)
    pinned = list(messages[:pinned_count])
    tail = list(messages[pinned_count:])

    if self.max_messages <= pinned_count:
        return pinned[: self.max_messages]

    keep_tail = self.max_messages - pinned_count
    units = _history_trim_units(tail)
    selected_reversed: list[list[Message]] = []
    retained_count = 0
    for unit in reversed(units):
        next_count = retained_count + len(unit)
        if next_count > keep_tail:
            break
        selected_reversed.append(unit)
        retained_count = next_count
    return pinned + [message for unit in reversed(selected_reversed) for message in unit]

TokenBudgetFilter

TokenBudgetFilter(
    max_tokens: int,
    token_counter: TokenCounterProtocol | None = None,
    pinned_roles: Sequence[str] = ("system",),
    trim_from_start: bool = True,
)

Bases: ContextFilter

Truncate messages to fit within a token budget.

属性：

名称	类型	描述
max_tokens		Maximum token budget allowed for the final message list.
token_counter		Token counter used to estimate message costs.
pinned_roles		Leading roles that remain pinned at the front of context.
trim_from_start		Whether trimming removes older tail messages first.

Configure token-budget trimming while preserving pinned roles.

源代码位于： jianmu/memory/context/filters.py

def __init__(
    self,
    max_tokens: int,
    token_counter: TokenCounterProtocol | None = None,
    pinned_roles: Sequence[str] = ("system",),
    trim_from_start: bool = True,
):
    """Configure token-budget trimming while preserving pinned roles."""
    self.max_tokens = max_tokens
    self.token_counter = token_counter or SimpleTokenCounter()
    self.pinned_roles = tuple(pinned_roles)
    self.trim_from_start = trim_from_start

apply

apply(
    messages: Sequence[Message],
    local_state: Any = None,
    global_state: Any = None,
    ctx: Any | None = None,
    **kwargs: Any,
) -> Sequence[Message]

Trim messages until the token budget is satisfied.

参数：

名称	类型	描述	默认
messages	Sequence[Message]	Incoming message sequence.	必需
local_state	Any	Local state object.	None
global_state	Any	Global state object.	None
ctx	Any | None	Execution context.	None
**kwargs	Any	Extra parameters.	{}

返回：

类型	描述
Sequence[Message]	Trimmed sequence of messages.

源代码位于： jianmu/memory/context/filters.py

def apply(
    self,
    messages: Sequence[Message],
    local_state: Any = None,
    global_state: Any = None,
    ctx: Any | None = None,
    **kwargs: Any,
) -> Sequence[Message]:
    """Trim messages until the token budget is satisfied.

    Args:
        messages: Incoming message sequence.
        local_state: Local state object.
        global_state: Global state object.
        ctx: Execution context.
        **kwargs: Extra parameters.

    Returns:
        Trimmed sequence of messages.
    """
    msgs = list(messages)
    if self.max_tokens is None:
        return msgs

    # Keep pinned prefix intact
    pinned_count = _pinned_prefix_count(msgs, self.pinned_roles)
    pinned = msgs[:pinned_count]
    tail = msgs[pinned_count:]

    def total_tokens(seq: Sequence[Message]) -> int:
        """Estimate token count for a message sequence.

        Args:
            seq: Message sequence to count.

        Returns:
            Estimated token count sum.
        """
        return self.token_counter.count_messages(seq)

    all_msgs = pinned + tail
    if total_tokens(all_msgs) <= self.max_tokens:
        return all_msgs

    tail_units = _history_trim_units(tail)

    # Truncate tail until within budget, preserving assistant tool-call
    # messages together with their matching tool observations.
    while tail_units and total_tokens(pinned + [msg for unit in tail_units for msg in unit]) > self.max_tokens:
        if self.trim_from_start:
            tail_units.pop(0)
        else:
            tail_units.pop()
    return pinned + [msg for unit in tail_units for msg in unit]

StaticPromptProvider

StaticPromptProvider(prompt: str)

Bases: MessageProvider

Inject a static system prompt.

属性：

名称	类型	描述
prompt		Static system prompt text emitted by the provider.

Initialize the provider.

参数：

名称	类型	描述	默认
prompt	str	Static system prompt text to inject when non-empty.	必需

源代码位于： jianmu/memory/context/providers.py

def __init__(self, prompt: str):
    """Initialize the provider.

    Args:
        prompt: Static system prompt text to inject when non-empty.
    """
    self.prompt = prompt

get_messages

get_messages(
    local_state: Any,
    global_state: Any = None,
    ctx: Any | None = None,
) -> Sequence[Message]

Return the configured static prompt as a system message.

参数：

名称	类型	描述	默认
local_state	Any	Node-local state object.	必需
global_state	Any	Optional shared state placeholder.	None
ctx	Any | None	Optional execution context.	None

返回：

类型	描述
Sequence[Message]	A sequence containing the system message or empty sequence.

源代码位于： jianmu/memory/context/providers.py

def get_messages(self, local_state: Any, global_state: Any = None, ctx: Any | None = None) -> Sequence[Message]:
    """Return the configured static prompt as a system message.

    Args:
        local_state: Node-local state object.
        global_state: Optional shared state placeholder.
        ctx: Optional execution context.

    Returns:
        A sequence containing the system message or empty sequence.
    """
    return [system(self.prompt)] if self.prompt else []

ToolsDescProvider

ToolsDescProvider(tools_desc: str)

Bases: MessageProvider

Inject tool descriptions as a system message.

属性：

名称	类型	描述
tools_desc		Preformatted tool description block rendered into context.

Initialize the provider.

参数：

名称	类型	描述	默认
tools_desc	str	Preformatted tool description block.	必需

源代码位于： jianmu/memory/context/providers.py

def __init__(self, tools_desc: str):
    """Initialize the provider.

    Args:
        tools_desc: Preformatted tool description block.
    """
    self.tools_desc = tools_desc

get_messages

get_messages(
    local_state: Any,
    global_state: Any = None,
    ctx: Any | None = None,
) -> Sequence[Message]

Return a system message describing available tools.

参数：

名称	类型	描述	默认
local_state	Any	Node-local state object.	必需
global_state	Any	Optional shared state placeholder.	None
ctx	Any | None	Optional execution context.	None

返回：

类型	描述
Sequence[Message]	A sequence containing the system message or empty sequence.

源代码位于： jianmu/memory/context/providers.py

def get_messages(self, local_state: Any, global_state: Any = None, ctx: Any | None = None) -> Sequence[Message]:
    """Return a system message describing available tools.

    Args:
        local_state: Node-local state object.
        global_state: Optional shared state placeholder.
        ctx: Optional execution context.

    Returns:
        A sequence containing the system message or empty sequence.
    """
    if not self.tools_desc:
        return []
    return [system(format_tools_prompt(self.tools_desc))]

StateHistoryProvider

StateHistoryProvider(
    state_key: str = "messages",
    working_memory: Any | None = None,
)

Bases: MessageProvider

Load conversation history from state and normalize runtime semantics.

属性：

名称	类型	描述
state_key		State field used to read prior messages.
working_memory		Optional working-memory adapter applied to history.

Initialize the provider.

参数：

名称	类型	描述	默认
state_key	str	State field that stores the message history.	'messages'
working_memory	Any | None	Optional working-memory adapter with build_view.	None

源代码位于： jianmu/memory/context/providers.py

def __init__(self, state_key: str = "messages", working_memory: Any | None = None):
    """Initialize the provider.

    Args:
        state_key: State field that stores the message history.
        working_memory: Optional working-memory adapter with ``build_view``.
    """
    self.state_key = state_key
    self.working_memory = working_memory

get_messages

get_messages(
    local_state: Any,
    global_state: Any = None,
    ctx: Any | None = None,
) -> Sequence[Message]

Build history messages from the current state snapshot.

This provider also annotates prior assistant tool plans and tool receipts so downstream models do not confuse them with new incoming human messages.

参数：

名称	类型	描述	默认
local_state	Any	Node-local state or message container.	必需
global_state	Any	Unused shared state placeholder for protocol parity.	None
ctx	Any | None	Optional runtime context that may expose working memory.	None

返回：

类型	描述
Sequence[Message]	Normalized message history ready for context assembly.

源代码位于： jianmu/memory/context/providers.py

def get_messages(self, local_state: Any, global_state: Any = None, ctx: Any | None = None) -> Sequence[Message]:
    """Build history messages from the current state snapshot.

    This provider also annotates prior assistant tool plans and tool receipts so
    downstream models do not confuse them with new incoming human messages.

    Args:
        local_state: Node-local state or message container.
        global_state: Unused shared state placeholder for protocol parity.
        ctx: Optional runtime context that may expose working memory.

    Returns:
        Normalized message history ready for context assembly.
    """
    raw = []
    if isinstance(local_state, (list, tuple)):
        raw = list(local_state)
    elif isinstance(local_state, dict):
        raw = list(local_state.get(self.state_key) or [])
    elif hasattr(local_state, self.state_key):
        raw = list(getattr(local_state, self.state_key) or [])
    msgs = [_coerce_message(item) for item in raw]
    normalized_msgs: list[Message] = []
    has_semantic_annotations = False
    for msg in msgs:
        normalized, changed = _normalize_history_message(msg)
        normalized_msgs.append(normalized)
        has_semantic_annotations = has_semantic_annotations or changed
    msgs = normalized_msgs

    if has_semantic_annotations:
        msgs = [
            system(get_history_semantics_prompt()),
            *msgs,
        ]

    wm = self.working_memory or getattr(ctx, "working_memory", None)
    if wm is not None and hasattr(wm, "build_view"):
        try:
            msgs = wm.build_view(msgs)
        except Exception as e:
            logger.debug("StateHistoryProvider working_memory.build_view failed: {}", e)
    return msgs

InMemoryCheckpointer

InMemoryCheckpointer()

Bases: CheckpointerProtocol

Simple in-memory checkpointer for tests and transient sessions.

属性：

名称	类型	描述
_store	Dict[str, dict]	In-memory checkpoint map keyed by thread ID.

Initialize the in-memory checkpoint map.

源代码位于： jianmu/memory/checkpoint/in_memory.py

def __init__(self):
    """Initialize the in-memory checkpoint map."""
    self._store: Dict[str, dict] = {}

save_checkpoint

save_checkpoint(
    thread_id: str,
    step: int,
    state_dump: dict,
    tree_state: Optional[dict] = None,
    runtime_metadata: Optional[dict] = None,
) -> None

Store a checkpoint payload in memory.

参数：

名称	类型	描述	默认
thread_id	str	Logical thread/session identifier.	必需
step	int	Step tick index.	必需
state_dump	dict	Serialized state dict.	必需
tree_state	Optional[dict]	Optional serialized behavior tree state.	None
runtime_metadata	Optional[dict]	Optional runner-internal metadata payload.	None

源代码位于： jianmu/memory/checkpoint/in_memory.py

def save_checkpoint(
    self,
    thread_id: str,
    step: int,
    state_dump: dict,
    tree_state: Optional[dict] = None,
    runtime_metadata: Optional[dict] = None,
) -> None:
    """Store a checkpoint payload in memory.

    Args:
        thread_id: Logical thread/session identifier.
        step: Step tick index.
        state_dump: Serialized state dict.
        tree_state: Optional serialized behavior tree state.
        runtime_metadata: Optional runner-internal metadata payload.
    """
    envelope = make_checkpoint_envelope(
        step=step,
        state_dump=copy.deepcopy(state_dump),
        tree_state=copy.deepcopy(tree_state) if tree_state is not None else None,
        runtime_metadata=copy.deepcopy(runtime_metadata) if runtime_metadata is not None else None,
    )
    self._store[thread_id] = envelope

get_checkpoint

get_checkpoint(thread_id: str) -> Optional[dict]

Load an in-memory checkpoint payload for one thread id.

参数：

名称	类型	描述	默认
thread_id	str	Logical thread/session identifier.	必需

返回：

类型	描述
Optional[dict]	The loaded checkpoint dictionary if found, or None.

源代码位于： jianmu/memory/checkpoint/in_memory.py

def get_checkpoint(self, thread_id: str) -> Optional[dict]:
    """Load an in-memory checkpoint payload for one thread id.

    Args:
        thread_id: Logical thread/session identifier.

    Returns:
        The loaded checkpoint dictionary if found, or None.
    """
    entry = self._store.get(thread_id)
    if not entry:
        return None

    data = copy.deepcopy(entry)
    return extract_checkpoint_payload(data, source="InMemoryCheckpointer")

FileCheckpointer

FileCheckpointer(storage_dir: str | None = None)

Bases: CheckpointerProtocol

JSONL file-based checkpointer for lightweight persistent runs.

FileCheckpointer appends one checkpoint envelope per line and restores only the latest line for a given thread id. It is appropriate for local apps, demos, and simple resumable workflows.

属性：

名称	类型	描述
storage_dir		Directory containing per-thread checkpoint JSONL files.

Initialize the JSONL checkpoint directory.

参数：

名称	类型	描述	默认
storage_dir	str | None	Optional directory override. When omitted, Jianmu config determines the checkpoint directory.	None

源代码位于： jianmu/memory/checkpoint/file_checkpointer.py

def __init__(self, storage_dir: str | None = None):
    """Initialize the JSONL checkpoint directory.

    Args:
        storage_dir: Optional directory override. When omitted, Jianmu
            config determines the checkpoint directory.
    """
    if storage_dir is None:
        from jianmu.config.loader import get_config
        storage_dir = get_config().paths.checkpoints_dir
    self.storage_dir = Path(storage_dir)
    self.storage_dir.mkdir(parents=True, exist_ok=True)

save_checkpoint

save_checkpoint(
    thread_id: str,
    step: int,
    state_dump: dict,
    tree_state: Optional[dict] = None,
    runtime_metadata: Optional[dict] = None,
) -> None

Persist a checkpoint payload to the backing store.

参数：

名称	类型	描述	默认
thread_id	str	Logical thread/session identifier.	必需
step	int	Step tick index.	必需
state_dump	dict	Serialized state dict.	必需
tree_state	Optional[dict]	Optional serialized behavior tree state.	None
runtime_metadata	Optional[dict]	Optional runner-internal metadata payload.	None

源代码位于： jianmu/memory/checkpoint/file_checkpointer.py

def save_checkpoint(
    self,
    thread_id: str,
    step: int,
    state_dump: dict,
    tree_state: Optional[dict] = None,
    runtime_metadata: Optional[dict] = None,
) -> None:
    """Persist a checkpoint payload to the backing store.

    Args:
        thread_id: Logical thread/session identifier.
        step: Step tick index.
        state_dump: Serialized state dict.
        tree_state: Optional serialized behavior tree state.
        runtime_metadata: Optional runner-internal metadata payload.
    """
    path = self.storage_dir / f"{thread_id}.jsonl"
    envelope = make_checkpoint_envelope(step, state_dump, tree_state, runtime_metadata)
    with path.open("a", encoding="utf-8") as f:
        f.write(json.dumps(envelope) + "\n")

get_checkpoint

get_checkpoint(thread_id: str) -> Optional[dict]

Load a checkpoint payload for one thread id.

参数：

名称	类型	描述	默认
thread_id	str	Logical thread/session identifier.	必需

返回：

类型	描述
Optional[dict]	The loaded checkpoint dictionary if found, or None.

源代码位于： jianmu/memory/checkpoint/file_checkpointer.py

def get_checkpoint(self, thread_id: str) -> Optional[dict]:
    """Load a checkpoint payload for one thread id.

    Args:
        thread_id: Logical thread/session identifier.

    Returns:
        The loaded checkpoint dictionary if found, or None.
    """
    path = self.storage_dir / f"{thread_id}.jsonl"
    if not path.exists():
        return None
    try:
        with path.open("r", encoding="utf-8") as f:
            lines = f.readlines()
            if not lines:
                return None
            last_line = lines[-1].strip()
            if not last_line:
                return None
            data = json.loads(last_line)
            return extract_checkpoint_payload(data, source="FileCheckpointer")
    except Exception:
        return None

NullCheckpointer

Bases: CheckpointerProtocol

No-op checkpointer used in tests or ephemeral runs.

save_checkpoint

save_checkpoint(
    thread_id: str,
    step: int,
    state_dump: dict,
    tree_state: Optional[dict] = None,
    runtime_metadata: Optional[dict] = None,
) -> None

Ignore checkpoint save requests.

参数：

名称	类型	描述	默认
thread_id	str	Session identifier.	必需
step	int	Step tick index.	必需
state_dump	dict	Serialized state dict.	必需
tree_state	Optional[dict]	Optional serialized behavior tree state.	None
runtime_metadata	Optional[dict]	Optional runner-internal metadata payload.	None

源代码位于： jianmu/memory/checkpoint/base.py

def save_checkpoint(
    self,
    thread_id: str,
    step: int,
    state_dump: dict,
    tree_state: Optional[dict] = None,
    runtime_metadata: Optional[dict] = None,
) -> None:
    """Ignore checkpoint save requests.

    Args:
        thread_id: Session identifier.
        step: Step tick index.
        state_dump: Serialized state dict.
        tree_state: Optional serialized behavior tree state.
        runtime_metadata: Optional runner-internal metadata payload.
    """
    return None

get_checkpoint

get_checkpoint(thread_id: str) -> Optional[dict]

Always return no checkpoint.

参数：

名称	类型	描述	默认
thread_id	str	Session identifier.	必需

返回：

类型	描述
Optional[dict]	Always returns None.

源代码位于： jianmu/memory/checkpoint/base.py

def get_checkpoint(self, thread_id: str) -> Optional[dict]:
    """Always return no checkpoint.

    Args:
        thread_id: Session identifier.

    Returns:
        Always returns None.
    """
    return None

LongTermMemoryProtocol

Bases: Protocol

Public capability protocol for long-term memory integrations.

This protocol models what a long-term memory system can do for an agent: record durable information and retrieve relevant memory later. It does not assume the backend is a key-value store or expose deletion as a core capability.

record

record(
    inputs: LongTermMemoryInput,
    *,
    namespace: tuple[str, ...] | None = None,
    **kwargs: Any,
) -> LongTermMemoryRecordResult

Record information into long-term memory.

参数：

名称	类型	描述	默认
inputs	LongTermMemoryInput	Text or message payload to persist.	必需
namespace	tuple[str, ...] | None	Optional namespace for isolating memory records.	None
**kwargs	Any	Backend-specific record options.	{}

返回：

类型	描述
LongTermMemoryRecordResult	Backend-defined record result payload.

源代码位于： jianmu/memory/long_term/base.py

def record(
    self,
    inputs: LongTermMemoryInput,
    *,
    namespace: tuple[str, ...] | None = None,
    **kwargs: Any,
) -> LongTermMemoryRecordResult:
    """Record information into long-term memory.

    Args:
        inputs: Text or message payload to persist.
        namespace: Optional namespace for isolating memory records.
        **kwargs: Backend-specific record options.

    Returns:
        Backend-defined record result payload.
    """
    ...

retrieve

retrieve(
    query: LongTermMemoryQuery,
    *,
    limit: int = 5,
    namespace: tuple[str, ...] | None = None,
    **kwargs: Any,
) -> LongTermMemoryRetrieved

Retrieve relevant information from long-term memory.

参数：

名称	类型	描述	默认
query	LongTermMemoryQuery	Text or message payload describing the retrieval target.	必需
limit	int	Maximum number of results to return.	5
namespace	tuple[str, ...] | None	Optional namespace for isolating memory retrieval.	None
**kwargs	Any	Backend-specific retrieval options.	{}

返回：

类型	描述
LongTermMemoryRetrieved	A list of retrieved memory items.

源代码位于： jianmu/memory/long_term/base.py

def retrieve(
    self,
    query: LongTermMemoryQuery,
    *,
    limit: int = 5,
    namespace: tuple[str, ...] | None = None,
    **kwargs: Any,
) -> LongTermMemoryRetrieved:
    """Retrieve relevant information from long-term memory.

    Args:
        query: Text or message payload describing the retrieval target.
        limit: Maximum number of results to return.
        namespace: Optional namespace for isolating memory retrieval.
        **kwargs: Backend-specific retrieval options.

    Returns:
        A list of retrieved memory items.
    """
    ...

format_retrieved

format_retrieved(
    retrieved: LongTermMemoryRetrieved, **kwargs: Any
) -> str

Render retrieved memory into LLM-consumable text.

参数：

名称	类型	描述	默认
retrieved	LongTermMemoryRetrieved	Retrieved memory items to render.	必需
**kwargs	Any	Backend-specific formatting options.	{}

返回：

类型	描述
str	Formatted text suitable for tools or prompt injection.

源代码位于： jianmu/memory/long_term/base.py

def format_retrieved(
    self,
    retrieved: LongTermMemoryRetrieved,
    **kwargs: Any,
) -> str:
    """Render retrieved memory into LLM-consumable text.

    Args:
        retrieved: Retrieved memory items to render.
        **kwargs: Backend-specific formatting options.

    Returns:
        Formatted text suitable for tools or prompt injection.
    """
    ...

Mem0LongTermMemory

Mem0LongTermMemory(
    api_key: str | None = None,
    config: dict[str, Any] | None = None,
    *,
    infer: bool = True,
)

Mem0-backed implementation of LongTermMemoryProtocol.

This integration treats Mem0 as a memory capability rather than as a key-value store. Recording and retrieval keep namespace explicit while leaving backend-specific result structure flexible.

属性：

名称	类型	描述
client		Mem0 client or local memory instance used for operations.
_is_client		Whether the hosted MemoryClient mode is active.
_infer		Whether semantic fact extraction is enabled during writes.

Initialize a Mem0-backed long-term memory instance.

参数：

名称	类型	描述	默认
api_key	str | None	Optional API key for hosted mem0.MemoryClient mode.	None
config	dict[str, Any] | None	Optional configuration passed to local mem0.Memory.	None
infer	bool	Whether Mem0 should perform semantic fact extraction when recording memory.	True

源代码位于： jianmu/memory/long_term/mem0.py

def __init__(
    self,
    api_key: str | None = None,
    config: dict[str, Any] | None = None,
    *,
    infer: bool = True,
) -> None:
    """Initialize a Mem0-backed long-term memory instance.

    Args:
        api_key: Optional API key for hosted ``mem0.MemoryClient`` mode.
        config: Optional configuration passed to local ``mem0.Memory``.
        infer: Whether Mem0 should perform semantic fact extraction when
            recording memory.
    """
    try:
        import mem0
    except ImportError as e:
        raise ImportError(
            "The 'mem0ai' package is required to use Mem0LongTermMemory. "
            "Please install it with `pip install mem0ai`."
        ) from e

    if api_key:
        self.client = mem0.MemoryClient(api_key=api_key)
        self._is_client = True
    else:
        if config is None:
            self.client = mem0.Memory()
        elif isinstance(config, dict):
            self.client = mem0.Memory.from_config(config)
        else:
            self.client = mem0.Memory(config=config)
        self._is_client = False
    self._infer = bool(infer)

record

record(
    inputs: LongTermMemoryInput,
    *,
    namespace: tuple[str, ...] | None = None,
    **kwargs: Any,
) -> LongTermMemoryRecordResult

Record information into Mem0.

参数：

名称	类型	描述	默认
inputs	LongTermMemoryInput	Text or message payload to persist.	必需
namespace	tuple[str, ...] | None	Optional Jianmu namespace mapped to Mem0 user_id.	None
**kwargs	Any	Optional backend-specific options such as metadata.	{}

返回：

类型	描述
LongTermMemoryRecordResult	A backend-shaped record result with normalized results when
LongTermMemoryRecordResult	available.

源代码位于： jianmu/memory/long_term/mem0.py

def record(
    self,
    inputs: LongTermMemoryInput,
    *,
    namespace: tuple[str, ...] | None = None,
    **kwargs: Any,
) -> LongTermMemoryRecordResult:
    """Record information into Mem0.

    Args:
        inputs: Text or message payload to persist.
        namespace: Optional Jianmu namespace mapped to Mem0 ``user_id``.
        **kwargs: Optional backend-specific options such as ``metadata``.

    Returns:
        A backend-shaped record result with normalized ``results`` when
        available.
    """
    text = self._coerce_text(inputs)
    if not text:
        return {"results": []}

    metadata = dict(kwargs.get("metadata") or {})
    key = kwargs.get("key")
    if isinstance(key, str) and key:
        metadata.setdefault("key", key)

    payload = self.client.add(
        text,
        user_id=self._namespace_id(namespace),
        metadata=metadata or None,
        infer=self._infer,
    )
    return {
        "results": [
            normalized
            for normalized in (
                self._normalize_item(item) for item in self._extract_results(payload)
            )
            if normalized is not None
        ],
        "raw": payload,
    }

retrieve

retrieve(
    query: LongTermMemoryQuery,
    *,
    limit: int = 5,
    namespace: tuple[str, ...] | None = None,
    **kwargs: Any,
) -> LongTermMemoryRetrieved

Retrieve relevant Mem0 memories.

参数：

名称	类型	描述	默认
query	LongTermMemoryQuery	Search query text or message payload.	必需
limit	int	Maximum number of results to return.	5
namespace	tuple[str, ...] | None	Optional Jianmu namespace mapped to Mem0 user_id.	None
**kwargs	Any	Reserved backend-specific options.	{}

返回：

类型	描述
LongTermMemoryRetrieved	A list of normalized retrieved memory dictionaries.

源代码位于： jianmu/memory/long_term/mem0.py

def retrieve(
    self,
    query: LongTermMemoryQuery,
    *,
    limit: int = 5,
    namespace: tuple[str, ...] | None = None,
    **kwargs: Any,
) -> LongTermMemoryRetrieved:
    """Retrieve relevant Mem0 memories.

    Args:
        query: Search query text or message payload.
        limit: Maximum number of results to return.
        namespace: Optional Jianmu namespace mapped to Mem0 ``user_id``.
        **kwargs: Reserved backend-specific options.

    Returns:
        A list of normalized retrieved memory dictionaries.
    """
    del kwargs
    text = self._coerce_text(query)
    if not text:
        return []

    payload = self.client.search(
        text,
        filters={"user_id": self._namespace_id(namespace)},
        top_k=limit,
    )
    return [
        normalized
        for normalized in (
            self._normalize_item(item) for item in self._extract_results(payload)
        )
        if normalized is not None
    ]

format_retrieved

format_retrieved(
    retrieved: LongTermMemoryRetrieved, **kwargs: Any
) -> str

Render retrieved memories into plain text.

参数：

名称	类型	描述	默认
retrieved	LongTermMemoryRetrieved	Normalized retrieved memory items.	必需
**kwargs	Any	Reserved formatting options.	{}

返回：

类型	描述
str	A readable text block suitable for tool output or prompt injection.

源代码位于： jianmu/memory/long_term/mem0.py

def format_retrieved(
    self,
    retrieved: LongTermMemoryRetrieved,
    **kwargs: Any,
) -> str:
    """Render retrieved memories into plain text.

    Args:
        retrieved: Normalized retrieved memory items.
        **kwargs: Reserved formatting options.

    Returns:
        A readable text block suitable for tool output or prompt injection.
    """
    del kwargs
    if not retrieved:
        return "No results."

    lines: list[str] = []
    for index, item in enumerate(retrieved, start=1):
        content = str(item.get("content", "")).strip()
        memory_id = item.get("id")
        prefix = f"{index}."
        if memory_id:
            prefix += f" id={memory_id}"
        score = item.get("score")
        if score is not None:
            prefix += f" score={score}"
        metadata = item.get("metadata")
        if metadata:
            lines.append(f"{prefix} content={content} metadata={metadata}")
        else:
            lines.append(f"{prefix} content={content}")
    return "\n".join(lines)

MemoryService

MemoryService(
    long_term_memory: LongTermMemoryProtocol,
    namespace: tuple[str, ...] = ("default",),
)

Facade over a namespace-scoped long-term memory backend.

属性：

名称	类型	描述
memory		Bound long-term memory backend implementation.
namespace		Default namespace used for read and write operations.

Bind the facade to a long-term memory backend and default namespace.

参数：

名称	类型	描述	默认
long_term_memory	LongTermMemoryProtocol	Backend used to record and retrieve memories.	必需
namespace	tuple[str, ...]	Default namespace used when a call does not override it.	('default',)

源代码位于： jianmu/memory/long_term/service.py

def __init__(
    self,
    long_term_memory: LongTermMemoryProtocol,
    namespace: tuple[str, ...] = ("default",),
) -> None:
    """Bind the facade to a long-term memory backend and default namespace.

    Args:
        long_term_memory: Backend used to record and retrieve memories.
        namespace: Default namespace used when a call does not override it.
    """
    self.memory = long_term_memory
    self.namespace = tuple(namespace)

add

add(
    content: str,
    *,
    metadata: dict[str, Any] | None = None,
    key: str,
    namespace: tuple[str, ...] | None = None,
) -> LongTermMemoryRecordResult

Record one memory value.

参数：

名称	类型	描述	默认
content	str	Text content to record.	必需
metadata	dict[str, Any] | None	Optional metadata stored with the memory item.	None
key	str	Stable backend key for the recorded memory.	必需
namespace	tuple[str, ...] | None	Optional namespace override for the write operation.	None

返回：

类型	描述
LongTermMemoryRecordResult	Backend-defined record result payload.

源代码位于： jianmu/memory/long_term/service.py

def add(
    self,
    content: str,
    *,
    metadata: dict[str, Any] | None = None,
    key: str,
    namespace: tuple[str, ...] | None = None,
) -> LongTermMemoryRecordResult:
    """Record one memory value.

    Args:
        content: Text content to record.
        metadata: Optional metadata stored with the memory item.
        key: Stable backend key for the recorded memory.
        namespace: Optional namespace override for the write operation.

    Returns:
        Backend-defined record result payload.
    """
    ns = tuple(namespace) if namespace is not None else self.namespace
    return self.memory.record(
        content,
        namespace=ns,
        metadata=metadata or {},
        key=key,
    )

search

search(
    query: str,
    *,
    limit: int = 5,
    namespace: tuple[str, ...] | None = None,
) -> LongTermMemoryRetrieved

Retrieve stored memories within the resolved namespace.

参数：

名称	类型	描述	默认
query	str	Query text to execute.	必需
limit	int	Maximum number of items to return.	5
namespace	tuple[str, ...] | None	Optional namespace override for the read operation.	None

返回：

类型	描述
LongTermMemoryRetrieved	Retrieved memory items from the backend.

源代码位于： jianmu/memory/long_term/service.py

def search(
    self,
    query: str,
    *,
    limit: int = 5,
    namespace: tuple[str, ...] | None = None,
) -> LongTermMemoryRetrieved:
    """Retrieve stored memories within the resolved namespace.

    Args:
        query: Query text to execute.
        limit: Maximum number of items to return.
        namespace: Optional namespace override for the read operation.

    Returns:
        Retrieved memory items from the backend.
    """
    ns = tuple(namespace) if namespace is not None else self.namespace
    return self.memory.retrieve(query, limit=limit, namespace=ns)

format_retrieved

format_retrieved(retrieved: LongTermMemoryRetrieved) -> str

Render retrieved memory items into text.

参数：

名称	类型	描述	默认
retrieved	LongTermMemoryRetrieved	Retrieved memory items to render.	必需

返回：

类型	描述
str	Text representation suitable for prompts or tools.

源代码位于： jianmu/memory/long_term/service.py

def format_retrieved(self, retrieved: LongTermMemoryRetrieved) -> str:
    """Render retrieved memory items into text.

    Args:
        retrieved: Retrieved memory items to render.

    Returns:
        Text representation suitable for prompts or tools.
    """
    return self.memory.format_retrieved(retrieved)

delete

delete(
    memory: Any, *, namespace: tuple[str, ...] | None = None
) -> None

Delete one memory item from the resolved namespace when supported.

参数：

名称	类型	描述	默认
memory	Any	Backend-defined memory record handle to delete.	必需
namespace	tuple[str, ...] | None	Optional namespace override for the delete operation.	None

源代码位于： jianmu/memory/long_term/service.py

def delete(self, memory: Any, *, namespace: tuple[str, ...] | None = None) -> None:
    """Delete one memory item from the resolved namespace when supported.

    Args:
        memory: Backend-defined memory record handle to delete.
        namespace: Optional namespace override for the delete operation.
    """
    ns = tuple(namespace) if namespace is not None else self.namespace
    delete = getattr(self.memory, "delete", None)
    if callable(delete):
        delete(memory, namespace=ns)

as_tools

as_tools() -> list[Tool]

Return the default read/write tool pair for this memory scope.

返回：

类型	描述
list[Tool]	A list containing search and add tools bound to this service.

源代码位于： jianmu/memory/long_term/service.py

def as_tools(self) -> list[Tool]:
    """Return the default read/write tool pair for this memory scope.

    Returns:
        A list containing search and add tools bound to this service.
    """
    from jianmu.memory.tools import MemoryAddTool, MemorySearchTool

    return [
        MemorySearchTool(memory=self),
        MemoryAddTool(memory=self),
    ]

SimpleStore

SimpleStore(db_path: str | None = None)

JSON-backed long-term memory implementation with namespace support.

属性：

名称	类型	描述
path		Filesystem path to the JSON backing store.
_data	Dict[str, Any]	In-memory nested dictionary representing persisted memory state.

Initialize the JSON-backed memory file from disk if present.

源代码位于： jianmu/memory/long_term/simple.py

def __init__(self, db_path: str | None = None) -> None:
    """Initialize the JSON-backed memory file from disk if present."""
    if db_path is None:
        from jianmu.config.loader import get_config

        db_path = get_config().paths.store
    self.path = Path(db_path)
    if self.path.exists():
        try:
            self._data: Dict[str, Any] = json.loads(
                self.path.read_text("utf-8")
            )
        except Exception:
            self._data = {}
    else:
        self._data = {}

record

record(
    inputs: LongTermMemoryInput,
    *,
    namespace: tuple[str, ...] | None = None,
    **kwargs: Any,
) -> LongTermMemoryRecordResult

Persist one memory item into the JSON store.

参数：

名称	类型	描述	默认
inputs	LongTermMemoryInput	Text or message payload to persist.	必需
namespace	tuple[str, ...] | None	Optional namespace used for storage isolation.	None
**kwargs	Any	Additional options such as key and metadata.	{}

返回：

类型	描述
LongTermMemoryRecordResult	Record result payload containing normalized stored items.

引发：

类型	描述
ValueError	If input validation fails.

源代码位于： jianmu/memory/long_term/simple.py

def record(
    self,
    inputs: LongTermMemoryInput,
    *,
    namespace: tuple[str, ...] | None = None,
    **kwargs: Any,
) -> LongTermMemoryRecordResult:
    """Persist one memory item into the JSON store.

    Args:
        inputs: Text or message payload to persist.
        namespace: Optional namespace used for storage isolation.
        **kwargs: Additional options such as ``key`` and ``metadata``.

    Returns:
        Record result payload containing normalized stored items.

    Raises:
        ValueError: If input validation fails.
    """
    text = _coerce_text(inputs)
    if not text:
        return {"results": []}

    key = str(kwargs.get("key") or "").strip()
    if not key:
        raise ValueError("SimpleStore.record requires a non-empty 'key'")

    metadata = dict(kwargs.get("metadata") or {})
    ns = _normalize_namespace(namespace)
    node = self._resolve(ns, create=True)
    node[key] = {"content": text, "metadata": metadata}
    self._persist()
    return {
        "results": [
            {
                "id": key,
                "content": text,
                "metadata": metadata,
            }
        ]
    }

retrieve

retrieve(
    query: LongTermMemoryQuery,
    *,
    limit: int = 5,
    namespace: tuple[str, ...] | None = None,
    **kwargs: Any,
) -> LongTermMemoryRetrieved

Search JSON-serialized values within one namespace.

SimpleStore is a local demo backend, so retrieval stays deliberately lightweight: exact substring match first, then fallback keyword overlap.

参数：

名称	类型	描述	默认
query	LongTermMemoryQuery	Query text to execute.	必需
limit	int	Maximum number of items to return.	5
namespace	tuple[str, ...] | None	Optional namespace used for storage isolation.	None
**kwargs	Any	Reserved backend-specific options.	{}

返回：

类型	描述
LongTermMemoryRetrieved	Retrieved memory items ordered by local relevance score.

源代码位于： jianmu/memory/long_term/simple.py

def retrieve(
    self,
    query: LongTermMemoryQuery,
    *,
    limit: int = 5,
    namespace: tuple[str, ...] | None = None,
    **kwargs: Any,
) -> LongTermMemoryRetrieved:
    """Search JSON-serialized values within one namespace.

    ``SimpleStore`` is a local demo backend, so retrieval stays deliberately
    lightweight: exact substring match first, then fallback keyword overlap.

    Args:
        query: Query text to execute.
        limit: Maximum number of items to return.
        namespace: Optional namespace used for storage isolation.
        **kwargs: Reserved backend-specific options.

    Returns:
        Retrieved memory items ordered by local relevance score.
    """
    del kwargs
    text_query = _coerce_text(query).lower()
    query_tokens = set(_tokenize_text(text_query))
    ns = _normalize_namespace(namespace)
    node = self._resolve(ns, create=False)
    if not isinstance(node, dict):
        return []

    scored: list[tuple[int, dict[str, Any]]] = []
    for key, value in node.items():
        raw_text = json.dumps(value, ensure_ascii=False)
        haystack = raw_text.lower()
        if text_query and text_query in haystack:
            score = len(query_tokens) + 1
        elif query_tokens:
            value_tokens = set(_tokenize_text(haystack))
            score = len(query_tokens & value_tokens)
            if score == 0:
                continue
        elif text_query:
            continue
        else:
            score = 1
        if isinstance(value, dict):
            content = value.get("content", "")
            metadata = dict(value.get("metadata") or {})
        else:
            content = value
            metadata = {}
        scored.append(
            (
                score,
                {
                    "id": key,
                    "content": content,
                    "metadata": metadata,
                },
            )
        )
    scored.sort(
        key=lambda item: (
            -item[0],
            str(item[1].get("id", "")),
        )
    )
    return [item for _, item in scored[:limit]]

format_retrieved

format_retrieved(
    retrieved: LongTermMemoryRetrieved, **kwargs: Any
) -> str

Render retrieved memory items into plain text.

参数：

名称	类型	描述	默认
retrieved	LongTermMemoryRetrieved	Retrieved memory items to render.	必需
**kwargs	Any	Reserved formatting options.	{}

返回：

类型	描述
str	Plain-text representation of retrieved memory items.

源代码位于： jianmu/memory/long_term/simple.py

def format_retrieved(
    self,
    retrieved: LongTermMemoryRetrieved,
    **kwargs: Any,
) -> str:
    """Render retrieved memory items into plain text.

    Args:
        retrieved: Retrieved memory items to render.
        **kwargs: Reserved formatting options.

    Returns:
        Plain-text representation of retrieved memory items.
    """
    del kwargs
    if not retrieved:
        return "No results."
    lines: list[str] = []
    for index, item in enumerate(retrieved, start=1):
        memory_id = item.get("id", index)
        content = item.get("content", "")
        metadata = item.get("metadata")
        if metadata:
            lines.append(
                f"{index}. id={memory_id} content={content} metadata={metadata}"
            )
        else:
            lines.append(f"{index}. id={memory_id} content={content}")
    return "\n".join(lines)

delete

delete(
    memory: Any, *, namespace: tuple[str, ...] | None = None
) -> None

Delete one retrieved memory item from the target namespace.

参数：

名称	类型	描述	默认
memory	Any	Memory record handle or dictionary containing an id.	必需
namespace	tuple[str, ...] | None	Optional namespace used for storage isolation.	None

源代码位于： jianmu/memory/long_term/simple.py

def delete(
    self,
    memory: Any,
    *,
    namespace: tuple[str, ...] | None = None,
) -> None:
    """Delete one retrieved memory item from the target namespace.

    Args:
        memory: Memory record handle or dictionary containing an ``id``.
        namespace: Optional namespace used for storage isolation.
    """
    memory_id = memory.get("id") if isinstance(memory, dict) else None
    if not isinstance(memory_id, str) or not memory_id:
        return
    ns = _normalize_namespace(namespace)
    node = self._resolve(ns, create=False)
    if memory_id in node:
        node.pop(memory_id)
        self._persist()

MemoryAddTool

MemoryAddTool(
    namespace: Tuple[str, ...] = ("default",),
    *,
    long_term_memory: LongTermMemoryProtocol | None = None,
    memory: MemoryService | None = None,
)

Bases: Tool

Record one piece of information into long-term memory.

属性：

名称	类型	描述
memory		Memory service used to persist records.
namespace		Default namespace used for writes.

Bind the add tool to a memory facade or namespace.

参数：

名称	类型	描述	默认
namespace	Tuple[str, ...]	Default namespace for record writes.	('default',)
long_term_memory	LongTermMemoryProtocol | None	Optional long-term memory capability backend.	None
memory	MemoryService | None	Optional high-level memory facade.	None

源代码位于： jianmu/memory/tools.py

def __init__(
    self,
    namespace: Tuple[str, ...] = ("default",),
    *,
    long_term_memory: LongTermMemoryProtocol | None = None,
    memory: MemoryService | None = None,
) -> None:
    """Bind the add tool to a memory facade or namespace.

    Args:
        namespace: Default namespace for record writes.
        long_term_memory: Optional long-term memory capability backend.
        memory: Optional high-level memory facade.
    """
    if memory is None:
        if long_term_memory is None:
            raise ValueError(
                "MemoryAddTool requires one of 'memory' or 'long_term_memory'"
            )
        memory = MemoryService(long_term_memory=long_term_memory, namespace=namespace)
    self.memory = memory
    self.namespace = memory.namespace

run async

run(
    content: Optional[str] = None,
    metadata: Optional[dict[str, Any]] = None,
    key: Optional[str] = None,
    namespace: Optional[List[str]] = None,
    **kwargs: Any,
) -> str

Store one memory value under the resolved namespace.

参数：

名称	类型	描述	默认
content	Optional[str]	The text content to store.	None
metadata	Optional[dict[str, Any]]	Optional dictionary of metadata.	None
key	Optional[str]	Optional explicit key to identify the memory.	None
namespace	Optional[List[str]]	Optional namespace path override.	None
**kwargs	Any	Extra parameters.	{}

返回：

类型	描述
str	Status message indicating success or failure.

源代码位于： jianmu/memory/tools.py

async def run(
    self,
    content: Optional[str] = None,
    metadata: Optional[dict[str, Any]] = None,
    key: Optional[str] = None,
    namespace: Optional[List[str]] = None,
    **kwargs: Any,
) -> str:
    """Store one memory value under the resolved namespace.

    Args:
        content: The text content to store.
        metadata: Optional dictionary of metadata.
        key: Optional explicit key to identify the memory.
        namespace: Optional namespace path override.
        **kwargs: Extra parameters.

    Returns:
        Status message indicating success or failure.
    """
    content = content or kwargs.get("input", "")
    if not content:
        return "Error: no content provided"

    ns = tuple(namespace) if namespace else self.namespace
    store_key = key or str(uuid.uuid4())
    try:
        result = self.memory.add(content, metadata=metadata, key=store_key, namespace=ns)
        results = result.get("results") if isinstance(result, dict) else None
        if isinstance(results, list) and results:
            first = results[0] if isinstance(results[0], dict) else {}
            memory_id = first.get("id")
            if memory_id:
                return f"Stored (id: {memory_id}, key: {store_key})"
        return f"Stored (key: {store_key})"
    except Exception as e:
        return f"Error storing memory: {e}"

MemorySearchTool

MemorySearchTool(
    namespace: Tuple[str, ...] = ("default",),
    top_k: int = 5,
    *,
    long_term_memory: LongTermMemoryProtocol | None = None,
    memory: MemoryService | None = None,
)

Bases: Tool

Search long-term memory from a memory backend.

This is a lightweight long-term memory retrieval tool, not a full RAG pipeline.

属性：

名称	类型	描述
memory		Memory service used to execute retrieval.
namespace		Default namespace used for searches.
top_k		Default maximum number of results returned by the tool.

Bind the search tool to a memory facade or namespace.

参数：

名称	类型	描述	默认
namespace	Tuple[str, ...]	Default namespace for retrieval.	('default',)
top_k	int	Default maximum number of results.	5
long_term_memory	LongTermMemoryProtocol | None	Optional long-term memory capability backend.	None
memory	MemoryService | None	Optional high-level memory facade.	None

源代码位于： jianmu/memory/tools.py

def __init__(
    self,
    namespace: Tuple[str, ...] = ("default",),
    top_k: int = 5,
    *,
    long_term_memory: LongTermMemoryProtocol | None = None,
    memory: MemoryService | None = None,
) -> None:
    """Bind the search tool to a memory facade or namespace.

    Args:
        namespace: Default namespace for retrieval.
        top_k: Default maximum number of results.
        long_term_memory: Optional long-term memory capability backend.
        memory: Optional high-level memory facade.
    """
    if memory is None:
        if long_term_memory is None:
            raise ValueError(
                "MemorySearchTool requires one of 'memory' or 'long_term_memory'"
            )
        memory = MemoryService(long_term_memory=long_term_memory, namespace=namespace)
    self.memory = memory
    self.namespace = memory.namespace
    self.top_k = top_k

run async

run(
    query: Optional[str] = None,
    k: Optional[int] = None,
    namespace: Optional[List[str]] = None,
    **kwargs: Any,
) -> str

Search memory records and return normalized result text.

参数：

名称	类型	描述	默认
query	Optional[str]	Query string to search for.	None
k	Optional[int]	Maximum number of results to return.	None
namespace	Optional[List[str]]	Optional list representing namespace path override.	None
**kwargs	Any	Extra parameters.	{}

返回：

类型	描述
str	Formatted search results string or error.

源代码位于： jianmu/memory/tools.py

async def run(
    self,
    query: Optional[str] = None,
    k: Optional[int] = None,
    namespace: Optional[List[str]] = None,
    **kwargs: Any,
) -> str:
    """Search memory records and return normalized result text.

    Args:
        query: Query string to search for.
        k: Maximum number of results to return.
        namespace: Optional list representing namespace path override.
        **kwargs: Extra parameters.

    Returns:
        Formatted search results string or error.
    """
    query = query or kwargs.get("input", "")
    if not query:
        return "Error: no query provided"

    ns = tuple(namespace) if namespace else self.namespace
    limit = k or self.top_k
    try:
        retrieved = self.memory.search(query, limit=limit, namespace=ns)
    except Exception as e:
        logger.debug("MemorySearchTool memory search failed: {}", e)
        retrieved = []

    rendered = self.memory.format_retrieved(retrieved)
    if rendered and rendered != "No results.":
        return rendered
    if not retrieved:
        return "No results."

    lines: List[str] = []
    for i, item in enumerate(retrieved, 1):
        if isinstance(item, dict):
            lines.append(f"{i}. {json.dumps(item, ensure_ascii=False)}")
        else:
            lines.append(f"{i}. {item}")
    return "\n".join(lines)

create_memory_tools

create_memory_tools(
    namespace: Tuple[str, ...] = ("default",),
    *,
    long_term_memory: LongTermMemoryProtocol | None = None,
    memory: MemoryService | None = None,
) -> list[Tool]

Create the default read/write tool pair for memory access.

参数：

名称	类型	描述	默认
namespace	Tuple[str, ...]	Default store namespace.	('default',)
long_term_memory	LongTermMemoryProtocol | None	Optional long-term memory capability backend.	None
memory	MemoryService | None	Optional high-level memory facade.	None

返回：

类型	描述
list[Tool]	List containing MemorySearchTool and MemoryAddTool.

引发：

类型	描述
ValueError	If input validation fails.

源代码位于： jianmu/memory/tools.py

def create_memory_tools(
    namespace: Tuple[str, ...] = ("default",),
    *,
    long_term_memory: LongTermMemoryProtocol | None = None,
    memory: MemoryService | None = None,
) -> list[Tool]:
    """Create the default read/write tool pair for memory access.

    Args:
        namespace: Default store namespace.
        long_term_memory: Optional long-term memory capability backend.
        memory: Optional high-level memory facade.

    Returns:
        List containing ``MemorySearchTool`` and ``MemoryAddTool``.

    Raises:
        ValueError: If input validation fails.
    """
    if memory is None:
        if long_term_memory is None:
            raise ValueError(
                "create_memory_tools requires one of 'memory' or 'long_term_memory'"
            )
        memory = MemoryService(long_term_memory=long_term_memory, namespace=namespace)
    return memory.as_tools()