Skip to content
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
77 changes: 77 additions & 0 deletions python/samples/concepts/memory/synap_memory.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,77 @@
# Copyright (c) Microsoft. All rights reserved.

"""Sample: long-term memory via the Synap plugin.

[Synap](https://maximem.ai) is a managed long-term memory layer for AI agents.
The `maximem-synap-semantic-kernel` package exposes it as a Semantic Kernel
plugin (`SynapPlugin`) with `search_memory` and `store_memory` kernel functions,
so a chat completion service can recall and persist facts across sessions.

Setup:
pip install maximem-synap-semantic-kernel maximem-synap semantic-kernel
export SYNAP_API_KEY=<your-key> # https://synap.maximem.ai
export OPENAI_API_KEY=<your-key>

Open source integration package:
https://github.com/maximem-ai/maximem_synap_sdk/tree/main/packages/integrations/synap-semantic-kernel
"""

import asyncio
import os

from maximem_synap import MaximemSynapSDK
from semantic_kernel import Kernel
from semantic_kernel.connectors.ai.function_choice_behavior import (
FunctionChoiceBehavior,
)
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion
from semantic_kernel.connectors.ai.open_ai.prompt_execution_settings.open_ai_prompt_execution_settings import (
OpenAIChatPromptExecutionSettings,
)
from semantic_kernel.contents.chat_history import ChatHistory
from synap_semantic_kernel import SynapPlugin


async def main() -> None:
sdk = MaximemSynapSDK(api_key=os.environ["SYNAP_API_KEY"])
await sdk.initialize()

kernel = Kernel()
kernel.add_service(OpenAIChatCompletion(ai_model_id="gpt-4o-mini"))
kernel.add_plugin(
SynapPlugin(sdk=sdk, user_id="demo-user-001", customer_id="demo-customer"),
plugin_name="synap",
)

chat_service = kernel.get_service()
settings = OpenAIChatPromptExecutionSettings(
function_choice_behavior=FunctionChoiceBehavior.Auto(),
)

history = ChatHistory()
history.add_system_message(
"You are a helpful assistant with long-term memory. "
"Use synap-search_memory to recall facts about the user. "
"Use synap-store_memory to save important new facts."
)

# Turn 1: teach
history.add_user_message(
"I'm a software engineer who's allergic to peanuts. Remember this."
)
response = await chat_service.get_chat_message_content(
chat_history=history, settings=settings, kernel=kernel
)
print(f"Assistant: {response}")
history.add_message(response)
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nitpick: get_chat_message_content returns ChatMessageContent | None. If response is None, add_message will raise TypeError. The None path is effectively dead code today (marked pragma: no cover) and other samples also skip this guard, so this is non-blocking—but adding a guard would be more defensive.

Suggested change
history.add_message(response)
if response:
history.add_message(response)


# Turn 2: recall
history.add_user_message("What do you know about my dietary restrictions?")
response = await chat_service.get_chat_message_content(
chat_history=history, settings=settings, kernel=kernel
)
print(f"Assistant: {response}")

Comment on lines +69 to +74
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The PR describes persistent cross-session memory, but history still contains the user's original fact from the first turn. The model can answer from short-term chat history even if synap-store_memory/synap-search_memory are never invoked. Re-create the ChatHistory before the recall turn to actually exercise the long-term memory path.

Suggested change
history.add_user_message("What do you know about my dietary restrictions?")
response = await chat_service.get_chat_message_content(
chat_history=history, settings=settings, kernel=kernel
)
print(f"Assistant: {response}")
# Turn 2: recall from a fresh session so Synap, not short-term chat history, supplies the fact
history = ChatHistory()
history.add_system_message(
"You are a helpful assistant with long-term memory. "
"Use synap-search_memory to recall facts about the user. "
"Use synap-store_memory to save important new facts."
)
history.add_user_message("What do you know about my dietary restrictions?")
response = await chat_service.get_chat_message_content(
chat_history=history, settings=settings, kernel=kernel
)
print(f"Assistant: {response}")


if __name__ == "__main__":
asyncio.run(main())
Loading