Python SDK 指南

使用 Hindsight Python SDK 进行记忆操作的完整指南。

安装

pip

pip install hindsight-client

Poetry

poetry add hindsight-client

Pipenv

pipenv install hindsight-client

快速开始

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()

客户端配置

基本配置

from hindsight_client import Hindsight

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

高级配置

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

环境变量

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"]
)

记忆库

创建记忆库

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}")

获取记忆库资料

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

列出记忆库中的记忆

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（存储记忆）

基本用法

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

携带上下文和元数据

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"
    }
)

携带时间戳

from datetime import datetime

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

批量操作

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
)

异步批量操作

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

Recall（搜索记忆）

基础搜索

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}")

带过滤器

# 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"]
)

包含实体信息

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}")

处理结果

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（在记忆之上推理）

基本查询

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

print(response.text)

携带上下文

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)

控制预算

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}")

结构化输出

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)

心智模型

心智模型是用户精心整理的、预计算的反思，会随着新记忆的添加保持最新。它们通过运行 reflect 查询生成，并可在观察整合后自动刷新。

创建心智模型

# 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}")

带选项创建

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}
)

列出心智模型

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"])

获取心智模型

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}")

刷新心智模型

# 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}")

更新心智模型

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}
)

删除心智模型

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

异步支持

SDK 支持对所有操作使用 async/await：

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())

并发操作

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())

错误处理

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}")

常见 HTTP 错误

状态码	原因	解决方法
401	API 密钥无效	检查 API 密钥
402	额度不足	为账户添加额度
404	bank_id 无效	确认记忆库存在
400	请求无效	检查请求参数

最佳实践

复用连接

在你的应用中复用客户端实例：

# 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)

上下文管理器

使用上下文管理器实现自动清理：

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()

日志

启用日志以便调试：

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

完整示例

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()