빠른 참조 코드 샘플

이 문서에서는 일반적인 에이전트 메모리 설정 및 API 수명 주기 작업에 대한 작고 집중적인 예를 수집합니다.

LLM/임베딩 설정

다음 예에서는 LLM 및 임베딩 모델 모두에 LiteLLM을 사용합니다.

LLM 구성

from oracleagentmemory.core.llms.llm import Llm

llm = Llm(
    model="YOUR_LLM_MODEL",
    api_base="YOUR_LLM_API_BASE",
    api_key="YOUR_LLM_API_KEY",
)

response = llm.generate("What is 2+2?")
print(response.text)

출력:

2+2 is equal to 4

포함 모델 구성

from oracleagentmemory.core.embedders.embedder import Embedder

embedder = Embedder(
    model="YOUR_EMBEDDING_MODEL",
    api_base="YOUR_EMBEDDING_API_BASE",
    api_key="YOUR_EMBEDDING_API_KEY",
)

embedding_matrix = embedder.embed(["The quick brown fox jumps over the lazy dog"])
print(embedding_matrix.shape)

출력:

(1, embedding_dimension)

API 설정

에이전트 메모리 구성 요소 구성

이 기능은 내장 모델 및 자동 메모리 추출을 위한 선택적 LLM과 함께 Oracle DB 연결 또는 풀을 사용합니다.

import oracledb

from oracleagentmemory.core.oracleagentmemory import OracleAgentMemory

db_pool = oracledb.SessionPool(
    user="YOUR DB USER",
    password="YOUR DB PASSWORD",
    dsn="YOUR DB CONNECT STRING",
)
memory = OracleAgentMemory(
    connection=db_pool,
    embedder=embedder,
    llm=llm,  # optional: enables automatic memory extraction during add_messages()
)

Oracle DB 메모리 구성 요소 구성

이 변형은 Oracle DB 접속 또는 풀을 사용하며 스키마 정책 및 테이블 이름 접두어를 설정하는 방법을 보여줍니다.

import oracledb

from oracleagentmemory.core import SchemaPolicy
from oracleagentmemory.core.oracleagentmemory import OracleAgentMemory

db_pool = oracledb.SessionPool(
    user="YOUR DB USER",
    password="YOUR DB PASSWORD",
    dsn="YOUR DB CONNECT STRING",
)
db_memory = OracleAgentMemory(
    connection=db_pool,
    embedder=embedder,
    llm=llm,  # optional
    schema_policy=SchemaPolicy.CREATE_IF_NECESSARY,
    memory_store_id="DEV_",
)

Oracle Hybrid DB 메모리 구성 요소 구성

이 변형은 저장된 검색 텍스트보다 Oracle 관리 하이브리드 검색을 사용으로 설정하고 관리되는 검색 인덱스 동기화 모드를 선택하는 방법을 보여줍니다.

SearchStrategy.HYBRID는 Oracle의 관리형 하이브리드 벡터 인덱스를 생성 또는 검증하며, 관리형 인덱스가 임베더의 데이터베이스 내 모델을 사용하도록 기본 임베더가 OracleDBEmbedder이어야 합니다. SearchStrategy.KEYWORD는 텍스트 전용입니다. 저장된 검색 텍스트로 순위가 지정되며 임베더가 필요하지 않습니다. 키워드 스키마는 로컬 벡터 저장 영역 없이 생성할 수 있으므로, 스키마를 다시 생성하거나 임베딩을 먼저 백필하지 않는 한 SearchStrategy.VECTOR를 사용하여 키워드 스키마를 다시 열지 마십시오. Oracle의 관리형 하이브리드 인덱스는 저장된 검색 텍스트를 기반으로 구축되기 때문에 여전히 OracleDBEmbedder를 사용하여 하이브리드 검색으로 업그레이드할 수 있습니다.

경고: 기존 데이터에 대해 하이브리드 인덱스가 처음 생성되면 Oracle은 저장된 검색 텍스트를 스캔하고 스키마 설정 중 관리되는 인덱스 상태를 구축합니다. SchemaPolicy.CREATE_IF_NECESSARY는 시간이 걸릴 수 있으며 대규모 스키마에 대한 데이터베이스 마이그레이션처럼 계획해야 합니다.

from oracleagentmemory.core import SchemaPolicy, SearchIndexSyncMode, SearchStrategy
from oracleagentmemory.core.embedders import OracleDBEmbedder
from oracleagentmemory.core.oracleagentmemory import OracleAgentMemory

db_embedder = OracleDBEmbedder(
    connection=db_pool,
    model="YOUR_DB_EMBEDDING_MODEL",
    embedding_dimension=384,
)

hybrid_db_memory = OracleAgentMemory(
    connection=db_pool,
    embedder=db_embedder,
    llm=llm,  # optional
    schema_policy=SchemaPolicy.CREATE_IF_NECESSARY,
    search_strategy=SearchStrategy.HYBRID,
    search_index_sync=SearchIndexSyncMode.ON_COMMIT,
)

API 수명 주기

글타래 만들기

선택적 스레드 ID, 유저 ID 및 에이전트 ID를 사용하여 스레드를 생성합니다.

thread = memory.create_thread(
    thread_id="thread_create_123",  # optional
    user_id="user_123",             # optional
    agent_id="agent_456",           # optional
)

print(thread.thread_id)

출력:

thread_create_123

기존 스레드 다시 열기

thread = memory.create_thread(
    thread_id="thread_reopen_123",
    user_id="user_123",
    agent_id="agent_456",
)

same_thread = memory.get_thread("thread_reopen_123")
print(same_thread.thread_id)

출력:

thread_reopen_123

기존 스레드 업데이트

update_thread()를 사용하여 스레드 메타데이터 또는 영구 런타임 구성 변경사항을 지속합니다. get_thread()에 전달된 재정의는 명시적으로 지속될 때까지 재개설된 핸들에만 영향을 줍니다.

thread = memory.create_thread(
    thread_id="thread_update_123",
    user_id="user_123",
    agent_id="agent_456",
)

loaded_thread = memory.get_thread(
    "thread_update_123",
    max_message_token_length=8_000,
)
print(loaded_thread.max_message_token_length)

updated_thread = memory.update_thread(
    "thread_update_123",
    metadata={"source": "support", "flags": {"vip": True}},
    max_message_token_length=8_000,
)
persisted_thread = memory.get_thread("thread_update_123")

print(updated_thread.metadata["flags"]["vip"])
print(persisted_thread.max_message_token_length)

출력:

8000
True
8000

get_thread()에 전달된 재정의는 일시적입니다. 스레드 메타데이터 또는 영구 런타임 구성 변경사항을 지속하려면 update_thread()를 호출합니다.

스레드 삭제

스레드 범위의 계단식 정리가 필요한 경우 이 작업을 사용합니다. SDK에서 관리하는 관련 메시지, 내구성 있는 메모리 및 지원 검색 데이터와 함께 스레드를 제거합니다.

thread = memory.create_thread(thread_id="thread_delete_123")

deleted = memory.delete_thread("thread_delete_123")
print(deleted)

출력:

1

스레드 범위의 계단식 정리가 필요한 경우 스레드 삭제를 사용합니다. 스레드를 해당 메시지, 메모리 및 SDK에서 관리하는 지원 검색 데이터와 함께 제거합니다.

사용자 프로파일 추가

user_profile_id = memory.add_user(
    "user_123",
    "The user prefers concise answers and works mostly with Python.",
)

print(user_profile_id)

출력:

user_123

에이전트 프로파일 추가

agent_profile_id = memory.add_agent(
    "agent_456",
    "A coding assistant specialized in debugging and code review.",
)

print(agent_profile_id)

출력:

agent_456

메모리 API에서 전역 메모리 추가

thread_id를 생략할 경우 메모리는 특정 스레드에 연결되지 않습니다. 반환된 값은 메모리 식별자입니다.

memory_id = memory.add_memory(
    "The user prefers short, bullet-point answers.",
    memory_type="preference",
    user_id="user_123",
    agent_id="agent_456",
)

print(memory_id)

출력:

mem:1

메모리 API에서 범위가 지정된 메모리 추가

반환된 값은 메모리 식별자입니다.

thread = memory.create_thread(
    thread_id="thread_scoped_123",
    user_id="user_123",
    agent_id="agent_456",
)

memory_id = memory.add_memory(
    "The user is planning a trip to Kyoto next month.",
    memory_type="fact",
    user_id="user_123",
    agent_id="agent_456",
    thread_id=thread.thread_id,
)

print(memory_id)

출력:

mem:2

메모리 API에서 메모리 갱신

update_memory()를 사용하여 기존 메모리 유사 레코드의 저장된 콘텐츠 또는 메타데이터를 식별자별로 바꿉니다.

thread = memory.create_thread(
    thread_id="thread_update_memory_api_123",
    user_id="user_123",
    agent_id="agent_456",
)
memory_id = memory.add_memory(
    "The user likes short status updates.",
    user_id=thread.user_id,
    agent_id=thread.agent_id,
    thread_id=thread.thread_id,
    metadata={"source": "chat"},
)

updated_memory_id = memory.update_memory(
    memory_id,
    content="The user prefers short status updates.",
    metadata={"source": "support"},
)

print(updated_memory_id)

출력:

mem:3

사용자 ID로 메모리 추가

반환된 값은 호출자가 제공한 메모리 식별자입니다.

memory_id = memory.add_memory(
    "The user prefers aisle seats on flights.",
    user_id="user_123",
    agent_id="agent_456",
    memory_id="travel_pref_001",
)

print(memory_id)

출력:

travel_pref_001

스레드 기본사항

스레드에 메시지 추가

메시지는 딕셔너리 또는 Message 객체로 전달될 수 있습니다. 선택적 메시지 ID, 시간 기록 및 메타데이터를 함께 저장할 수 있습니다.

from oracleagentmemory.apis import Message

thread = memory.create_thread(
    thread_id="thread_messages_123",
    user_id="user_123",
    agent_id="agent_456",
)

message_ids = thread.add_messages(
    [
        Message(
            id="msg_user_001",
            role="user",
            content="I prefer window seats on flights.",
            timestamp="2026-03-27T09:00:00Z",
            metadata={"source": "chat", "channel": "web"},
        ),
        {
            "id": "msg_assistant_001",
            "role": "assistant",
            "content": "Noted. I will keep that in mind.",
            "timestamp": "2026-03-27T09:00:05Z",
            "metadata": {"source": "assistant"},
        },
    ]
)

print(message_ids)

출력:

['msg_user_001', 'msg_assistant_001']

다시 읽기 스레드 메시지

저장된 모든 메시지 또는 슬라이스는 start 및 end를 사용하여 읽을 수 있습니다.

thread = memory.create_thread(thread_id="thread_read_messages_123")
thread.add_messages(
    [
        {"role": "user", "content": "Message 1"},
        {"role": "assistant", "content": "Message 2"},
        {"role": "user", "content": "Message 3"},
    ]
)

default_messages = thread.get_messages()
all_messages = thread.get_messages(end=None)
middle_messages = thread.get_messages(start=1, end=3)

print([message.content for message in default_messages])

print([message.content for message in all_messages])

print([message.content for message in middle_messages])

출력:

짧은 스레드에서 바인딩된 기본값은 여전히 모든 메시지를 반환합니다.

['Message 1', 'Message 2', 'Message 3']

['Message 1', 'Message 2', 'Message 3']

['Message 2', 'Message 3']

ID별 현재 스레드에서 메시지 삭제

메시지를 삭제하면 현재 스레드에서 원시 메시지 행만 제거됩니다. 해당 메시지에서 생성된 파생 메모리 또는 기타 다운스트림 아티팩트는 검색 가능한 상태로 유지될 수 있으며 여전히 컨텍스트 카드 출력에 영향을 줄 수 있습니다. 스레드와 연관된 메시지 및 메모리를 함께 삭제해야 하는 경우 대신 delete_thread()를 사용합니다. 다른 스레드에서 식별자를 전달해도 0가 반환됩니다.

thread = memory.create_thread(thread_id="thread_delete_message_123")
message_ids = thread.add_messages(
    [
        {"role": "user", "content": "Message to delete"},
    ]
)

deleted = thread.delete_message(message_ids[0])
print(deleted)

출력:

1

그러면 현재 스레드에서 원시 메시지 행만 제거됩니다. 해당 메시지에서 생성된 파생 메모리 또는 기타 다운스트림 아티팩트는 자동으로 삭제되지 않으며 검색 가능한 상태로 유지되거나 컨텍스트 카드 출력에 나타날 수 있습니다. memory.delete_thread(thread.thread_id)를 사용하여 연관된 메시지 및 메모리와 함께 스레드를 삭제합니다. 메시지가 다른 스레드가 소유한 ID에 대한 반환 0를 삭제합니다.

ID별로 현재 스레드에서 메시지 업데이트

스레드 범위의 메시지 업데이트는 현재 스레드가 소유한 원시 메시지에만 영향을 줍니다. 저장된 역할 및 시간 기록 값은 보존되며, 자동 추출이 사용으로 설정된 경우 메시지 콘텐츠를 편집하면 add_messages()와 동일한 내역 창 규칙을 사용하여 편집된 메시지에 대한 추출이 즉시 재실행됩니다. 이전 스레드 내역만 지원 컨텍스트로 사용할 수 있습니다. 이후 메시지는 즉시 전달 중에 무시되며, 편집된 내용에서 새 메모리를 추가하는 동안 동일한 새로 고침을 통해 기존의 파생된 메모리가 제자리에 유지됩니다.

thread = memory.create_thread(thread_id="thread_update_message_123")
thread.add_messages(
    [
        {
            "id": "msg_update_001",
            "role": "user",
            "content": "Original message text.",
            "timestamp": "2026-03-27T10:00:00Z",
            "metadata": {"source": "chat"},
        }
    ]
)

updated_message_id = thread.update_message(
    "msg_update_001",
    content="Edited message text.",
    metadata={"source": "support"},
)
print(updated_message_id)

출력:

msg_update_001

메시지 업데이트는 저장된 역할 및 시간 기록 값을 보존합니다. 자동 추출이 사용으로 설정된 경우 콘텐츠 편집은 add_messages()와 동일한 내역 창 규칙을 사용하여 편집된 메시지에 대한 추출을 즉시 재실행합니다. 이후 메시지는 즉시 전달 중에 무시됩니다. 기존 파생 메모리는 그대로 유지되며 새로 고치는 동안 편집된 컨텐츠 메모리가 새로 추가됩니다.

스레드 핸들에서 메모리 추가

반환된 값은 메모리 식별자입니다.

thread = memory.create_thread(
    thread_id="thread_add_memory_123",
    user_id="user_123",
    agent_id="agent_456",
)

memory_id = thread.add_memory(
    "Use pytest for this repository's test suite.",
    memory_type="guideline",
)
print(memory_id)

출력:

mem:4

ID별로 현재 스레드에서 메모리 업데이트

스레드 범위 업데이트는 현재 스레드가 소유한 메모리와 유사한 레코드에만 영향을 줍니다. 다른 스레드에서 식별자를 전달하면 KeyError가 증가합니다.

thread = memory.create_thread(
    thread_id="thread_update_memory_123",
    user_id="user_123",
    agent_id="agent_456",
)
memory_id = thread.add_memory(
    "The user likes jasmine tea.",
    metadata={"source": "survey"},
)

updated_memory_id = thread.update_memory(
    memory_id,
    content="The user likes jasmine tea in the afternoon.",
    metadata={"source": "support"},
)
print(updated_memory_id)

출력:

mem:5

스레드 업데이트는 현재 스레드에 범위가 지정되고 다른 스레드가 소유한 누락된 ID 또는 ID에 대해 KeyError를 인상합니다.

ID별 현재 스레드에서 메모리 삭제

스레드 삭제는 현재 스레드에 범위가 지정됩니다. 다른 스레드에서 식별자를 전달하면 0가 반환됩니다.

thread = memory.create_thread(thread_id="thread_delete_memory_123")
memory_id = thread.add_memory("Temporary memory to delete.")

deleted = thread.delete_memory(memory_id)
print(deleted)

출력:

1

스레드 삭제는 현재 스레드에 범위가 지정되고 다른 스레드가 소유한 ID에 대해 0을 반환합니다.

컨텍스트 카드 작성

thread = memory.create_thread(thread_id="thread_context_card_123")
thread.add_messages(
    [
        {"role": "user", "content": "I am planning a trip to Kyoto next spring."},
    ]
)
thread.add_memory("The user is planning a trip to Kyoto.")

context_card = thread.get_context_card()
print(context_card.content)

출력:

<context_card>
    The user is planning a trip to Kyoto.
</context_card>

스레드 요약 작성

thread = memory.create_thread(thread_id="thread_summary_123")
thread.add_messages(
    [
        {"role": "user", "content": "Hello"},
        {"role": "assistant", "content": "Hi, how can I help?"},
        {"role": "user", "content": "Please summarize this thread."},
    ]
)

summary = thread.get_summary()
print(summary.content)

출력:

user (-): Hello
- assistant (-): Hi, how can I help?
- user (-): Please summarize this thread.

마지막 N개 메시지를 제외한 요약 작성

thread = memory.create_thread(thread_id="thread_summary_except_last_123")
thread.add_messages(
    [
        {"role": "user", "content": "First message"},
        {"role": "assistant", "content": "Second message"},
        {"role": "user", "content": "Third message"},
    ]
)

summary = thread.get_summary(except_last=1)
print(summary.content)

출력:

user (-): First message
- assistant (-): Second message

토큰 예산으로 요약 작성

thread = memory.create_thread(thread_id="thread_summary_budget_123")
thread.add_messages(
    [
        {"role": "user", "content": "Message 1"},
        {"role": "assistant", "content": "Message 2"},
        {"role": "user", "content": "Message 3"},
        {"role": "assistant", "content": "Message 4"},
    ]
)

summary = thread.get_summary(token_budget=20)
print(summary.content)

출력:

(truncated)
user (-): Message 1
...

검색

명시적 범위 지정이 없는 스레드에서 검색

스레드 레벨 검색은 명시적 범위를 전달하지 않을 때 스레드 기본값을 사용합니다.

thread = memory.create_thread(
    thread_id="thread_search_default_123",
    user_id="user_123",
    agent_id="agent_456",
)
thread.add_memory("The user likes pizza.")
thread.add_memory("The user likes cats.")

results = thread.search("pizza", max_results=5)
print([result.content for result in results])

출력:

['The user likes pizza.']

범위 지정을 사용하여 메모리 API에서 검색

API 레벨에서 user_id, agent_id 및 thread_id를 사용하여 SearchScope까지 검색 범위를 지정할 수 있습니다. 최상위 레벨 클라이언트 검색의 경우 명시적 사용자 범위를 제공합니다. 범위가 지정되지 않은 레코드를 의도적으로 원하는 경우에만 user_id=None를 사용합니다. 생략된 값, 명시적 None 및 정확한 일치 플래그가 각 API 계층에서 분석되는 방법에 대한 요약은 범위 해결을 참조하십시오.

from oracleagentmemory.apis.searchscope import SearchScope

thread = memory.create_thread(
    thread_id="thread_memory_search_123",
    user_id="user_123",
    agent_id="agent_456",
)
thread.add_memory("The user likes hiking in the Alps.")

results = memory.search(
    "hiking",
    scope=SearchScope(
        user_id="user_123",
        agent_id="agent_456",
        thread_id="thread_memory_search_123",
        exact_thread_match=True,
    ),
    max_results=5,
)

print([result.content for result in results])

출력:

['The user likes hiking in the Alps.']

메타데이터 필터링으로 검색

저장된 메타데이터에 요청된 부분 매핑이 포함된 레코드만 검색에서 고려해야 하는 경우 metadata_filter를 사용합니다. 다중 필터 키는 AND 의미와 결합되고, 중첩된 딕셔너리는 중첩된 메타데이터 필드와 일치하며, 베어 목록 값은 정확히 일치해야 합니다. 배열 멤버쉽을 테스트하려면 필드 레벨 연산자 사전(예: {"tags": {"$array_contains": "outdoor"}})을 사용합니다. 목록이 있는 "$array_contains"에는 나열된 값이 모두 필요하고, "$array_contains_any"에는 나열된 값이 하나 이상 필요하며, "$not"는 연산자 사전 또는 원시 정확한 일치 값을 포함하여 동일한 필드에서 다른 필드 레벨 표현식을 부정합니다. 누락된 필드를 포함하여 양수 표현식이 실패하는 경우에도 부정 표현식이 일치합니다. 음수 배열 멤버쉽도 배열이 아닌 필드와 일치합니다.

from oracleagentmemory.apis.searchscope import SearchScope

thread = memory.create_thread(
    thread_id="thread_metadata_filter_123",
    user_id="user_123",
    agent_id="agent_456",
)
thread.add_memory(
    "The user likes alpine hiking.",
    metadata={"source": "survey", "category": {"kind": "travel"}, "tags": ["outdoor"]},
)
thread.add_memory(
    "The user likes indoor climbing.",
    metadata={"source": "chat", "category": {"kind": "sports"}, "tags": ["indoor"]},
)

results = memory.search(
    "hiking",
    scope=SearchScope(user_id="user_123"),
    max_results=5,
    record_types=["memory"],
    metadata_filter={"source": "survey"},
)

print([result.content for result in results])

outdoor_results = memory.search(
    "hiking",
    scope=SearchScope(user_id="user_123"),
    max_results=5,
    record_types=["memory"],
    metadata_filter={
        "source": "survey",
        "tags": {"$array_contains": "outdoor"},
    },
)

print([result.content for result in outdoor_results])

출력:

['The user likes alpine hiking.']

['The user likes alpine hiking.']

메모 또는 메시지만 검색

검색 결과를 특정 저장된 레코드 유형으로 제한하려면 record_types를 사용합니다.

thread = memory.create_thread(thread_id="thread_entity_type_search_123")
thread.add_messages(
    [
        {"role": "user", "content": "I mentioned pizza in a message."},
    ]
)
thread.add_memory("The user likes pizza.")

memory_results = thread.search("pizza", max_results=5, record_types=["memory"])
message_results = thread.search("pizza", max_results=5, record_types=["message"])

print([result.content for result in memory_results])

print([result.content for result in message_results])

출력:

['The user likes pizza.']

['I mentioned pizza in a message.']

전체 코드

#Copyright © 2026 Oracle and/or its affiliates.
#This software is under the Apache License 2.0
#(LICENSE-APACHE or http://www.apache.org/licenses/LICENSE-2.0) or Universal Permissive License
#(UPL) 1.0 (LICENSE-UPL or https://oss.oracle.com/licenses/upl), at your option.

#Oracle Agent Memory Code Example - Reference Sheet
#--------------------------------------------------

##Configure a LiteLLM LLM

from oracleagentmemory.core.llms.llm import Llm

llm = Llm(
    model="YOUR_LLM_MODEL",
    api_base="YOUR_LLM_API_BASE",
    api_key="YOUR_LLM_API_KEY",
)

response = llm.generate("What is 2+2?")
print(response.text)
#2+2 is equal to 4


##Configure a LiteLLM embedding model

from oracleagentmemory.core.embedders.embedder import Embedder

embedder = Embedder(
    model="YOUR_EMBEDDING_MODEL",
    api_base="YOUR_EMBEDDING_API_BASE",
    api_key="YOUR_EMBEDDING_API_KEY",
)

embedding_matrix = embedder.embed(["The quick brown fox jumps over the lazy dog"])
print(embedding_matrix.shape)
#(1, embedding_dimension)


##Configure an Oracle Memory component

import oracledb

from oracleagentmemory.core.oracleagentmemory import OracleAgentMemory

db_pool = oracledb.SessionPool(
    user="YOUR DB USER",
    password="YOUR DB PASSWORD",
    dsn="YOUR DB CONNECT STRING",
)

memory = OracleAgentMemory(
    connection=db_pool,
    embedder=embedder,
    llm=llm,  # optional: enables automatic memory extraction during add_messages()
)


##Configure an Oracle DB component

import oracledb

from oracleagentmemory.core import SchemaPolicy
from oracleagentmemory.core.oracleagentmemory import OracleAgentMemory

db_pool = oracledb.SessionPool(
    user="YOUR DB USER",
    password="YOUR DB PASSWORD",
    dsn="YOUR DB CONNECT STRING",
)

db_memory = OracleAgentMemory(
    connection=db_pool,
    embedder=embedder,
    llm=llm,  # optional
    schema_policy=SchemaPolicy.CREATE_IF_NECESSARY,
    memory_store_id="DEV_",
)


##Configure an Oracle Hybrid DB component

from oracleagentmemory.core import SchemaPolicy, SearchIndexSyncMode, SearchStrategy
from oracleagentmemory.core.embedders import OracleDBEmbedder
from oracleagentmemory.core.oracleagentmemory import OracleAgentMemory

db_embedder = OracleDBEmbedder(
    connection=db_pool,
    model="YOUR_DB_EMBEDDING_MODEL",
    embedding_dimension=384,
)

hybrid_db_memory = OracleAgentMemory(
    connection=db_pool,
    embedder=db_embedder,
    llm=llm,  # optional
    schema_policy=SchemaPolicy.CREATE_IF_NECESSARY,
    search_strategy=SearchStrategy.HYBRID,
    search_index_sync=SearchIndexSyncMode.ON_COMMIT,
)


##Create a thread

thread = memory.create_thread(
    thread_id="thread_create_123",  # optional
    user_id="user_123",          # optional
    agent_id="agent_456",        # optional
)

print(thread.thread_id)
#thread_create_123


##Re open an existing thread

thread = memory.create_thread(
    thread_id="thread_reopen_123",
    user_id="user_123",
    agent_id="agent_456",
)

same_thread = memory.get_thread("thread_reopen_123")
print(same_thread.thread_id)
#thread_reopen_123


##Update an existing thread

thread = memory.create_thread(
    thread_id="thread_update_123",
    user_id="user_123",
    agent_id="agent_456",
)

loaded_thread = memory.get_thread(
    "thread_update_123",
    max_message_token_length=8_000,
)
print(loaded_thread.max_message_token_length)
#8000

updated_thread = memory.update_thread(
    "thread_update_123",
    metadata={"source": "support", "flags": {"vip": True}},
    max_message_token_length=8_000,
)
persisted_thread = memory.get_thread("thread_update_123")

print(updated_thread.metadata["flags"]["vip"])
#True
print(persisted_thread.max_message_token_length)
#8000
#Overrides passed to get_thread() are temporary. Call update_thread()
#to persist thread metadata or durable runtime-config changes.


##Delete a thread

thread = memory.create_thread(thread_id="thread_delete_123")

deleted = memory.delete_thread("thread_delete_123")
print(deleted)
#1
#Use thread deletion when you need thread-scoped cascading cleanup.
#It removes the thread together with its messages, memories,
#and backing retrieval data managed by the SDK.


##Add a user profile

user_profile_id = memory.add_user(
    "user_123",
    "The user prefers concise answers and works mostly with Python.",
)

print(user_profile_id)
#user_123


##Add an agent profile

agent_profile_id = memory.add_agent(
    "agent_456",
    "A coding assistant specialized in debugging and code review.",
)

print(agent_profile_id)
#agent_456


##Add a global memory from the memory API

memory_id = memory.add_memory(
    "The user prefers short, bullet-point answers.",
    user_id="user_123",
    agent_id="agent_456",
)

print(memory_id)
#mem:1


##Add a scoped memory from the memory API

thread = memory.create_thread(
    thread_id="thread_scoped_123",
    user_id="user_123",
    agent_id="agent_456",
)

memory_id = memory.add_memory(
    "The user is planning a trip to Kyoto next month.",
    user_id="user_123",
    agent_id="agent_456",
    thread_id=thread.thread_id,
)

print(memory_id)
#mem:2


##Update a memory from the memory API

thread = memory.create_thread(
    thread_id="thread_update_memory_api_123",
    user_id="user_123",
    agent_id="agent_456",
)
memory_id = memory.add_memory(
    "The user likes short status updates.",
    user_id=thread.user_id,
    agent_id=thread.agent_id,
    thread_id=thread.thread_id,
    metadata={"source": "chat"},
)

updated_memory_id = memory.update_memory(
    memory_id,
    content="The user prefers short status updates.",
    metadata={"source": "support"},
)

print(updated_memory_id)
#mem:3


##Add a memory with a custom ID

memory_id = memory.add_memory(
    "The user prefers aisle seats on flights.",
    user_id="user_123",
    agent_id="agent_456",
    memory_id="travel_pref_001",
)

print(memory_id)
#travel_pref_001


##Add messages to a thread

from oracleagentmemory.apis import Message

thread = memory.create_thread(
    thread_id="thread_messages_123",
    user_id="user_123",
    agent_id="agent_456",
)

message_ids = thread.add_messages(
    [
        Message(
            id="msg_user_001",
            role="user",
            content="I prefer window seats on flights.",
            timestamp="2026-03-27T09:00:00Z",
            metadata={"source": "chat", "channel": "web"},
        ),
        {
            "id": "msg_assistant_001",
            "role": "assistant",
            "content": "Noted. I will keep that in mind.",
            "timestamp": "2026-03-27T09:00:05Z",
            "metadata": {"source": "assistant"},
        },
    ]
)

print(message_ids)
#['msg_user_001', 'msg_assistant_001']


##Read back thread messages

thread = memory.create_thread(thread_id="thread_read_messages_123")
thread.add_messages(
    [
        {"role": "user", "content": "Message 1"},
        {"role": "assistant", "content": "Message 2"},
        {"role": "user", "content": "Message 3"},
    ]
)

default_messages = thread.get_messages()
all_messages = thread.get_messages(end=None)
middle_messages = thread.get_messages(start=1, end=3)

print([message.content for message in default_messages])
#On short threads, the bounded default still returns all messages.
#['Message 1', 'Message 2', 'Message 3']
print([message.content for message in all_messages])
#['Message 1', 'Message 2', 'Message 3']
print([message.content for message in middle_messages])
#['Message 2', 'Message 3']


##Delete a message from the current thread by ID

thread = memory.create_thread(thread_id="thread_delete_message_123")
message_ids = thread.add_messages(
    [
        {"role": "user", "content": "Message to delete"},
    ]
)

deleted = thread.delete_message(message_ids[0])
print(deleted)
#1
#This removes only the raw message row from the current thread.
#Derived memories or other downstream artifacts created from that message
#are not deleted automatically and may remain searchable or appear in
#context-card output. Use memory.delete_thread(thread.thread_id) to delete
#the thread together with its associated messages and memories.
#Message deletes return 0 for IDs owned by another thread.


##Update a message from the current thread by ID

thread = memory.create_thread(thread_id="thread_update_message_123")
thread.add_messages(
    [
        {
            "id": "msg_update_001",
            "role": "user",
            "content": "Original message text.",
            "timestamp": "2026-03-27T10:00:00Z",
            "metadata": {"source": "chat"},
        }
    ]
)

updated_message_id = thread.update_message(
    "msg_update_001",
    content="Edited message text.",
    metadata={"source": "support"},
)
print(updated_message_id)
#msg_update_001
#Message updates preserve stored role and timestamp values.
#When automatic extraction is enabled, content edits immediately rerun
#extraction for the edited message using the same history-window
#rules as add_messages().
#Later messages are ignored during that immediate pass.
#Existing derived memories stay in place while new edited-content
#memories are added during that refresh.


##Add a memory from a thread handle

thread = memory.create_thread(
    thread_id="thread_add_memory_123",
    user_id="user_123",
    agent_id="agent_456",
)

memory_id = thread.add_memory("The user likes jasmine tea.")
print(memory_id)
#mem:4


##Update a memory from the current thread by ID

thread = memory.create_thread(
    thread_id="thread_update_memory_123",
    user_id="user_123",
    agent_id="agent_456",
)
memory_id = thread.add_memory(
    "The user likes jasmine tea.",
    metadata={"source": "survey"},
)

updated_memory_id = thread.update_memory(
    memory_id,
    content="The user likes jasmine tea in the afternoon.",
    metadata={"source": "support"},
)
print(updated_memory_id)
#mem:5
#Thread updates are scoped to the current thread and raise KeyError
#for missing IDs or IDs owned by another thread.


##Delete a memory from the current thread by ID

thread = memory.create_thread(thread_id="thread_delete_memory_123")
memory_id = thread.add_memory("Temporary memory to delete.")

deleted = thread.delete_memory(memory_id)
print(deleted)
#1
#Thread deletes are scoped to the current thread and return 0 for IDs owned by another thread.


##Build a context card

thread = memory.create_thread(thread_id="thread_context_card_123")
thread.add_messages(
    [
        {"role": "user", "content": "I am planning a trip to Kyoto next spring."},
    ]
)
thread.add_memory("The user is planning a trip to Kyoto.")

context_card = thread.get_context_card()
print(context_card.content)
#<context_card>
#The user is planning a trip to Kyoto.
#</context_card>


##Build a thread summary

thread = memory.create_thread(thread_id="thread_summary_123")
thread.add_messages(
    [
        {"role": "user", "content": "Hello"},
        {"role": "assistant", "content": "Hi, how can I help?"},
        {"role": "user", "content": "Please summarize this thread."},
    ]
)

summary = thread.get_summary()
print(summary.content)
#user (-): Hello
#- assistant (-): Hi, how can I help?
#- user (-): Please summarize this thread.


##Build a summary excluding the last N messages

thread = memory.create_thread(thread_id="thread_summary_except_last_123")
thread.add_messages(
    [
        {"role": "user", "content": "First message"},
        {"role": "assistant", "content": "Second message"},
        {"role": "user", "content": "Third message"},
    ]
)

summary = thread.get_summary(except_last=1)
print(summary.content)
#user (-): First message
#- assistant (-): Second message


##Build a summary with a token budget

thread = memory.create_thread(thread_id="thread_summary_budget_123")
thread.add_messages(
    [
        {"role": "user", "content": "Message 1"},
        {"role": "assistant", "content": "Message 2"},
        {"role": "user", "content": "Message 3"},
        {"role": "assistant", "content": "Message 4"},
    ]
)

summary = thread.get_summary(token_budget=20)
print(summary.content)
#(truncated)
#user (-): Message 1
#...


##Search from a thread with no explicit scoping

thread = memory.create_thread(
    thread_id="thread_search_default_123",
    user_id="user_123",
    agent_id="agent_456",
)
thread.add_memory("The user likes pizza.")
thread.add_memory("The user likes cats.")

results = thread.search("pizza", max_results=5)
print([result.content for result in results])
#['The user likes pizza.']


##Search from the memory API with scoping

from oracleagentmemory.apis.searchscope import SearchScope

thread = memory.create_thread(
    thread_id="thread_memory_search_123",
    user_id="user_123",
    agent_id="agent_456",
)
thread.add_memory("The user likes hiking in the Alps.")

results = memory.search(
    "hiking",
    scope=SearchScope(
        user_id="user_123",
        agent_id="agent_456",
        thread_id="thread_memory_search_123",
        exact_thread_match=True,
    ),
    max_results=5,
)

print([result.content for result in results])
#['The user likes hiking in the Alps.']


##Search with metadata filtering

from oracleagentmemory.apis.searchscope import SearchScope

thread = memory.create_thread(
    thread_id="thread_metadata_filter_123",
    user_id="user_123",
    agent_id="agent_456",
)
thread.add_memory(
    "The user likes alpine hiking.",
    metadata={"source": "survey", "category": {"kind": "travel"}, "tags": ["outdoor"]},
)
thread.add_memory(
    "The user likes indoor climbing.",
    metadata={"source": "chat", "category": {"kind": "sports"}, "tags": ["indoor"]},
)

results = memory.search(
    "hiking",
    scope=SearchScope(user_id="user_123"),
    max_results=5,
    record_types=["memory"],
    metadata_filter={"source": "survey"},
)

print([result.content for result in results])
#['The user likes alpine hiking.']

outdoor_results = memory.search(
    "hiking",
    scope=SearchScope(user_id="user_123"),
    max_results=5,
    record_types=["memory"],
    metadata_filter={
        "source": "survey",
        "tags": {"$array_contains": "outdoor"},
    },
)

print([result.content for result in outdoor_results])
#['The user likes alpine hiking.']


##Search only memories or messages

thread = memory.create_thread(thread_id="thread_entity_type_search_123")
thread.add_messages(
    [
        {"role": "user", "content": "I mentioned pizza in a message."},
    ]
)
thread.add_memory("The user likes pizza.")

memory_results = thread.search("pizza", max_results=5, record_types=["memory"])
message_results = thread.search("pizza", max_results=5, record_types=["message"])

print([result.content for result in memory_results])
#['The user likes pizza.']
print([result.content for result in message_results])
#['I mentioned pizza in a message.']