Memory

from __future__ import annotations

from typing import Annotated, TypedDict

from langchain_core.messages import AIMessage, BaseMessage
from langgraph.graph import END, StateGraph
from langgraph.graph.message import add_messages
from langgraph.graph.state import RunnableConfig

from ibm_watsonx_orchestrate_sdk import Client


class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]


def _latest_user_message(messages: list[BaseMessage]) -> str:
    for message in reversed(messages):
        if getattr(message, "type", "") == "human":
            content = getattr(message, "content", "")
            if isinstance(content, str) and content.strip():
                return content
    return ""


def create_agent(config: RunnableConfig):
    client = Client.from_runnable_config(config)

    def agent_node(state: AgentState):
        user_text = _latest_user_message(state.get("messages", []))
        if not user_text:
            return {"messages": [AIMessage(content="No user message found.")]}

        search_response = client.memory.search(query=user_text, limit=3)

        if search_response.results:
            memory_text = "\n".join(item.content for item in search_response.results)
            response_text = f"I found related memories:\n{memory_text}"
        else:
            response_text = "I do not have anything relevant in memory yet."

        return {"messages": [AIMessage(content=response_text)]}

    builder = StateGraph(AgentState)
    builder.add_node("agent", agent_node)
    builder.set_entry_point("agent")
    builder.add_edge("agent", END)
    return builder

./.venv/bin/orchestrate agents import \
  --experimental-package-root examples/custom_agents/local_memory_sdk_agent

client.memory.add_messages(...)

client.memory.search(...)
client.memory.list(...)
client.memory.retrieve(...)

client.memory.delete(memory_id="...")
client.memory.delete_all()

client.memory.add_messages(
    messages=[{"role": "user", "content": "I prefer coffee"}],
    memory_type="preference",
    infer=False,
    metadata={"source": "manual-test"},
)

response = client.memory.search(query="favorite drink", limit=3)

response = client.memory.search(
    query="favorite drink",
    limit=3,
    memory_type="preference",
)

add_messages(
    *,
    messages,
    infer=None,
    memory_type=None,
    metadata=None,
    agent_id=None,
    run_id=None,
    sensitivity_classification=None,
    source_reference=None,
)

search(
    *,
    query,
    limit=10,
    memory_type=None,
    expanded_query=None,
    recall=None,
)

retrieve(query, limit=10)

list(*, limit=100, offset=0)

delete(*, memory_id)

delete_all()

{"role": "user", "content": "I prefer coffee"}

metadata={
    "source": "chat",
    "source_request_id": "req-123",
    "scope": "user_profile",
    "case_id": "case-42",
    "ticket_id": "ticket-99",
}

from ibm_watsonx_orchestrate_sdk import Client


def remember_preference(config, text: str):
    client = Client.from_runnable_config(config)
    return client.memory.add_messages(
        messages=[{"role": "user", "content": text}],
        memory_type="preference",
        infer=False,
        metadata={"source": "agent"},
    )


def recall_preferences(config, query: str):
    client = Client.from_runnable_config(config)
    return client.memory.search(
        query=query,
        memory_type="preference",
        limit=3,
    )