Python SDK Guide
Complete guide to using the Hindsight Python SDK for memory operations.
Installation
- pip
- Poetry
- Pipenv
pip install hindsight-client
poetry add hindsight-client
pipenv 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)
Basic Search
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
| Status | Cause | Solution |
|---|---|---|
| 401 | Invalid API key | Check your API key |
| 402 | Insufficient credits | Add credits to your account |
| 404 | Invalid bank_id | Verify the bank exists |
| 400 | Invalid request | Check 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()