Cognitive History¶

Long-running agents accumulate extensive execution history. Without management, this history would eventually overflow the LLM's context window. The framework's CognitiveHistory implements a three-tier memory system that automatically compresses older memories while keeping recent details readily available.

In this tutorial, we'll build a multi-step data collection agent with a deliberately small memory configuration (working_memory_size=1, short_term_size=2) so we can observe how the memory tiers work with just a handful of steps.

Initialize¶

In [1]:

  Copied!     
 
import os

model_name = os.environ.get("MODEL_NAME")
api_key = os.environ.get("API_KEY")
api_base = os.environ.get("BASE_URL")
import os model_name = os.environ.get("MODEL_NAME") api_key = os.environ.get("API_KEY") api_base = os.environ.get("BASE_URL")

In [ ]:

  Copied!     
 
from bridgic.llms.openai import OpenAILlm, OpenAIConfiguration

llm = OpenAILlm(
    api_key=api_key,
    api_base=api_base,
    timeout=120,
    configuration=OpenAIConfiguration(
        model=model_name,
        temperature=0.0,
        max_tokens=16384,
    ),
)
from bridgic.llms.openai import OpenAILlm, OpenAIConfiguration llm = OpenAILlm( api_key=api_key, api_base=api_base, timeout=120, configuration=OpenAIConfiguration( model=model_name, temperature=0.0, max_tokens=16384, ), )

In [3]:

  Copied!     
 
from bridgic.core.agentic.tool_specs import FunctionToolSpec


async def scrape_page(url: str) -> str:
    """Scrape data from a web page"""
    return f"Scraped data from {url}: 150 records found, 12 fields per record"


async def clean_data(source: str, rules: str) -> str:
    """Clean and validate scraped data"""
    return f"Cleaned data from {source}: removed 12 duplicates, fixed 5 format issues"


async def store_data(table: str, record_count: int) -> str:
    """Store processed data in the database"""
    return f"Stored {record_count} records in table '{table}'"


scrape_page_tool = FunctionToolSpec.from_raw(scrape_page)
clean_data_tool = FunctionToolSpec.from_raw(clean_data)
store_data_tool = FunctionToolSpec.from_raw(store_data)
from bridgic.core.agentic.tool_specs import FunctionToolSpec async def scrape_page(url: str) -> str: """Scrape data from a web page""" return f"Scraped data from {url}: 150 records found, 12 fields per record" async def clean_data(source: str, rules: str) -> str: """Clean and validate scraped data""" return f"Cleaned data from {source}: removed 12 duplicates, fixed 5 format issues" async def store_data(table: str, record_count: int) -> str: """Store processed data in the database""" return f"Stored {record_count} records in table '{table}'" scrape_page_tool = FunctionToolSpec.from_raw(scrape_page) clean_data_tool = FunctionToolSpec.from_raw(clean_data) store_data_tool = FunctionToolSpec.from_raw(store_data)

The Three-Tier Memory Model¶

CognitiveHistory organizes agent execution history into three tiers, each with a different level of detail:

Tier	Content	LLM Visibility
Working Memory	Most recent N steps	Full details — tool calls, arguments, return values, everything
Short-term Memory	Next M older steps	One-line summaries; full details available on demand via Acquiring
Long-term Memory	Oldest steps	LLM-compressed paragraph; individual step details no longer available

Working Memory is the agent's immediate context. The LLM sees every detail of recent steps, so it can reason precisely about what just happened.
Short-term Memory trades detail for capacity. Steps are stored as summaries, but the agent can "look back" and request full details when needed (via the Acquiring policy).
Long-term Memory is the most compressed tier. When Short-term Memory overflows, the LLM compresses the oldest summaries into a single paragraph. Individual step access is no longer possible, but the gist is preserved.

CognitiveHistory Configuration¶

You configure the memory tiers by setting three parameters when creating a CognitiveHistory instance. For this tutorial, we'll use deliberately small values so the tiers activate within just a few steps:

In [4]:

  Copied!     
 
from bridgic.amphibious import CognitiveHistory

# Small configuration for this tutorial
# — only 1 step in working memory, 2 in short-term, compress after 2 pending
history = CognitiveHistory(
    working_memory_size=1,
    short_term_size=2,
    compress_threshold=2,
)
from bridgic.amphibious import CognitiveHistory # Small configuration for this tutorial # — only 1 step in working memory, 2 in short-term, compress after 2 pending history = CognitiveHistory( working_memory_size=1, short_term_size=2, compress_threshold=2, )

Watching Memory Tiers in Action¶

Let's build a data collection agent that scrapes, cleans, and stores data from two sources. With working_memory_size=1 and short_term_size=2, the tiers will activate after just a few tool calls:

After 1 step: everything in Working Memory
After 2 steps: 1 in Working, 1 moves to Short-term
After 3 steps: 1 in Working, 2 in Short-term (Short-term is full)
After 4+ steps: overflow pushes oldest steps toward Long-term compression

In [ ]:

  Copied!     
 
from bridgic.amphibious import AmphibiousAutoma, CognitiveContext, CognitiveWorker, think_unit


class DataCollector(AmphibiousAutoma[CognitiveContext]):
    collector = think_unit(
        CognitiveWorker.inline(
            "Process data sources one at a time: scrape the page, clean the data, "
            "then store it. Move to the next source after storing. ",
            verbose_prompt=True,
        ),
        max_attempts=8,
    )

    async def on_agent(self, ctx: CognitiveContext):
        await self.collector
from bridgic.amphibious import AmphibiousAutoma, CognitiveContext, CognitiveWorker, think_unit class DataCollector(AmphibiousAutoma[CognitiveContext]): collector = think_unit( CognitiveWorker.inline( "Process data sources one at a time: scrape the page, clean the data, " "then store it. Move to the next source after storing. ", verbose_prompt=True, ), max_attempts=8, ) async def on_agent(self, ctx: CognitiveContext): await self.collector

In [7]:

  Copied!     
 
agent = DataCollector(llm=llm, verbose=True)
result = await agent.arun(
    goal=(
        "Collect data from 2 sources: site-a.com/data and site-b.com/data. "
        "For each source: scrape, clean, and store the data."
    ),
    tools=[scrape_page_tool, clean_data_tool, store_data_tool],
    cognitive_history=CognitiveHistory(
        working_memory_size=1,
        short_term_size=2,
        compress_threshold=2,
    ),
)
print(result)
agent = DataCollector(llm=llm, verbose=True) result = await agent.arun( goal=( "Collect data from 2 sources: site-a.com/data and site-b.com/data. " "For each source: scrape, clean, and store the data." ), tools=[scrape_page_tool, clean_data_tool, store_data_tool], cognitive_history=CognitiveHistory( working_memory_size=1, short_term_size=2, compress_threshold=2, ), ) print(result)

[18:49:03.416] [Router] (_amphibious_automa.py:1573) Auto-detecting execution mode
[18:49:03.417] [Router] (_amphibious_automa.py:1579) Detected AGENT mode
[18:49:03.417] [Observe] (_amphibious_automa.py:861) _PromptWorker: None
[18:49:08.899] [Think] (_cognitive_worker.py:332) Message 1 (system, 347 tokens):
Process data sources one at a time: scrape the page, clean the data, then store it. Move to the next source after storing. Respond in JSON format.

# Available Tools (with parameters):
• scrape_page: Scrape data from a web page
  - url (string) [required]
• clean_data: Clean and validate scraped data
  - source (string) [required]
  - rules (string) [required]
• store_data: Store processed data in the database
  - table (string) [required]
  - record_count (integer) [required]

# Context Acquiring
If the context contains progressively disclosed information (e.g. skills, history steps) and you want to inspect the details, use the **details** field to request them. The framework will expand these items in the next round. Batch all requests in a single output. When using this field, leave step_content and output empty.

## Field format:
- **details**: [{field: "skills", index: 0}, ...]
  Available fields: **skills** (view a skill's full workflow), **cognitive_history** (view the full result of a previous step)

# Output Fields
- **step_content**: Your analysis and reasoning for this step
- **finish**: Set True when the sub-task is fully complete (default: False)
- **details**: Available fields: **skills**, **cognitive_history**. example: [{field: 'skills', index: 0}, ...]
- **output**: Tool calls to execute: [{tool, tool_arguments: [{name: 'param', value: 'value'}]}]

[18:49:08.899] [Think] (_cognitive_worker.py:332) Message 2 (user, 51 tokens):
Based on the context below, decide your next action.

Goal: Collect data from 2 sources: site-a.com/data and site-b.com/data. For each source: scrape, clean, and store the data.
Execution History: (none)
[18:49:08.899] [Think] (_cognitive_worker.py:332) Total: 398 tokens (cumulative: 398)
[18:49:08.900] [Think] (_amphibious_automa.py:867) _PromptWorker: finish=False, step=Starting data collection from 2 sources. Beginning with the first source (site-a.com/data). First step is to scrape the page data.
[18:49:08.903] [Act] (_amphibious_automa.py:873) _PromptWorker:
{
    "content": "Starting data collection from 2 sources. Beginning with the first source (site-a.com/data). First step is to scrape the page data.",
    "result": {
        "results": [
            {
                "tool_id": "call_0",
                "tool_name": "scrape_page",
                "tool_arguments": {
                    "url": "site-a.com/data"
                },
                "tool_result": "Scraped data from site-a.com/data: 150 records found, 12 fields per record",
                "success": true,
                "error": null
            }
        ]
    },
    "metadata": {},
    "status": null
}
[18:49:08.903] [Observe] (_amphibious_automa.py:861) _PromptWorker: None
[18:49:33.553] [Think] (_cognitive_worker.py:332) Message 1 (system, 347 tokens):
Process data sources one at a time: scrape the page, clean the data, then store it. Move to the next source after storing. Respond in JSON format.

# Available Tools (with parameters):
• scrape_page: Scrape data from a web page
  - url (string) [required]
• clean_data: Clean and validate scraped data
  - source (string) [required]
  - rules (string) [required]
• store_data: Store processed data in the database
  - table (string) [required]
  - record_count (integer) [required]

# Context Acquiring
If the context contains progressively disclosed information (e.g. skills, history steps) and you want to inspect the details, use the **details** field to request them. The framework will expand these items in the next round. Batch all requests in a single output. When using this field, leave step_content and output empty.

## Field format:
- **details**: [{field: "skills", index: 0}, ...]
  Available fields: **skills** (view a skill's full workflow), **cognitive_history** (view the full result of a previous step)

# Output Fields
- **step_content**: Your analysis and reasoning for this step
- **finish**: Set True when the sub-task is fully complete (default: False)
- **details**: Available fields: **skills**, **cognitive_history**. example: [{field: 'skills', index: 0}, ...]
- **output**: Tool calls to execute: [{tool, tool_arguments: [{name: 'param', value: 'value'}]}]

[18:49:33.553] [Think] (_cognitive_worker.py:332) Message 2 (user, 150 tokens):
Based on the context below, decide your next action.

Goal: Collect data from 2 sources: site-a.com/data and site-b.com/data. For each source: scrape, clean, and store the data.
Execution History:
  [Working Memory (0-0)]
    [0] Starting data collection from 2 sources. Beginning with the first source (site-a.com/data). First step is to scrape the page data.
   Result: results=[ActionStepResult(tool_id='call_0', tool_name='scrape_page', tool_arguments={'url': 'site-a.com/data'}, tool_result='Scraped data from site-a.com/data: 150 records found, 12 fields per record', success=True, error=None)]
[18:49:33.553] [Think] (_cognitive_worker.py:332) Total: 497 tokens (cumulative: 895)
[18:49:33.557] [Think] (_amphibious_automa.py:867) _PromptWorker: finish=False, step=After successfully scraping site-a.com/data (150 records), the next step is to clean the data. Applying standard validation and normalization rules to ensure consistency and quality before storage.
[18:49:33.560] [Act] (_amphibious_automa.py:873) _PromptWorker:
{
    "content": "After successfully scraping site-a.com/data (150 records), the next step is to clean the data. Applying standard validation and normalization rules to ensure consistency and quality before storage.",
    "result": {
        "results": [
            {
                "tool_id": "call_0",
                "tool_name": "clean_data",
                "tool_arguments": {
                    "source": "site-a.com/data",
                    "rules": "remove_duplicates,handle_missing_values,validate_data_types,normalize_formats"
                },
                "tool_result": "Cleaned data from site-a.com/data: removed 12 duplicates, fixed 5 format issues",
                "success": true,
                "error": null
            }
        ]
    },
    "metadata": {},
    "status": null
}
[18:49:33.560] [Observe] (_amphibious_automa.py:861) _PromptWorker: None
[18:49:55.964] [Think] (_cognitive_worker.py:332) Message 1 (system, 347 tokens):
Process data sources one at a time: scrape the page, clean the data, then store it. Move to the next source after storing. Respond in JSON format.

# Available Tools (with parameters):
• scrape_page: Scrape data from a web page
  - url (string) [required]
• clean_data: Clean and validate scraped data
  - source (string) [required]
  - rules (string) [required]
• store_data: Store processed data in the database
  - table (string) [required]
  - record_count (integer) [required]

# Context Acquiring
If the context contains progressively disclosed information (e.g. skills, history steps) and you want to inspect the details, use the **details** field to request them. The framework will expand these items in the next round. Batch all requests in a single output. When using this field, leave step_content and output empty.

## Field format:
- **details**: [{field: "skills", index: 0}, ...]
  Available fields: **skills** (view a skill's full workflow), **cognitive_history** (view the full result of a previous step)

# Output Fields
- **step_content**: Your analysis and reasoning for this step
- **finish**: Set True when the sub-task is fully complete (default: False)
- **details**: Available fields: **skills**, **cognitive_history**. example: [{field: 'skills', index: 0}, ...]
- **output**: Tool calls to execute: [{tool, tool_arguments: [{name: 'param', value: 'value'}]}]

[18:49:55.964] [Think] (_cognitive_worker.py:332) Message 2 (user, 255 tokens):
Based on the context below, decide your next action.

Goal: Collect data from 2 sources: site-a.com/data and site-b.com/data. For each source: scrape, clean, and store the data.
Execution History:
  (Steps [0-0]: summary only, query details via details)
  [Short-term Memory (0-0), query details via 'details']
    [0] Starting data collection from 2 sources. Beginning with the first source (site-a.com/data). First step is to scrape the page data.
  [Working Memory (1-1)]
    [1] After successfully scraping site-a.com/data (150 records), the next step is to clean the data. Applying standard validation and normalization rules to ensure consistency and quality before storage.
   Result: results=[ActionStepResult(tool_id='call_0', tool_name='clean_data', tool_arguments={'source': 'site-a.com/data', 'rules': 'remove_duplicates,handle_missing_values,validate_data_types,normalize_formats'}, tool_result='Cleaned data from site-a.com/data: removed 12 duplicates, fixed 5 format issues', success=True, error=None)]
[18:49:55.964] [Think] (_cognitive_worker.py:332) Total: 602 tokens (cumulative: 1497)
[18:49:55.967] [Think] (_amphibious_automa.py:867) _PromptWorker: finish=False, step=After cleaning the data from site-a.com/data (138 valid records remaining), the next step is to store it in the database. Once stored, proceed to process the second source (site-b.com/data).
[18:49:55.969] [Act] (_amphibious_automa.py:873) _PromptWorker:
{
    "content": "After cleaning the data from site-a.com/data (138 valid records remaining), the next step is to store it in the database. Once stored, proceed to process the second source (site-b.com/data).",
    "result": {
        "results": [
            {
                "tool_id": "call_0",
                "tool_name": "store_data",
                "tool_arguments": {
                    "table": "collected_data",
                    "record_count": 138
                },
                "tool_result": "Stored 138 records in table 'collected_data'",
                "success": true,
                "error": null
            }
        ]
    },
    "metadata": {},
    "status": null
}
[18:49:55.969] [Observe] (_amphibious_automa.py:861) _PromptWorker: None
[18:50:04.446] [Think] (_cognitive_worker.py:332) Message 1 (system, 347 tokens):
Process data sources one at a time: scrape the page, clean the data, then store it. Move to the next source after storing. Respond in JSON format.

# Available Tools (with parameters):
• scrape_page: Scrape data from a web page
  - url (string) [required]
• clean_data: Clean and validate scraped data
  - source (string) [required]
  - rules (string) [required]
• store_data: Store processed data in the database
  - table (string) [required]
  - record_count (integer) [required]

# Context Acquiring
If the context contains progressively disclosed information (e.g. skills, history steps) and you want to inspect the details, use the **details** field to request them. The framework will expand these items in the next round. Batch all requests in a single output. When using this field, leave step_content and output empty.

## Field format:
- **details**: [{field: "skills", index: 0}, ...]
  Available fields: **skills** (view a skill's full workflow), **cognitive_history** (view the full result of a previous step)

# Output Fields
- **step_content**: Your analysis and reasoning for this step
- **finish**: Set True when the sub-task is fully complete (default: False)
- **details**: Available fields: **skills**, **cognitive_history**. example: [{field: 'skills', index: 0}, ...]
- **output**: Tool calls to execute: [{tool, tool_arguments: [{name: 'param', value: 'value'}]}]

[18:50:04.446] [Think] (_cognitive_worker.py:332) Message 2 (user, 278 tokens):
Based on the context below, decide your next action.

Goal: Collect data from 2 sources: site-a.com/data and site-b.com/data. For each source: scrape, clean, and store the data.
Execution History:
  (Steps [0-1]: summary only, query details via details)
  [Short-term Memory (0-1), query details via 'details']
    [0] Starting data collection from 2 sources. Beginning with the first source (site-a.com/data). First step is to scrape the page data.
    [1] After successfully scraping site-a.com/data (150 records), the next step is to clean the data. Applying standard validation and normalization rules to ensure consistency and quality before storage.
  [Working Memory (2-2)]
    [2] After cleaning the data from site-a.com/data (138 valid records remaining), the next step is to store it in the database. Once stored, proceed to process the second source (site-b.com/data).
   Result: results=[ActionStepResult(tool_id='call_0', tool_name='store_data', tool_arguments={'table': 'collected_data', 'record_count': 138}, tool_result="Stored 138 records in table 'collected_data'", success=True, error=None)]
[18:50:04.446] [Think] (_cognitive_worker.py:332) Total: 625 tokens (cumulative: 2122)
[18:50:04.449] [Think] (_amphibious_automa.py:867) _PromptWorker: finish=False, step=Proceeding to process the second data source (site-b.com/data). The first step is to scrape the page data from this source.
[18:50:04.453] [Act] (_amphibious_automa.py:873) _PromptWorker:
{
    "content": "Proceeding to process the second data source (site-b.com/data). The first step is to scrape the page data from this source.",
    "result": {
        "results": [
            {
                "tool_id": "call_0",
                "tool_name": "scrape_page",
                "tool_arguments": {
                    "url": "site-b.com/data"
                },
                "tool_result": "Scraped data from site-b.com/data: 150 records found, 12 fields per record",
                "success": true,
                "error": null
            }
        ]
    },
    "metadata": {},
    "status": null
}
[18:50:04.453] [Observe] (_amphibious_automa.py:861) _PromptWorker: None
[18:50:18.374] [Think] (_cognitive_worker.py:332) Message 1 (system, 347 tokens):
Process data sources one at a time: scrape the page, clean the data, then store it. Move to the next source after storing. Respond in JSON format.

# Available Tools (with parameters):
• scrape_page: Scrape data from a web page
  - url (string) [required]
• clean_data: Clean and validate scraped data
  - source (string) [required]
  - rules (string) [required]
• store_data: Store processed data in the database
  - table (string) [required]
  - record_count (integer) [required]

# Context Acquiring
If the context contains progressively disclosed information (e.g. skills, history steps) and you want to inspect the details, use the **details** field to request them. The framework will expand these items in the next round. Batch all requests in a single output. When using this field, leave step_content and output empty.

## Field format:
- **details**: [{field: "skills", index: 0}, ...]
  Available fields: **skills** (view a skill's full workflow), **cognitive_history** (view the full result of a previous step)

# Output Fields
- **step_content**: Your analysis and reasoning for this step
- **finish**: Set True when the sub-task is fully complete (default: False)
- **details**: Available fields: **skills**, **cognitive_history**. example: [{field: 'skills', index: 0}, ...]
- **output**: Tool calls to execute: [{tool, tool_arguments: [{name: 'param', value: 'value'}]}]

[18:50:18.374] [Think] (_cognitive_worker.py:332) Message 2 (user, 320 tokens):
Based on the context below, decide your next action.

Goal: Collect data from 2 sources: site-a.com/data and site-b.com/data. For each source: scrape, clean, and store the data.
Execution History:
  (Steps [1-2]: summary only, query details via details)
  [Long-term Pending (0-0)]
    [0] Starting data collection from 2 sources. Beginning with the first source (site-a.com/data). First step is to scrape the page data.
  [Short-term Memory (1-2), query details via 'details']
    [1] After successfully scraping site-a.com/data (150 records), the next step is to clean the data. Applying standard validation and normalization rules to ensure consistency and quality before storage.
    [2] After cleaning the data from site-a.com/data (138 valid records remaining), the next step is to store it in the database. Once stored, proceed to process the second source (site-b.com/data).
  [Working Memory (3-3)]
    [3] Proceeding to process the second data source (site-b.com/data). The first step is to scrape the page data from this source.
   Result: results=[ActionStepResult(tool_id='call_0', tool_name='scrape_page', tool_arguments={'url': 'site-b.com/data'}, tool_result='Scraped data from site-b.com/data: 150 records found, 12 fields per record', success=True, error=None)]
[18:50:18.374] [Think] (_cognitive_worker.py:332) Total: 667 tokens (cumulative: 2789)
[18:50:18.376] [Think] (_amphibious_automa.py:867) _PromptWorker: finish=False, step=After successfully scraping site-b.com/data (150 records), the next step is to clean the data. Applying the same standard validation and normalization rules used for site-a to ensure consistency and quality before storage.
[18:50:18.378] [Act] (_amphibious_automa.py:873) _PromptWorker:
{
    "content": "After successfully scraping site-b.com/data (150 records), the next step is to clean the data. Applying the same standard validation and normalization rules used for site-a to ensure consistency and quality before storage.",
    "result": {
        "results": [
            {
                "tool_id": "call_0",
                "tool_name": "clean_data",
                "tool_arguments": {
                    "source": "site-b.com/data",
                    "rules": "standard validation and normalization rules"
                },
                "tool_result": "Cleaned data from site-b.com/data: removed 12 duplicates, fixed 5 format issues",
                "success": true,
                "error": null
            }
        ]
    },
    "metadata": {},
    "status": null
}
[18:50:18.379] [Observe] (_amphibious_automa.py:861) _PromptWorker: None
[18:50:40.042] [Think] (_cognitive_worker.py:332) Message 1 (system, 347 tokens):
Process data sources one at a time: scrape the page, clean the data, then store it. Move to the next source after storing. Respond in JSON format.

# Available Tools (with parameters):
• scrape_page: Scrape data from a web page
  - url (string) [required]
• clean_data: Clean and validate scraped data
  - source (string) [required]
  - rules (string) [required]
• store_data: Store processed data in the database
  - table (string) [required]
  - record_count (integer) [required]

# Context Acquiring
If the context contains progressively disclosed information (e.g. skills, history steps) and you want to inspect the details, use the **details** field to request them. The framework will expand these items in the next round. Batch all requests in a single output. When using this field, leave step_content and output empty.

## Field format:
- **details**: [{field: "skills", index: 0}, ...]
  Available fields: **skills** (view a skill's full workflow), **cognitive_history** (view the full result of a previous step)

# Output Fields
- **step_content**: Your analysis and reasoning for this step
- **finish**: Set True when the sub-task is fully complete (default: False)
- **details**: Available fields: **skills**, **cognitive_history**. example: [{field: 'skills', index: 0}, ...]
- **output**: Tool calls to execute: [{tool, tool_arguments: [{name: 'param', value: 'value'}]}]

[18:50:40.042] [Think] (_cognitive_worker.py:332) Message 2 (user, 394 tokens):
Based on the context below, decide your next action.

Goal: Collect data from 2 sources: site-a.com/data and site-b.com/data. For each source: scrape, clean, and store the data.
Execution History:
  (Steps [2-3]: summary only, query details via details)
  [Long-term Pending (0-1)]
    [0] Starting data collection from 2 sources. Beginning with the first source (site-a.com/data). First step is to scrape the page data.
    [1] After successfully scraping site-a.com/data (150 records), the next step is to clean the data. Applying standard validation and normalization rules to ensure consistency and quality before storage.
  [Short-term Memory (2-3), query details via 'details']
    [2] After cleaning the data from site-a.com/data (138 valid records remaining), the next step is to store it in the database. Once stored, proceed to process the second source (site-b.com/data).
    [3] Proceeding to process the second data source (site-b.com/data). The first step is to scrape the page data from this source.
  [Working Memory (4-4)]
    [4] After successfully scraping site-b.com/data (150 records), the next step is to clean the data. Applying the same standard validation and normalization rules used for site-a to ensure consistency and quality before storage.
   Result: results=[ActionStepResult(tool_id='call_0', tool_name='clean_data', tool_arguments={'source': 'site-b.com/data', 'rules': 'standard validation and normalization rules'}, tool_result='Cleaned data from site-b.com/data: removed 12 duplicates, fixed 5 format issues', success=True, error=None)]
[18:50:40.042] [Think] (_cognitive_worker.py:332) Total: 741 tokens (cumulative: 3530)
[18:50:40.048] [Think] (_amphibious_automa.py:867) _PromptWorker: finish=True, step=After cleaning site-b.com/data (138 valid records remaining), the next step is to store the processed data in the database. Once stored, the entire data collection task for both sources will be complete.
[18:50:40.051] [Act] (_amphibious_automa.py:873) _PromptWorker:
{
    "content": "After cleaning site-b.com/data (138 valid records remaining), the next step is to store the processed data in the database. Once stored, the entire data collection task for both sources will be complete.",
    "result": {
        "results": [
            {
                "tool_id": "call_0",
                "tool_name": "store_data",
                "tool_arguments": {
                    "table": "collected_data",
                    "record_count": 138
                },
                "tool_result": "Stored 138 records in table 'collected_data'",
                "success": true,
                "error": null
            }
        ]
    },
    "metadata": {},
    "status": null
}
==================================================
  DataCollector-8c3caa9f | Completed
Tokens: 3530 | Time: 96.64s
==================================================
After cleaning site-b.com/data (138 valid records remaining), the next step is to store the processed data in the database. Once stored, the entire data collection task for both sources will be complete.

Inspecting the Memory State¶

After the run, let's inspect what each memory tier contains. The summary() method shows the full memory structure:

In [8]:

  Copied!     
 
history = agent.context.cognitive_history
print(f"Total steps: {len(history)}")
print(f"Working memory size: {history.working_memory_size}")
print(f"Short-term size: {history.short_term_size}")
print(f"Compress threshold: {history.compress_threshold}")
print()
for line in history.summary():
    print(line)
history = agent.context.cognitive_history print(f"Total steps: {len(history)}") print(f"Working memory size: {history.working_memory_size}") print(f"Short-term size: {history.short_term_size}") print(f"Compress threshold: {history.compress_threshold}") print() for line in history.summary(): print(line)

Total steps: 6
Working memory size: 1
Short-term size: 2
Compress threshold: 2

[Long-term Pending (0-2)]
  [0] Starting data collection from 2 sources. Beginning with the first source (site-a.com/data). First step is to scrape the page data.
  [1] After successfully scraping site-a.com/data (150 records), the next step is to clean the data. Applying standard validation and normalization rules to ensure consistency and quality before storage.
  [2] After cleaning the data from site-a.com/data (138 valid records remaining), the next step is to store it in the database. Once stored, proceed to process the second source (site-b.com/data).
[Short-term Memory (3-4), query details via 'details']
  [3] Proceeding to process the second data source (site-b.com/data). The first step is to scrape the page data from this source.
  [4] After successfully scraping site-b.com/data (150 records), the next step is to clean the data. Applying the same standard validation and normalization rules used for site-a to ensure consistency and quality before storage.
[Working Memory (5-5)]
  [5] After cleaning site-b.com/data (138 valid records remaining), the next step is to store the processed data in the database. Once stored, the entire data collection task for both sources will be complete.
   Result: results=[ActionStepResult(tool_id='call_0', tool_name='store_data', tool_arguments={'table': 'collected_data', 'record_count': 138}, tool_result="Stored 138 records in table 'collected_data'", success=True, error=None)]

In the output above you can see the three tiers:

Working Memory — only the most recent step is shown in full detail (the last tool call with all its arguments and results)
Short-term Memory — the 2 steps before that appear as one-line summaries. The agent can still request their full details via the Acquiring policy
Long-term Memory — the earliest steps have been compressed by the LLM into a summary paragraph. Individual step details are no longer accessible

With just 6 steps (scrape + clean + store for each of 2 sources), our small memory configuration pushed the agent through all three tiers.

Acquiring + Cognitive History¶

When the agent needs to recall details from an older step (one that has been moved to Short-term Memory), it uses the Acquiring policy. Here's how it works:

The LLM sees a summary like "Step 3: Scraped data from site-b.com"
It requests details: details: [{field: "cognitive_history", index: 3}]
The framework reveals the full step (tool calls, arguments, results)
The LLM re-thinks with the additional information

This means the agent never permanently loses access to Short-term Memory — it can always "look back" when needed, without keeping everything in the prompt at all times.

Memory Migration Flow¶

As the agent accumulates steps, memory automatically migrates through the tiers:

New step added
    |
    v
[Working Memory] ── latest N steps (full detail)
    |
    v (overflow)
[Short-term Memory] ── next M steps (summaries, details on demand)
    |
    v (overflow, triggers LLM compression)
[Long-term Memory] ── compressed paragraph (no individual step access)

The migration is automatic and transparent:

When a new step is added and Working Memory is full, the oldest working step moves to Short-term Memory (compressed to a summary).
When Short-term Memory exceeds compress_threshold pending steps, the oldest batch is compressed by the LLM into the Long-term Memory paragraph.
The agent keeps running without interruption — it doesn't need to know about the compression happening behind the scenes.

Custom History Configuration¶

You can customize the memory tiers by creating a custom context with your own CognitiveHistory defaults:

In [9]:

  Copied!     
 
from pydantic import Field, ConfigDict


class CustomContext(CognitiveContext):
    model_config = ConfigDict(arbitrary_types_allowed=True)

    # Override the default CognitiveHistory with custom sizes
    cognitive_history: CognitiveHistory = Field(
        default_factory=lambda: CognitiveHistory(
            working_memory_size=8,
            short_term_size=30,
            compress_threshold=15,
        )
    )
from pydantic import Field, ConfigDict class CustomContext(CognitiveContext): model_config = ConfigDict(arbitrary_types_allowed=True) # Override the default CognitiveHistory with custom sizes cognitive_history: CognitiveHistory = Field( default_factory=lambda: CognitiveHistory( working_memory_size=8, short_term_size=30, compress_threshold=15, ) )

What have we learnt?¶

In this tutorial, we explored the framework's automatic memory management:

CognitiveHistory implements a three-tier memory system: Working Memory (full detail), Short-term Memory (summaries with on-demand details), and Long-term Memory (LLM-compressed paragraph).
Memory tiers are configured via working_memory_size, short_term_size, and compress_threshold.
The framework automatically migrates steps from Working → Short-term → Long-term as the agent accumulates history.
Acquiring + Short-term Memory: The LLM can always request full details of any step in Short-term Memory via the Acquiring policy — nothing is permanently lost until Long-term compression.
This system prevents context window overflow in long-running agents while preserving the ability to recall important details.