Skip to main content

Python SDK Guide

Complete guide to using the Hindsight Python SDK for memory operations.

Installation

pip install hindsight-client

Quick Start

from hindsight_client import Hindsight

# Initialize the client
client = Hindsight(
base_url="https://api.hindsight.vectorize.io",
api_key="your-api-key"
)

# Create a memory bank
bank = client.create_bank(
bank_id="my-assistant",
name="My Assistant"
)

# Store a memory
client.retain(
bank_id="my-assistant",
content="The user prefers concise responses and dark mode."
)

# Retrieve memories
result = client.recall(
bank_id="my-assistant",
query="What are the user's preferences?"
)

for memory in result.results:
print(memory.text)

# Get an AI-powered answer
response = client.reflect(
bank_id="my-assistant",
query="How should I format my responses for this user?"
)

print(response.text)

# Close the client when done
client.close()

Client Configuration

Basic Configuration

from hindsight_client import Hindsight

client = Hindsight(
base_url="https://api.hindsight.vectorize.io",
api_key="your-api-key"
)

Advanced Configuration

client = Hindsight(
base_url="https://api.hindsight.vectorize.io",
api_key="your-api-key",
timeout=60.0 # Request timeout in seconds
)

Environment Variables

export HINDSIGHT_API_KEY="your-api-key"
export HINDSIGHT_BASE_URL="https://api.hindsight.vectorize.io"
import os
from hindsight_client import Hindsight

client = Hindsight(
base_url=os.environ["HINDSIGHT_BASE_URL"],
api_key=os.environ["HINDSIGHT_API_KEY"]
)

Memory Banks

Create a Bank

bank = client.create_bank(
bank_id="customer-support-agent",
name="Customer Support Agent",
background="This agent handles customer inquiries for an e-commerce platform",
disposition={
"skepticism": 3,
"literalism": 2,
"empathy": 4
}
)

print(f"Created bank: {bank.bank_id}")

Get Bank Profile

# Bank profile is returned when creating a bank
# or accessible via list_memories

List Memories in a Bank

result = client.list_memories(
bank_id="my-assistant",
limit=100,
offset=0
)

print(f"Total memories: {result.total}")
for memory in result.items:
print(f"- {memory}")

Retain (Store Memories)

Basic Usage

client.retain(
bank_id="my-assistant",
content="User mentioned they work remotely and prefer async communication."
)

With Context and Metadata

client.retain(
bank_id="my-assistant",
content="Customer reported a bug with the checkout process.",
context="Support ticket conversation",
metadata={
"ticket_id": "TKT-12345",
"priority": "high"
}
)

With Timestamp

from datetime import datetime

client.retain(
bank_id="my-assistant",
content="User signed up for premium plan.",
timestamp=datetime.now()
)

Batch Operations

items = [
{"content": "User is based in Pacific timezone"},
{"content": "User prefers email over phone calls"},
{"content": "User has been a customer for 3 years"}
]

client.retain_batch(
bank_id="my-assistant",
items=items
)

Async Batch Operations

# For large batches, use async processing
client.retain_batch(
bank_id="my-assistant",
items=items,
retain_async=True
)

Recall (Search Memories)

result = client.recall(
bank_id="my-assistant",
query="What communication preferences does the user have?"
)

for memory in result.results:
print(f"[{memory.type}] {memory.text}")

With Filters

# Limit results and set search budget
result = client.recall(
bank_id="my-assistant",
query="project deadlines",
max_tokens=4096,
budget="mid" # "low", "mid", or "high"
)

# Filter by memory type
result = client.recall(
bank_id="my-assistant",
query="user preferences",
types=["observation"]
)

Include Entity Information

result = client.recall(
bank_id="my-assistant",
query="Tell me about Alice",
include_entities=True,
max_entity_tokens=1000
)

# Access entities
if result.entities:
for entity in result.entities:
print(f"Entity: {entity}")

Processing Results

result = client.recall(
bank_id="my-assistant",
query="user info"
)

if not result.results:
print("No relevant memories found")
else:
for memory in result.results:
print(f"Type: {memory.type}")
print(f"Content: {memory.text}")
print("---")

Reflect (Reasoning Over Memories)

Basic Query

response = client.reflect(
bank_id="my-assistant",
query="What should I know about this customer before our call?"
)

print(response.text)

With Context

response = client.reflect(
bank_id="my-assistant",
query="What are their main pain points?",
context="We're preparing for a product review meeting"
)

print(response.text)

With Budget Control

response = client.reflect(
bank_id="my-assistant",
query="Summarize our relationship with this customer",
budget="high" # "low", "mid", or "high"
)

print(response.text)

# Access source memories
for source in response.based_on:
print(f"Based on: {source}")

Structured Output

response = client.reflect(
bank_id="my-assistant",
query="Summarize the key points about this customer"
)

# If structured_output was requested
if response.structured_output:
print(response.structured_output)

Mental Models

Mental models are user-curated pre-computed reflections that stay current as new memories are added. They are generated by running a reflect query and can be automatically refreshed after observation consolidation.

Create a Mental Model

# Creation runs asynchronously via reflect
result = client.create_mental_model(
bank_id="my-assistant",
name="User Profile",
source_query="What do we know about this user's preferences and background?"
)

print(f"Operation ID: {result.operation_id}")

Create with Options

result = client.create_mental_model(
bank_id="my-assistant",
name="Team Directory",
source_query="Who works here and what do they do?",
tags=["team", "directory"],
max_tokens=4096,
trigger={"refresh_after_consolidation": True}
)

List Mental Models

models = client.list_mental_models(bank_id="my-assistant")

for model in models.items:
print(f"{model.name}: {model.content[:100]}...")

# Filter by tags
models = client.list_mental_models(bank_id="my-assistant", tags=["team"])

Get a Mental Model

model = client.get_mental_model(
bank_id="my-assistant",
mental_model_id="mm_abc123"
)

print(f"Name: {model.name}")
print(f"Content: {model.content}")
print(f"Last refreshed: {model.last_refreshed_at}")

Refresh a Mental Model

# Re-run the source query to update the content
result = client.refresh_mental_model(
bank_id="my-assistant",
mental_model_id="mm_abc123"
)

print(f"Refresh operation: {result.operation_id}")

Update a Mental Model

model = client.update_mental_model(
bank_id="my-assistant",
mental_model_id="mm_abc123",
name="Updated Profile",
source_query="What are the user's key preferences?",
trigger={"refresh_after_consolidation": True}
)

Delete a Mental Model

client.delete_mental_model(
bank_id="my-assistant",
mental_model_id="mm_abc123"
)

Async Support

The SDK supports async/await for all operations:

import asyncio
from hindsight_client import Hindsight

async def main():
client = Hindsight(
base_url="https://api.hindsight.vectorize.io",
api_key="your-api-key"
)

# Async operations
await client.aretain(
bank_id="my-assistant",
content="Async memory storage"
)

result = await client.arecall(
bank_id="my-assistant",
query="async test"
)

for memory in result.results:
print(memory.text)

response = await client.areflect(
bank_id="my-assistant",
query="What do you know?"
)

print(response.text)

await client.aclose()

asyncio.run(main())

Concurrent Operations

import asyncio
from hindsight_client import Hindsight

async def store_multiple(client, bank_id, contents):
tasks = [
client.aretain(bank_id=bank_id, content=content)
for content in contents
]
return await asyncio.gather(*tasks)

async def main():
client = Hindsight(
base_url="https://api.hindsight.vectorize.io",
api_key="your-api-key"
)

contents = [
"Memory 1",
"Memory 2",
"Memory 3"
]

await store_multiple(client, "my-assistant", contents)
print(f"Stored {len(contents)} memories")

await client.aclose()

asyncio.run(main())

Error Handling

from hindsight_client import Hindsight

client = Hindsight(
base_url="https://api.hindsight.vectorize.io",
api_key="your-api-key"
)

try:
result = client.recall(
bank_id="invalid-bank",
query="test"
)
except Exception as e:
# Check error response
print(f"Error: {e}")

Common HTTP Errors

StatusCauseSolution
401Invalid API keyCheck your API key
402Insufficient creditsAdd credits to your account
404Invalid bank_idVerify the bank exists
400Invalid requestCheck request parameters

Best Practices

Connection Reuse

Reuse the client instance across your application:

# Good: Single client instance
client = Hindsight(
base_url="https://api.hindsight.vectorize.io",
api_key="your-api-key"
)

def process_user(bank_id, message):
client.retain(bank_id=bank_id, content=message)
return client.recall(bank_id=bank_id, query=message)

Context Managers

Use context managers for automatic cleanup:

from hindsight_client import Hindsight

# The client should be closed when done
client = Hindsight(
base_url="https://api.hindsight.vectorize.io",
api_key="your-api-key"
)

try:
result = client.recall(bank_id="my-assistant", query="test")
finally:
client.close()

Logging

Enable logging for debugging:

import logging

logging.basicConfig(level=logging.DEBUG)

client = Hindsight(
base_url="https://api.hindsight.vectorize.io",
api_key="your-api-key"
)
# Now you'll see request/response logs

Complete Example

from hindsight_client import Hindsight

def main():
# Initialize client
client = Hindsight(
base_url="https://api.hindsight.vectorize.io",
api_key="your-api-key"
)

try:
bank_id = "demo-assistant"

# Create a memory bank
bank = client.create_bank(
bank_id=bank_id,
name="Demo Assistant",
disposition={
"skepticism": 3,
"literalism": 2,
"empathy": 4
}
)
print(f"Using bank: {bank.bank_id}")

# Store some memories
memories_to_store = [
"User's name is Alice and she is a product manager",
"Alice prefers bullet points over long paragraphs",
"Alice works at TechCorp and manages the mobile team",
"Alice mentioned she's interested in AI automation"
]

for content in memories_to_store:
client.retain(bank_id=bank_id, content=content)
print(f"Stored: {content[:50]}...")

# Search for relevant memories
result = client.recall(
bank_id=bank_id,
query="What does Alice do for work?"
)

print("\nRelevant memories:")
for memory in result.results:
print(f" - {memory.text}")

# Get an AI-synthesized answer
response = client.reflect(
bank_id=bank_id,
query="Write a brief introduction for Alice"
)

print(f"\nAI Response:\n{response.text}")

except Exception as e:
print(f"Error: {e}")

finally:
client.close()

if __name__ == "__main__":
main()