Memory as an agent tool

pip install memsy openai

RECALL_TOOL = {
    "type": "function",
    "function": {
        "name": "recall_memory",
        "description": (
            "Search this user's long-term memory for things they have "
            "told you in past sessions — preferences, decisions, prior "
            "context, and facts about their work. Call this when the "
            "user refers to something from before ('like last time', "
            "'the usual', 'what we decided'), when a personal "
            "preference would change your answer, or when you are "
            "missing context the user plainly expects you to have. Do "
            "not call it for general knowledge or for facts already "
            "stated in this conversation."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": (
                        "What to look for, as a natural-language "
                        "phrase. Search is semantic, so describe the "
                        "topic rather than guessing keywords."
                    ),
                }
            },
            "required": ["query"],
        },
    },
}

import os

from memsy import MemsyClient
from memsy.exceptions import MemsyError

memsy = MemsyClient(
    base_url=os.environ["MEMSY_BASE_URL"],
    api_key=os.environ["MEMSY_API_KEY"],
)


def make_recall(actor_id: str):
    """Bind the actor once; the model only ever supplies `query`."""

    def recall(query: str) -> str:
        try:
            # No threshold: the default 0.0 returns the top 5 on any
            # tier. A floor cuts noise, but the useful value is
            # tier-dependent — see /docs/searching-memory#threshold.
            found = memsy.search(query, actor_id=actor_id, limit=5)
        except MemsyError:
            # Tell the model the truth: the lookup failed, the user has
            # no history. Otherwise it reports "you have never mentioned
            # that" — a confident, wrong claim.
            return "Memory lookup failed. Proceed without it."

        if not found.results:
            return "No relevant memories found."
        return "\n".join(f"- {r.content}" for r in found.results)

    return recall

import json

from openai import OpenAI

openai = OpenAI()

MAX_STEPS = 6


def run(actor_id: str, user_message: str) -> str:
    recall = make_recall(actor_id)
    messages = [
        {
            "role": "system",
            "content": (
                "You are a personal assistant with access to the user's "
                "long-term memory. Use the recall_memory tool when past "
                "context would change your answer."
            ),
        },
        {"role": "user", "content": user_message},
    ]

    for _ in range(MAX_STEPS):
        response = openai.chat.completions.create(
            model="gpt-4o-mini", messages=messages, tools=[RECALL_TOOL]
        )
        message = response.choices[0].message
        messages.append(message)

        if not message.tool_calls:
            return message.content or ""

        for call in message.tool_calls:
            # `tool_calls` is a union of function and custom tool calls,
            # and only the function variant carries `.function`. Every
            # call still needs a reply: an assistant message whose tool
            # calls go unanswered is rejected on the next request.
            if call.type != "function":
                messages.append(
                    {
                        "role": "tool",
                        "tool_call_id": call.id,
                        "content": "Unsupported tool call.",
                    }
                )
                continue

            # `required: ["query"]` is not enforced unless the tool is
            # declared with `strict: True`, and a truncated stream can
            # cut the JSON mid-object. Either way this must answer the
            # call, not raise out of run() and take the whole turn.
            try:
                query = json.loads(call.function.arguments)["query"]
            except (json.JSONDecodeError, KeyError, TypeError):
                messages.append(
                    {
                        "role": "tool",
                        "tool_call_id": call.id,
                        "content": "Could not read the tool arguments.",
                    }
                )
                continue

            messages.append(
                {
                    "role": "tool",
                    "tool_call_id": call.id,
                    "content": recall(query),
                }
            )

    # Out of steps. Returning a truthful failure beats returning the
    # model's last half-finished thought as if it were an answer.
    return "I could not finish that request."

def chat(actor_id: str, session_id: str, user_message: str) -> str:
    answer = run(actor_id, user_message)

    memsy.ingest(
        [
            EventPayload(
                actor_id=actor_id,
                session_id=session_id,
                kind="user_message",
                content=user_message,
            ),
            EventPayload(
                actor_id=actor_id,
                session_id=session_id,
                kind="assistant_message",
                content=answer,
            ),
        ]
    )
    return answer

from memsy import EventPayload, MemsyClient