WayFlowでのエージェント・メモリーの使用

この記事では、エージェント・メモリーをWayFlowエージェントに接続して、エージェントがセッション間で永続ファクトを再利用できるようにする方法を学習します。

ヒント:パッケージの設定については、エージェント・メモリーの開始を参照してください。この例にローカルのOracle AI Databaseが必要な場合は、Run Oracle AI Database Locallyを参照してください。WayFlowについてさらに学習するには、(https://github.com/oracle/wayflow)を参照してください。

エージェント・メモリーおよびWayFlowの構成

ユーザーのエージェント・メモリー・スレッドを作成し、search_memoryツールをWayFlowに公開して、OpenAIModelを使用してWayFlow Agentを構築します。search_memoryツールは、エージェント・メモリーAPIをユーザーおよびエージェントIDで問い合せるため、現在のスレッドに制限されるのではなく、以前のセッションから永続ファクトをリカバリできます。また、エージェント・メモリー・クライアントは、独自LLMを使用して最近のスレッド・メッセージから永続メモリーを定期的に抽出しますが、WayFlow Agentは、返信およびツール・コールに独自のOpenAIModelを引き続き使用します。

import oracledb

from oracleagentmemory.core.embedders.embedder import Embedder
from oracleagentmemory.core.llms.llm import Llm
from oracleagentmemory.core.oracleagentmemory import OracleAgentMemory
from wayflowcore.agent import Agent
from wayflowcore.models import OpenAIModel
from wayflowcore.tools import tool

embedder = Embedder(
    model="YOUR_EMBEDDING_MODEL",
    api_base="YOUR_EMBEDDING_API_BASE",
    api_key="YOUR_EMBEDDING_API_KEY",
)
llm = Llm(
    model="gpt-4.1-mini",
    api_key="YOUR_OPENAI_API_KEY",
)
db_pool = oracledb.SessionPool(
    user="YOUR DB USER",
    password="YOUR DB PASSWORD",
    dsn="YOUR DB CONNECT STRING",
)
wayflow_llm = OpenAIModel(
    model_id="gpt-4.1-mini",
    api_key="YOUR_OPENAI_API_KEY",
)


#Keep these identifiers stable for the same assistant and end user so memory
#is scoped consistently across threads and sessions.
agent_id = "support_agent"
user_id = "user_123"
memory = OracleAgentMemory(
    connection=db_pool,
    embedder=embedder,
    llm=llm,
)
memory_thread = memory.create_thread(
    thread_id="wayflow_memory_demo",
    user_id=user_id,
    agent_id=agent_id,
)

from typing import Annotated


@tool
def search_memory(
    query: Annotated[str, "Question to search in Oracle Agent Memory"],
) -> Annotated[str, "Top matching memory content"]:
    """Search Oracle Agent Memory for durable user facts relevant to the current request."""
    results = memory.search(
        query=query,
        user_id=user_id,
        agent_id=agent_id,
        max_results=1,
        record_types=["memory"],
    )
    if not results:
        return "No relevant memory found."
    return results[0].content

assistant = Agent(
    llm=wayflow_llm,
    agent_id=agent_id,
    custom_instruction=(
        "You are a support agent. When the user asks about durable facts from "
        "prior sessions, call the search_memory tool before answering."
    ),
    tools=[search_memory],
)

セッション後にユーザー・コンテキストを保持

各WayFlowセッションの後に、交換されたメッセージをエージェント・メモリーに追加し、後で再利用する必要がある永続ファクトをすべて格納します。

session_1 = assistant.start_conversation()
user_message = (
    "I am John, a Python developer and I need help debugging a payment service."
)
session_1.append_user_message(user_message)
session_1.execute()
assistant_reply = session_1.get_last_message().content

print(assistant_reply)
#I can help with that. What error are you seeing?

#add_messages will add messages to the DB and extract memories automatically
memory_thread.add_messages(
    [
        {"role": "user", "content": user_message},
        {"role": "assistant", "content": assistant_reply},
    ]
)
#add_memory adds memory to the DB
memory_thread.add_memory("The user is John, a Python developer.")

新しいWayFlowセッションでのメモリーの再利用

後のセッションが開始したら、同じエージェント・メモリー・スレッドを再オープンし、WayFlowエージェントがsearch_memoryをコールして以前のユーザー・コンテキストをリカバリできるようにします。

memory_thread = memory.get_thread("wayflow_memory_demo")
assistant = Agent(
    llm=wayflow_llm,
    agent_id=agent_id,
    custom_instruction=(
        "You are a support agent. When the user asks about durable facts from "
        "prior sessions, call the search_memory tool before answering."
    ),
    tools=[search_memory],
)

session_2 = assistant.start_conversation()
session_2.append_user_message("Who am I?")
session_2.execute()
remembered_reply = session_2.get_last_message().content

print(remembered_reply)

出力:

The user is John, a Python developer.

高度な使用

より緊密な統合のために、ConversationMessageAddedEventエントリをバッファリングし、実行の最後にエージェント・メモリーに書き込むWayFlowイベント・リスナーを登録できます。これにより、エージェント・メモリーの更新パスがメイン・スレッド・コードから切り離されたまま、最後に交換されたメッセージが保持されます。

from wayflowcore.events.event import ConversationMessageAddedEvent
from wayflowcore.events.eventlistener import GenericEventListener, register_event_listeners

pending_messages: list[dict[str, str]] = []


def _buffer_thread_message(event: ConversationMessageAddedEvent) -> None:
    if event.streamed:
        return
    pending_messages.append(
        {
            "role": event.message.role,
            "content": event.message.content,
        }
    )


message_listener = GenericEventListener(
    [ConversationMessageAddedEvent],
    _buffer_thread_message,
)

with register_event_listeners([message_listener]):
    session_3 = assistant.start_conversation()
    session_3.append_user_message("Please remember that I prefer concise code reviews.")
    session_3.execute()

if pending_messages:
    memory_thread.add_messages(pending_messages)

自動抽出の無効化

メッセージのみを永続化し、永続メモリーを手動で追加する場合は、extract_memories=Falseを使用してエージェント・メモリー・クライアントを作成し、永続メモリー行を自分で挿入します。

manual_memory = OracleAgentMemory(
    connection=db_pool,
    embedder=embedder,
    extract_memories=False,
)
manual_memory_thread = manual_memory.create_thread(
    thread_id="wayflow_manual_memory_demo",
    user_id=user_id,
    agent_id=agent_id,
)
manual_memory_thread.add_messages(
    [
        {
            "role": "user",
            "content": "Please remember that I prefer concise code reviews.",
        },
        {
            "role": "assistant",
            "content": "Understood. I will keep responses concise.",
        },
    ]
)
manual_memory_thread.add_memory("The user prefers concise code reviews.")

まとめ

この記事では、エージェントメモリを WayFlowエージェントに接続し、各セッション後に交換されたメッセージと永続メモリを永続化し、後の実行で以前のユーザーコンテキストを再利用する方法を学びました。

ヒント:エージェント・メモリーをWayFlowと統合する方法を学習した後、エージェント・メモリーをLangGraphと統合することもできます。

完全コード

#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.
#Agent Memory Code Example - Integration with WayFlow
#-----------------------------------------------------

##Configure Oracle Memory and WayFlow

import oracledb

from oracleagentmemory.core.embedders.embedder import Embedder
from oracleagentmemory.core.llms.llm import Llm
from oracleagentmemory.core.oracleagentmemory import OracleAgentMemory
from wayflowcore.agent import Agent
from wayflowcore.models import OpenAIModel
from wayflowcore.tools import tool

embedder = Embedder(
    model="YOUR_EMBEDDING_MODEL",
    api_base="YOUR_EMBEDDING_API_BASE",
    api_key="YOUR_EMBEDDING_API_KEY",
)
llm = Llm(
    model="gpt-4.1-mini",
    api_key="YOUR_OPENAI_API_KEY",
)
db_pool = oracledb.SessionPool(
    user="YOUR DB USER",
    password="YOUR DB PASSWORD",
    dsn="YOUR DB CONNECT STRING",
)
wayflow_llm = OpenAIModel(
    model_id="gpt-4.1-mini",
    api_key="YOUR_OPENAI_API_KEY",
)

#Keep these identifiers stable for the same assistant and end user so memory
#is scoped consistently across threads and sessions.
agent_id = "support_agent"
user_id = "user_123"
memory = OracleAgentMemory(
    connection=db_pool,
    embedder=embedder,
    llm=llm,
)
memory_thread = memory.create_thread(
    thread_id="wayflow_memory_demo",
    user_id=user_id,
    agent_id=agent_id,
)

from typing import Annotated

@tool
def search_memory(
    query: Annotated[str, "Question to search in Oracle Agent Memory"],
) -> Annotated[str, "Top matching memory content"]:
    """Search Oracle Agent Memory for durable user facts relevant to the current request."""
    results = memory.search(
        query=query,
        user_id=user_id,
        agent_id=agent_id,
        max_results=1,
        record_types=["memory"],
    )
    if not results:
        return "No relevant memory found."
    return results[0].content

assistant = Agent(
    llm=wayflow_llm,
    agent_id=agent_id,
    custom_instruction=(
        "You are a support agent. When the user asks about durable facts from "
        "prior sessions, call the search_memory tool before answering."
    ),
    tools=[search_memory],
)

##Persist user context after a session

session_1 = assistant.start_conversation()
user_message = (
    "I am John, a Python developer and I need help debugging a payment service."
)
session_1.append_user_message(user_message)
session_1.execute()
assistant_reply = session_1.get_last_message().content

print(assistant_reply)
#I can help with that. What error are you seeing?

memory_thread.add_messages(
    [
        {"role": "user", "content": user_message},
        {"role": "assistant", "content": assistant_reply},
    ]
)
memory_thread.add_memory("The user is John, a Python developer.")

##Reuse memory in a new WayFlow session

memory_thread = memory.get_thread("wayflow_memory_demo")
assistant = Agent(
    llm=wayflow_llm,
    agent_id=agent_id,
    custom_instruction=(
        "You are a support agent. When the user asks about durable facts from "
        "prior sessions, call the search_memory tool before answering."
    ),
    tools=[search_memory],
)

session_2 = assistant.start_conversation()
session_2.append_user_message("Who am I?")
session_2.execute()
remembered_reply = session_2.get_last_message().content

print(remembered_reply)
#The user is John, a Python developer.

##Advanced use event listeners

from wayflowcore.events.event import ConversationMessageAddedEvent
from wayflowcore.events.eventlistener import GenericEventListener, register_event_listeners

pending_messages: list[dict[str, str]] = []

def _buffer_thread_message(event: ConversationMessageAddedEvent) -> None:
    if event.streamed:
        return
    pending_messages.append(
        {
            "role": event.message.role,
            "content": event.message.content,
        }
    )

message_listener = GenericEventListener(
    [ConversationMessageAddedEvent],
    _buffer_thread_message,
)

with register_event_listeners([message_listener]):
    session_3 = assistant.start_conversation()
    session_3.append_user_message("Please remember that I prefer concise code reviews.")
    session_3.execute()

if pending_messages:
    memory_thread.add_messages(pending_messages)

##Disable automatic memory extraction

manual_memory = OracleAgentMemory(
    connection=db_pool,
    embedder=embedder,
    extract_memories=False,
)
manual_memory_thread = manual_memory.create_thread(
    thread_id="wayflow_manual_memory_demo",
    user_id=user_id,
    agent_id=agent_id,
)
manual_memory_thread.add_messages(
    [
        {
            "role": "user",
            "content": "Please remember that I prefer concise code reviews.",
        },
        {
            "role": "assistant",
            "content": "Understood. I will keep responses concise.",
        },
    ]
)
manual_memory_thread.add_memory("The user prefers concise code reviews.")