Cortex API

Cortex is Teleon’s memory system for AI agents.

Role

Cortex provides persistent, searchable memory you can use to store and retrieve information across runs. It is designed to be simple (6 core methods) while supporting production needs like scope enforcement (multi-tenancy), memory layers, and auto-context injection.

When to use

Add user / customer memory that should persist across agent executions.
Retrieve context with semantic search (with optional exact-match filters).
Enforce isolation between tenants using a mandatory scope (for example, customer_id).
Use auto-save + auto-context injection to reduce boilerplate.

Quick start

from teleon import TeleonClient
 
client = TeleonClient(api_key="tlk_live_xxx")
 
@client.agent(name="support", cortex=True)
async def support_agent(query: str, customer_id: str, cortex):
    await cortex.store(content=f"Customer asked: {query}", customer_id=customer_id)
    results = await cortex.search(query="previous issues", filter={"customer_id": customer_id})
    return "ok"

Configuration

Simple enable

@client.agent(name="agent", cortex=True)

This enables:

Auto-save conversations after each execution
Auto-context injection before execution
In-memory storage (development mode)

Full configuration

@client.agent(
    name="support",
    cortex={
        "auto": True,
        "fields": ["query", "type"],
        "scope": ["customer_id"],
        "name": "shared-memory",
        "layers": {
            "team": {"scope": ["team_id"]},
            "personal": {"scope": ["user_id"]},
        },
        "auto_context": {
            "enabled": True,
            "history_limit": 10,
            "relevant_limit": 5,
            "max_tokens": 2000,
            "filter": {},
        },
    },
)
async def support_agent(query: str, customer_id: str, cortex):
    ...

Configuration options

Option	Type	Default	Description
`auto`	bool	`True`	Auto-save conversations after execution
`fields`	list	`None`	Fields to capture (None = all args)
`scope`	list	`[]`	Mandatory isolation fields for multi-tenancy
`name`	str	agent name	Memory namespace (for sharing)
`layers`	dict	`{}`	Custom memory layers
`auto_context`	dict	see below	Auto-context configuration

API reference

`store()`

Save content with any fields.

entry_id = await cortex.store(
    content="User prefers dark mode",
    ttl=3600,
    upsert=True,
    customer_id="alice",
    type="preference",
    tags=["ui", "settings"],
)

Parameters:

content (str): Text content to store
ttl (int, optional): Time-to-live in seconds
upsert (bool, optional): Update existing entry if fields match
**fields: Any additional fields to store

Returns: Entry ID (str)

`search()`

Semantic search with optional filter.

results = await cortex.search(
    query="billing issues",
    filter={"customer_id": "alice"},
    limit=10,
)
 
for entry in results:
    print(f"Score: {entry.score}")
    print(f"Content: {entry.content}")
    print(f"Fields: {entry.fields}")

Parameters:

query (str, optional): Search query for semantic matching
filter (dict, optional): Field filters (exact match)
limit (int): Maximum results (default: 10)

Returns: List of Entry objects with relevance scores

`get()`

Get entries by filter (faster than search when you don’t need semantic matching).

history = await cortex.get(
    filter={"customer_id": "alice", "type": "query"},
    limit=50,
)

Parameters:

filter (dict): Field filters (exact match)
limit (int): Maximum results (default: 10)

Returns: List of Entry objects ordered by created_at DESC

`update()`

Update entries matching a filter.

count = await cortex.update(
    filter={"customer_id": "alice", "type": "preference"},
    content="Now prefers light mode",
)
 
count = await cortex.update(
    filter={"lead_id": "123"},
    status="qualified",
    score=85,
)

Parameters:

filter (dict): Filter to match entries
content (str, optional): New content
**fields: Fields to update (merged with existing)

Returns: Number of entries updated (int)

`delete()`

Delete entries matching a filter.

deleted = await cortex.delete(filter={"user_id": "alice"})
print(f"Deleted {deleted} entries")

Parameters:

filter (dict): Filter to match entries

Returns: Number of entries deleted (int)

`count()`

Count entries matching a filter.

total = await cortex.count()
queries = await cortex.count(filter={"type": "query"})

Parameters:

filter (dict, optional): Filter to match (counts all if None)

Returns: Number of matching entries (int)

Scope enforcement (multi-tenancy)

Scope ensures data isolation between tenants. When you define scope fields, they are automatically enforced on all operations.

@client.agent(
    name="support",
    cortex={
        "scope": ["customer_id"],
    },
)
async def support_agent(query: str, customer_id: str, cortex):
    await cortex.store(content="Note", type="note")
    results = await cortex.search(query="issues")

How it works:

Scope fields are extracted from function arguments
Every store() automatically includes scope values
Every search(), get(), update(), delete(), and count() is filtered by scope
Users cannot override scope values

Memory layers

Layers provide hierarchical memory organization. Each layer has its own scope.

@client.agent(
    name="support",
    cortex={
        "scope": ["team_id", "user_id"],
        "layers": {
            "company": {"scope": []},
            "team": {"scope": ["team_id"]},
            "personal": {"scope": ["user_id"]},
        },
    },
)
async def support_agent(query: str, team_id: str, user_id: str, cortex):
    await cortex.company.store(content="Company policy update")
    await cortex.team.store(content="Team standup notes")
    await cortex.personal.store(content="My personal reminder")

Auto-context injection

Cortex automatically retrieves relevant context before agent execution and makes it available via cortex.context.

@client.agent(
    name="support",
    cortex={
        "auto_context": {
            "enabled": True,
            "history_limit": 10,
            "relevant_limit": 5,
            "max_tokens": 2000,
            "filter": {"type": "important"},
        }
    },
)
async def support_agent(query: str, cortex):
    if cortex.context:
        context_text = cortex.context.text

Context properties:

cortex.context.entries
cortex.context.text
len(cortex.context)
bool(cortex.context)

Storage backends

Development (default)

In-memory storage for development and testing.

@client.agent(name="agent", cortex=True)

PostgreSQL with pgvector

from teleon.cortex import set_storage_backend, PostgresBackend
 
backend = PostgresBackend(
    host="localhost",
    port=5432,
    database="teleon",
    user="postgres",
    password="secret",
)
set_storage_backend(backend)

Requirements:

PostgreSQL with pgvector extension
pip install asyncpg

Redis with RediSearch

from teleon.cortex import set_storage_backend, RedisBackend
 
backend = RedisBackend(
    host="localhost",
    port=6379,
    password="secret",
)
set_storage_backend(backend)

Requirements:

Redis Stack (includes RediSearch)
pip install redis[asyncio]

Examples

Customer support agent

@client.agent(
    name="support",
    cortex={
        "auto": True,
        "scope": ["customer_id"],
        "auto_context": {
            "enabled": True,
            "history_limit": 20,
            "relevant_limit": 5,
        },
    },
)
async def support_agent(query: str, customer_id: str, cortex):
    if cortex.context:
        context = f"\nPrevious interactions:\n{cortex.context.text}"
    else:
        context = ""
 
    response = await llm.complete(
        f"Customer query: {query}{context}\n\nProvide helpful response:"
    )
 
    if "resolved" in response.lower():
        await cortex.store(
            content=f"Resolved: {query} -> {response}",
            type="resolution",
            topic=extract_topic(query),
        )
 
    return response

Sales agent with lead scoring

@client.agent(
    name="sales",
    cortex={
        "scope": ["lead_id"],
        "fields": ["query", "intent"],
    },
)
async def sales_agent(query: str, lead_id: str, cortex):
    history = await cortex.get(filter={}, limit=50)
    interaction_count = await cortex.count()
 
    if interaction_count > 5:
        await cortex.update(
            filter={"type": "lead_profile"},
            status="qualified",
            score=min(100, interaction_count * 10),
        )
 
    similar = await cortex.search(
        query=query,
        filter={"type": "successful_deal"},
    )

Guide: /docs/guides/cortex-memory
Tutorial: /docs/tutorials/customer-support-agent

Cortex API

Role

When to use

Quick start

Configuration

Simple enable

Full configuration

Configuration options

API reference

`store()`

`search()`

`get()`

`update()`

`delete()`

`count()`

Scope enforcement (multi-tenancy)

Memory layers

Auto-context injection

Storage backends

Development (default)

PostgreSQL with pgvector

Redis with RediSearch

Examples

Customer support agent

Sales agent with lead scoring

Next

On this page