Memory Module

Memory module.

class atloop.memory.AgentState(step: int = 0, phase: str = 'DISCOVER', last_error: ~atloop.memory.state.LastError = <factory>, memory: ~atloop.memory.state.Memory = <factory>, artifacts: ~atloop.memory.state.Artifacts = <factory>, budget_used: ~atloop.memory.state.BudgetUsed = <factory>)[source]

Bases: object

Agent execution state.

__init__(step: int = 0, phase: str = 'DISCOVER', last_error: ~atloop.memory.state.LastError = <factory>, memory: ~atloop.memory.state.Memory = <factory>, artifacts: ~atloop.memory.state.Artifacts = <factory>, budget_used: ~atloop.memory.state.BudgetUsed = <factory>) → None
classmethod from_dict(data: Dict[str, Any]) → AgentState[source]

Create from dictionary.

phase: str = 'DISCOVER'
step: int = 0
to_dict() → Dict[str, Any][source]

Convert to dictionary.

last_error: LastError
memory: Memory
artifacts: Artifacts
budget_used: BudgetUsed
class atloop.memory.LastError(summary: str = '', repro_cmd: str = '', raw_stderr_tail: str = '', error_signature: str = '')[source]

Bases: object

Last error information.

__init__(summary: str = '', repro_cmd: str = '', raw_stderr_tail: str = '', error_signature: str = '') → None
error_signature: str = ''
raw_stderr_tail: str = ''
repro_cmd: str = ''
summary: str = ''
class atloop.memory.Memory(created_files: ~typing.List[str] = <factory>, attempts: ~typing.List[~typing.Dict[str, ~typing.Any]] = <factory>, key_files: ~typing.List[~typing.Dict[str, ~typing.Any]] = <factory>, notes: ~typing.List[str] = <factory>, tool_results_history: ~typing.List[~typing.Dict[str, ~typing.Any]] = <factory>, modified_files_content: ~typing.List[~typing.Dict[str, ~typing.Any]] = <factory>, action_history: ~typing.List[~typing.Dict[str, ~typing.Any]] = <factory>, plan: str | ~typing.List[~typing.Any] = <factory>, task_summary: str = '', important_decisions: ~typing.List[~typing.Dict[str, ~typing.Any]] = <factory>, milestones: ~typing.List[~typing.Dict[str, ~typing.Any]] = <factory>, learnings: ~typing.List[str] = <factory>, skill_cache: ~typing.Dict[str, ~typing.Dict[str, ~typing.Any]] = <factory>, decisions: ~typing.List[~typing.Dict[str, ~typing.Any]] = <factory>, llm_responses: ~typing.List[~typing.Dict[str, ~typing.Any]] = <factory>)[source]

Bases: object

Memory for tracking decisions and attempts.

Memory is organized into three categories:

  1. FACTS (used for LLM context): - created_files, modified_files_content, tool_results_history - These are objective, verifiable facts from tool execution

  2. LONG-TERM (used for LLM context): - plan, task_summary, milestones, learnings - These are validated/verified information

  3. DEBUG-ONLY (NOT fed back to LLM): - decisions, llm_responses - These contain LLM’s interpretations which could cause feedback loops

Memory 模块负责： - ✓ 原始数据的存储和管理 - ✓ 各个条目的格式转换和控制输出（考虑约束：单条长度、字符串映射等） - ✓ 提供统一的格式化接口，返回可直接注入 prompt 的字符串

Memory 模块不负责： - ❌ 数据压缩（由独立的 CompressionPolicy 负责） - ❌ 数据重要性评分（由独立的 Scorer 负责，如果使用）

__init__(created_files: ~typing.List[str] = <factory>, attempts: ~typing.List[~typing.Dict[str, ~typing.Any]] = <factory>, key_files: ~typing.List[~typing.Dict[str, ~typing.Any]] = <factory>, notes: ~typing.List[str] = <factory>, tool_results_history: ~typing.List[~typing.Dict[str, ~typing.Any]] = <factory>, modified_files_content: ~typing.List[~typing.Dict[str, ~typing.Any]] = <factory>, action_history: ~typing.List[~typing.Dict[str, ~typing.Any]] = <factory>, plan: str | ~typing.List[~typing.Any] = <factory>, task_summary: str = '', important_decisions: ~typing.List[~typing.Dict[str, ~typing.Any]] = <factory>, milestones: ~typing.List[~typing.Dict[str, ~typing.Any]] = <factory>, learnings: ~typing.List[str] = <factory>, skill_cache: ~typing.Dict[str, ~typing.Dict[str, ~typing.Any]] = <factory>, decisions: ~typing.List[~typing.Dict[str, ~typing.Any]] = <factory>, llm_responses: ~typing.List[~typing.Dict[str, ~typing.Any]] = <factory>) → None
get_formatted_context(state: AgentState, task_goal: str | None = None, max_length: int | None = None, format_options: Dict[str, Any] | None = None, tool_registry: Any | None = None) → str[source]

获取格式化后的记忆上下文，可直接注入到 prompt 中。

这是 Memory 模块的主要输出接口，返回格式化的字符串。

Parameters:

  • state – AgentState 实例（需要访问 memory, last_error, artifacts）

  • task_goal – 任务目标（可选，用于任务概览）

  • max_length – 最大长度限制（可选）

  • format_options –

    格式选项（可选，会覆盖配置中的默认值） - tool_results_count: int (默认从 MemoryConfig 读取) - steps_summary_count: int (默认从 MemoryConfig 读取) - include_file_content: bool (默认从 MemoryConfig 读取) - max_file_content_length: int (默认从 MemoryConfig 读取)

    注意：所有默认值现在从 MemoryConfig 读取，确保单一数据源。 可以通过 format_options 参数覆盖特定调用的值。 - string_mappings: Dict[str, str] (字符串映射规则)

  • tool_registry – 工具注册表（用于输出限制策略）

Returns:

格式化后的字符串，可直接用于 prompt 注入 格式：符合 MEMORY_PROMPT_FORMAT_DEMO.md 中定义的格式

task_summary: str = ''
created_files: List[str]
attempts: List[Dict[str, Any]]
key_files: List[Dict[str, Any]]
notes: List[str]
tool_results_history: List[Dict[str, Any]]
modified_files_content: List[Dict[str, Any]]
action_history: List[Dict[str, Any]]
plan: str | List[Any]
important_decisions: List[Dict[str, Any]]
milestones: List[Dict[str, Any]]
learnings: List[str]
skill_cache: Dict[str, Dict[str, Any]]
decisions: List[Dict[str, Any]]
llm_responses: List[Dict[str, Any]]
class atloop.memory.Artifacts(current_diff: str = '', test_results: str = '', verification_success: bool | None = None, dod_result: Any | None = None)[source]

Bases: object

Artifacts produced during execution.

__init__(current_diff: str = '', test_results: str = '', verification_success: bool | None = None, dod_result: Any | None = None) → None
current_diff: str = ''
dod_result: Any | None = None
test_results: str = ''
verification_success: bool | None = None
class atloop.memory.BudgetUsed(llm_calls: int = 0, tool_calls: int = 0, wall_time_sec: int = 0)[source]

Bases: object

Budget usage tracking.

__init__(llm_calls: int = 0, tool_calls: int = 0, wall_time_sec: int = 0) → None
llm_calls: int = 0
tool_calls: int = 0
wall_time_sec: int = 0
class atloop.memory.MemorySummarizer[source]

Bases: object

Summarize agent memory for LLM input.

static get_memory_overview(state: AgentState) → str[source]

Get a brief overview of memory for terminal output.

Parameters:

state – Agent state

Returns:

Brief overview string (single line, compact format)

static summarize(state: AgentState, max_length: int | None = None, task_goal: str | None = None, tool_registry: Any | None = None) → str[source]

Summarize agent state memory.

Parameters:

  • state – Agent state

  • max_length – Maximum summary length (defaults to config value)

  • task_goal – Optional task goal for completion detection

Returns:

Summary string

MemoryManager

class atloop.memory.memory_manager.MemoryManager[source]

Bases: object

Manager for dynamic memory updates.

static update_plan(state: AgentState, plan: str, reason: str | None = None) → None[source]

Update the execution plan in long-term memory.

Parameters:

  • state – Agent state

  • plan – New plan text

  • reason – Optional reason for the update

static add_important_decision(state: AgentState, content: str, step: int | None = None, context: Dict[str, Any] | None = None) → None[source]

Add an important decision to long-term memory.

Parameters:

  • state – Agent state

  • content – Decision content

  • step – Step number (defaults to current step)

  • context – Optional context information

static add_milestone(state: AgentState, content: str, step: int | None = None, context: Dict[str, Any] | None = None) → None[source]

Add a milestone to long-term memory.

Parameters:

  • state – Agent state

  • content – Milestone content

  • step – Step number (defaults to current step)

  • context – Optional context information

static add_learning(state: AgentState, learning: str, step: int | None = None) → None[source]

Add a learning to long-term memory.

Parameters:

  • state – Agent state

  • learning – Learning content

  • step – Step number (defaults to current step)

static update_task_summary(state: AgentState, summary: str) → None[source]

Update task summary in long-term memory.

Parameters:

  • state – Agent state

  • summary – Task summary text

static get_long_term_memory_summary(state: AgentState) → str[source]

Get a summary of long-term memory.

Parameters:

state – Agent state

Returns:

Summary string

MemorySummarizer

class atloop.memory.summarizer.MemorySummarizer[source]

Bases: object

Summarize agent memory for LLM input.

static get_memory_overview(state: AgentState) → str[source]

Get a brief overview of memory for terminal output.

Parameters:

state – Agent state

Returns:

Brief overview string (single line, compact format)

static summarize(state: AgentState, max_length: int | None = None, task_goal: str | None = None, tool_registry: Any | None = None) → str[source]

Summarize agent state memory.

Parameters:

  • state – Agent state

  • max_length – Maximum summary length (defaults to config value)

  • task_goal – Optional task goal for completion detection

Returns:

Summary string

MemoryCompressor

class atloop.memory.compressor.MemoryCompressor[source]

Bases: object

Compress old memory to prevent unbounded growth.

ATTEMPTS_KEEP_RECENT = 10
DECISIONS_KEEP_RECENT = 5
IMPORTANT_DECISIONS_KEEP = 20
MILESTONES_KEEP = 20
LEARNINGS_KEEP = 10
static compress_if_needed(state: AgentState, memory_config=None, llm_client=None) → bool[source]

Compress memory if it exceeds limits.

Phase 4: Enhanced with LLM compression and deduplication.

Parameters:

  • state – Agent state

  • memory_config – Optional MemoryConfig instance (if None, uses defaults)

  • llm_client – Optional LLMClient for LLM compression (if None, LLM compression is skipped)

Returns:

True if compression was performed, False otherwise

AgentState

class atloop.memory.state.AgentState(step: int = 0, phase: str = 'DISCOVER', last_error: ~atloop.memory.state.LastError = <factory>, memory: ~atloop.memory.state.Memory = <factory>, artifacts: ~atloop.memory.state.Artifacts = <factory>, budget_used: ~atloop.memory.state.BudgetUsed = <factory>)[source]

Bases: object

Agent execution state.

step: int = 0
phase: str = 'DISCOVER'
last_error: LastError
memory: Memory
artifacts: Artifacts
budget_used: BudgetUsed
to_dict() → Dict[str, Any][source]

Convert to dictionary.

classmethod from_dict(data: Dict[str, Any]) → AgentState[source]

Create from dictionary.

__init__(step: int = 0, phase: str = 'DISCOVER', last_error: ~atloop.memory.state.LastError = <factory>, memory: ~atloop.memory.state.Memory = <factory>, artifacts: ~atloop.memory.state.Artifacts = <factory>, budget_used: ~atloop.memory.state.BudgetUsed = <factory>) → None