3
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@engchinain日本オラクル株式会社

Oracle AI Agent MemoryをLangGraph + OCI Cohere Embed v4 + Grok 4.20 Reasoningで試してみた

3
Posted at

はじめに

本記事では、Oracle AI Database の Agent Memory を LangGraph と組み合わせて試してみます。

Agent Memory を利用すると、AI Agent の会話メッセージや、ユーザーに関する durable memory を Oracle Database に保存できます。

LangGraph 側では、保存済みの memory を検索し、その内容を LLM に渡して応答に反映します。

今回は公式ドキュメントの LangGraph 連携サンプルをベースに、以下の2パターンを試します。

  1. Prebuilt ReAct Agent
    search_memory tool を LangGraph agent に渡し、必要に応じて memory を検索する方式

  2. Custom Flow
    StateGraph(MessagesState) を使い、model node の中で Agent Memory を検索して system message に注入する方式

今回は以下の構成で動作確認します。

項目 使用するもの
Database Oracle AI Database Free Lite
Agent Memory SDK oracleagentmemory==26.4.0
LangGraph StateGraph(MessagesState)
Embedding Model oci/cohere.embed-v4.0
LLM oci/xai.grok-4.20-reasoning
LangChain LLM wrapper ChatLiteLLM
DB接続 oracledb.SessionPool

1. Oracle AI Database Free Lite を起動

まず、Oracle AI Database Free Lite のコンテナを起動します。

export ORACLE_PWD='<your-secure-password>'

イメージを取得します。

docker pull container-registry.oracle.com/database/free:latest-lite

データを永続化するため、Docker Volume を作成します。

docker volume create OracleDBData

コンテナを起動します。

docker run -d \
  --name oracle-free-lite \
  -p 1521:1521 \
  -e ORACLE_PWD="$ORACLE_PWD" \
  -v OracleDBData:/opt/oracle/oradata \
  container-registry.oracle.com/database/free:latest-lite

起動ログを確認します。

docker logs -f oracle-free-lite

以下のメッセージが表示されれば準備完了です。

DATABASE IS READY TO USE!

2. PDB に接続確認

コンテナ内の SQL*Plus で接続します。

docker exec -it oracle-free-lite sqlplus system/"$ORACLE_PWD"@FREEPDB1

PDB 名を確認します。

SELECT sys_context('USERENV', 'CON_NAME') AS container_name FROM dual;

以下のように FREEPDB1 が表示されれば OK です。

CONTAINER_NAME
----------------
FREEPDB1

SQL*Plus から抜けます。

exit

3. Agent Memory 用ユーザーを作成

今回は dmuser という専用ユーザーを作成します。

docker exec -it oracle-free-lite sqlplus system/"$ORACLE_PWD"@FREEPDB1

以下の SQL を実行します。

CREATE TABLESPACE dmuser_ts
DATAFILE '/opt/oracle/oradata/FREE/FREEPDB1/dmuser_ts01.dbf'
SIZE 200M
AUTOEXTEND ON NEXT 100M
SEGMENT SPACE MANAGEMENT AUTO;

CREATE USER dmuser IDENTIFIED BY "CHOOSE_A_STRONG_PASSWORD";

GRANT CREATE SESSION, CREATE TABLE, CREATE SEQUENCE, CREATE VIEW, CREATE PROCEDURE TO dmuser;

ALTER USER dmuser DEFAULT TABLESPACE dmuser_ts;
ALTER USER dmuser QUOTA UNLIMITED ON dmuser_ts;

CHOOSE_A_STRONG_PASSWORD は任意の強力なパスワードに置き換えてください。

作成したユーザーを確認します。

SELECT username, account_status, default_tablespace, temporary_tablespace
FROM dba_users
WHERE username = 'DMUSER';

SELECT privilege
FROM dba_sys_privs
WHERE grantee = 'DMUSER'
ORDER BY privilege;

確認が終わったら SQL*Plus を終了します。

exit

4. Python 依存パッケージをインストール

作業用ディレクトリを作成します。

mkdir try-oracle-agent-memory
cd try-oracle-agent-memory

Python プロジェクトを初期化します。

uv init

必要なパッケージを追加します。

uv add "oracleagentmemory==26.4.0" \
  oracledb \
  oci \
  python-dotenv \
  langchain \
  langchain-core \
  langgraph \
  langchain-litellm

5. 設定ファイルを作成

DB 接続情報、OCI 設定、利用モデルを .env にまとめます。

vi .env

.env の例です。

# Oracle Database
ORACLE_MEMORY_DB_USER=dmuser
ORACLE_MEMORY_DB_PASSWORD=<your-app-user-password>
ORACLE_MEMORY_DB_CONNECT_STRING=localhost:1521/FREEPDB1
ORACLE_MEMORY_TABLE_NAME_PREFIX=T_LANGGRAPH_MEM_

# Schema policy
# 初回は RECREATE でスキーマを作成します。
# 既存スキーマを再利用したい場合は REQUIRE_EXISTING に変更します。
ORACLE_MEMORY_SCHEMA_POLICY=RECREATE

# OCI config
OCI_CONFIG_FILE=~/.oci/config
OCI_PROFILE=DEFAULT
OCI_COMPARTMENT_ID=

# Oracle Agent Memory models
OCI_EMBEDDING_MODEL=oci/cohere.embed-v4.0
OCI_LLM_MODEL=oci/xai.grok-4.20-reasoning

# Demo identifiers
AGENT_ID=support_agent
USER_ID=user_123

ORACLE_MEMORY_DB_PASSWORD には、先ほど dmuser 作成時に指定したパスワードを設定します。

OCI 認証情報は OCI_CONFIG_FILE と OCI_PROFILE で指定した config から読み込みます。

OCI_COMPARTMENT_ID を .env に指定した場合はその値を使います。空の場合は ~/.oci/config の compartment_id を使います。

~/.oci/config の例です。

[DEFAULT]
user=ocid1.user.oc1...
fingerprint=xx:xx:xx:xx
tenancy=ocid1.tenancy.oc1...
region=ap-tokyo-1
key_file=/home/opc/.oci/oci_api_key.pem
compartment_id=ocid1.compartment.oc1...

.env にはパスワードや OCID などの情報を含むため、Git 管理対象から除外します。

echo ".env" >> .gitignore

6. LangGraph + Agent Memory のサンプルコード

以下の Python ファイルを作成します。

vi try_agent_memory_langgraph_oci.py

try_agent_memory_langgraph_oci.py の中身です。

import os
from typing import Annotated

from dotenv import load_dotenv

load_dotenv()

os.environ["LITELLM_LOCAL_MODEL_COST_MAP"] = "True"

import oci
import oracledb

from langchain.agents import create_agent
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_core.tools import tool
from langchain_litellm import ChatLiteLLM
from langgraph.graph import END, START, MessagesState, StateGraph

from oracleagentmemory.core import SchemaPolicy
from oracleagentmemory.core.embedders.embedder import Embedder
from oracleagentmemory.core.llms.llm import Llm
from oracleagentmemory.core.oracleagentmemory import OracleAgentMemory


def get_optional_env(name: str) -> str | None:
    value = os.environ.get(name)
    if value is None:
        return None

    value = value.strip()
    return value or None


def get_required_env(name: str) -> str:
    value = get_optional_env(name)
    if not value:
        raise RuntimeError(f"{name} is not set. Please check your .env file.")
    return value


def get_schema_policy() -> SchemaPolicy:
    schema_policy_name = os.environ.get(
        "ORACLE_MEMORY_SCHEMA_POLICY",
        "RECREATE",
    ).upper()

    try:
        return getattr(SchemaPolicy, schema_policy_name)
    except AttributeError as exc:
        valid_values = ", ".join(
            name for name in dir(SchemaPolicy) if name.isupper()
        )
        raise RuntimeError(
            f"Invalid ORACLE_MEMORY_SCHEMA_POLICY: {schema_policy_name}. "
            f"Valid values are: {valid_values}"
        ) from exc


def load_oci_common_args() -> dict:
    oci_config_file = os.environ.get("OCI_CONFIG_FILE", "~/.oci/config")
    oci_profile = os.environ.get("OCI_PROFILE", "DEFAULT")

    config = oci.config.from_file(
        file_location=os.path.expanduser(oci_config_file),
        profile_name=oci_profile,
    )

    compartment_id = get_optional_env("OCI_COMPARTMENT_ID") or config.get(
        "compartment_id"
    )

    if not compartment_id:
        raise RuntimeError(
            "OCI compartment id is not set. "
            "Please set OCI_COMPARTMENT_ID in .env or compartment_id in ~/.oci/config."
        )

    return {
        "oci_compartment_id": compartment_id,
        "oci_region": config["region"],
        "oci_user": config["user"],
        "oci_fingerprint": config["fingerprint"],
        "oci_tenancy": config["tenancy"],
        "oci_key_file": os.path.expanduser(config["key_file"]),
    }


def create_db_pool() -> oracledb.SessionPool:
    db_user = os.environ.get("ORACLE_MEMORY_DB_USER", "dmuser")
    db_password = get_required_env("ORACLE_MEMORY_DB_PASSWORD")
    db_connect_string = os.environ.get(
        "ORACLE_MEMORY_DB_CONNECT_STRING",
        "localhost:1521/FREEPDB1",
    )

    return oracledb.SessionPool(
        user=db_user,
        password=db_password,
        dsn=db_connect_string,
        min=1,
        max=4,
        increment=1,
        homogeneous=True,
    )


def create_agent_memory(
    db_pool: oracledb.SessionPool,
    schema_policy: SchemaPolicy,
    extract_memories: bool = True,
) -> OracleAgentMemory:
    oci_common_args = load_oci_common_args()

    embedding_model = os.environ.get(
        "OCI_EMBEDDING_MODEL",
        "oci/cohere.embed-v4.0",
    )
    llm_model = os.environ.get(
        "OCI_LLM_MODEL",
        "oci/xai.grok-4.20-reasoning",
    )

    embedder = Embedder(
        model=embedding_model,
        **oci_common_args,
    )

    llm = Llm(
        model=llm_model,
        **oci_common_args,
    )

    table_name_prefix = os.environ.get(
        "ORACLE_MEMORY_TABLE_NAME_PREFIX",
        "T_LANGGRAPH_MEM_",
    )

    return OracleAgentMemory(
        connection=db_pool,
        embedder=embedder,
        llm=llm,
        extract_memories=extract_memories,
        schema_policy=schema_policy,
        table_name_prefix=table_name_prefix,
    )


def create_langgraph_llm() -> ChatLiteLLM:
    oci_common_args = load_oci_common_args()

    llm_model = os.environ.get(
        "OCI_LLM_MODEL",
        "oci/xai.grok-4.20-reasoning",
    )

    return ChatLiteLLM(
        model=llm_model,
        temperature=0,
        model_kwargs=oci_common_args,
    )


def print_messages(messages: list) -> None:
    for message in messages:
        print(f"[{message.role}] {message.content}")


db_pool = create_db_pool()
agent_id = os.environ.get("AGENT_ID", "support_agent")
user_id = os.environ.get("USER_ID", "user_123")

agent_memory = create_agent_memory(
    db_pool=db_pool,
    schema_policy=get_schema_policy(),
    extract_memories=True,
)

langgraph_llm = create_langgraph_llm()


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

    if not results:
        return "No relevant memory found."

    return "\n".join(result.content for result in results)


def _latest_user_message(state: MessagesState) -> str:
    for message in reversed(state["messages"]):
        if getattr(message, "type", None) == "human":
            return str(message.content)
        if getattr(message, "role", None) == "user":
            return str(message.content)
    return ""


def _build_memory_context(query: str) -> str:
    results = agent_memory.search(
        query=query,
        user_id=user_id,
        agent_id=agent_id,
        max_results=3,
        record_types=["memory"],
    )

    memory_context = "\n".join(f"- {result.content}" for result in results)

    return memory_context or "- No relevant memory found."


def run_react_agent_demo() -> None:
    print("\n==============================")
    print("Prebuilt ReAct Agent Demo")
    print("==============================")

    react_agent = create_agent(
        model=langgraph_llm,
        tools=[search_memory],
        system_prompt=(
            "You are a support agent. When the user asks about durable facts from "
            "prior sessions, call the search_memory tool before answering."
        ),
    )

    react_memory_thread = agent_memory.create_thread(
        thread_id="langgraph_react_memory_demo",
        user_id=user_id,
        agent_id=agent_id,
    )

    react_session_1 = react_agent.invoke(
        {
            "messages": [
                HumanMessage(
                    content=(
                        "I am John, a Python developer and I need help debugging "
                        "a payment service."
                    )
                )
            ]
        }
    )

    react_assistant_reply = react_session_1["messages"][-1].content

    print("\nReact session 1 assistant reply:")
    print(react_assistant_reply)

    react_memory_thread.add_messages(
        [
            {
                "role": "user",
                "content": (
                    "I am John, a Python developer and I need help debugging "
                    "a payment service."
                ),
            },
            {
                "role": "assistant",
                "content": react_assistant_reply,
            },
        ]
    )

    react_memory_thread.add_memory("The user is John, a Python developer.")

    react_memory_thread = agent_memory.get_thread("langgraph_react_memory_demo")

    react_session_2 = react_agent.invoke(
        {
            "messages": [
                HumanMessage(content="Who am I?")
            ]
        }
    )

    react_remembered_reply = react_session_2["messages"][-1].content

    print("\nReact session 2 assistant reply:")
    print(react_remembered_reply)


def run_custom_flow_demo() -> None:
    print("\n==============================")
    print("Custom StateGraph Flow Demo")
    print("==============================")

    def call_model(state: MessagesState):
        query = _latest_user_message(state)
        memory_context = _build_memory_context(query)

        response = langgraph_llm.invoke(
            [
                SystemMessage(
                    content=(
                        "You are a support agent. Use the durable memory below when it is "
                        "relevant to the current user request.\n\n"
                        f"Durable memory:\n{memory_context}"
                    )
                ),
                *state["messages"],
            ]
        )

        return {"messages": [response]}

    builder = StateGraph(MessagesState)
    builder.add_node("call_model", call_model)
    builder.add_edge(START, "call_model")
    builder.add_edge("call_model", END)

    flow_graph = builder.compile()

    flow_memory_thread = agent_memory.create_thread(
        thread_id="langgraph_flow_memory_demo",
        user_id=user_id,
        agent_id=agent_id,
    )

    flow_session_1 = flow_graph.invoke(
        {
            "messages": [
                HumanMessage(
                    content=(
                        "I am John, a Python developer and I need help debugging "
                        "a payment service."
                    )
                )
            ]
        }
    )

    flow_assistant_reply = flow_session_1["messages"][-1].content

    print("\nFlow session 1 assistant reply:")
    print(flow_assistant_reply)

    flow_memory_thread.add_messages(
        [
            {
                "role": "user",
                "content": (
                    "I am John, a Python developer and I need help debugging "
                    "a payment service."
                ),
            },
            {
                "role": "assistant",
                "content": flow_assistant_reply,
            },
        ]
    )

    flow_memory_thread.add_memory("The user is John, a Python developer.")

    flow_memory_thread = agent_memory.get_thread("langgraph_flow_memory_demo")

    print("\nMessages stored in Oracle:")
    print_messages(flow_memory_thread.get_messages())

    flow_session_2 = flow_graph.invoke(
        {
            "messages": [
                HumanMessage(content="Who am I?")
            ]
        }
    )

    flow_remembered_reply = flow_session_2["messages"][-1].content

    print("\nFlow session 2 assistant reply:")
    print(flow_remembered_reply)


def run_manual_memory_demo() -> None:
    print("\n==============================")
    print("Manual Memory Demo")
    print("==============================")

    manual_agent_memory = create_agent_memory(
        db_pool=db_pool,
        schema_policy=SchemaPolicy.REQUIRE_EXISTING,
        extract_memories=False,
    )

    manual_memory_thread = manual_agent_memory.create_thread(
        thread_id="langgraph_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.")

    manual_results = manual_agent_memory.search(
        query="code review preference",
        user_id=user_id,
        agent_id=agent_id,
        max_results=3,
        record_types=["memory"],
    )

    print("\nManual memory search results:")
    for result in manual_results:
        print(f"- [{result.record.record_type}] {result.content}")


if __name__ == "__main__":
    run_react_agent_demo()
    run_custom_flow_demo()
    run_manual_memory_demo()

7. 実行

実行します。

uv run python try_agent_memory_langgraph_oci.py

実際の実行結果は以下です。

==============================
Prebuilt ReAct Agent Demo
==============================

React session 1 assistant reply:
Hi John, happy to help debug your Python payment service! Please share more details—like any error messages, relevant code snippets, logs, or the specific issue you're seeing (e.g., integration with a gateway like Stripe/PayPal, transaction failures, etc.)—and I'll dive in.

React session 2 assistant reply:
Based on our prior conversations, you're John, a Python developer who was looking for help debugging a payment service. How can I assist you today?

==============================
Custom StateGraph Flow Demo
==============================

Flow session 1 assistant reply:
Hello John! Thanks for confirming—I'm here to help you debug the payment service.

To get started efficiently, could you share:
- The specific issue or error you're seeing (e.g. traceback, unexpected behavior, logs)?
- Any relevant code snippets (especially around payment processing, API calls, or transaction handling)?
- What you've already tried?

The more details you provide, the faster we can isolate and fix it.

Messages stored in Oracle:
[user] I am John, a Python developer and I need help debugging a payment service.
[assistant] Hello John! Thanks for confirming—I'm here to help you debug the payment service.

To get started efficiently, could you share:
- The specific issue or error you're seeing (e.g. traceback, unexpected behavior, logs)?
- Any relevant code snippets (especially around payment processing, API calls, or transaction handling)?
- What you've already tried?

The more details you provide, the faster we can isolate and fix it.

Flow session 2 assistant reply:
You are **John**, a Python developer who needs help debugging a payment service.

How can I assist you with that today?

==============================
Manual Memory Demo
==============================

Manual memory search results:
- [memory] The user prefers concise code reviews.
- [memory] The user is John, a Python developer.
- [memory] The user is John, a Python developer.

今回のサンプルでは、Prebuilt ReAct Agent と Custom StateGraph Flow の両方で、以下の durable memory を保存しています。

The user is John, a Python developer.

そのため、最後の Manual Memory Demo の検索結果では、手動で追加した memory に加えて、前の2つの demo で保存した John に関する memory も返っています。

- [memory] The user prefers concise code reviews.
- [memory] The user is John, a Python developer.
- [memory] The user is John, a Python developer.

これは同じ user_id と agent_id を使って複数の demo を続けて実行しているためです。

Prebuilt ReAct Agent では、search_memory tool を agent に渡しています。

react_agent = create_agent(
    model=langgraph_llm,
    tools=[search_memory],
    system_prompt=(
        "You are a support agent. When the user asks about durable facts from "
        "prior sessions, call the search_memory tool before answering."
    ),
)

Custom Flow では、call_model node の中で Agent Memory を検索し、検索結果を system message に含めています。

query = _latest_user_message(state)
memory_context = _build_memory_context(query)

SystemMessage(
    content=(
        "You are a support agent. Use the durable memory below when it is "
        "relevant to the current user request.\n\n"
        f"Durable memory:\n{memory_context}"
    )
)

このように、Agent Memory と LangGraph を組み合わせることで、過去セッションで保存した durable memory を後続の agent flow で利用できます。

8. DB に作成されたテーブルを確認する

必要に応じて、DB 側に作成されたオブジェクトも確認できます。

docker exec -it oracle-free-lite sqlplus dmuser/"<your-app-user-password>"@FREEPDB1

テーブル一覧を確認します。

SELECT table_name
FROM user_tables
WHERE table_name LIKE 'T_LANGGRAPH_MEM_%'
ORDER BY table_name;

実際の実行結果は以下です。

TABLE_NAME
--------------------------------------------------------------------------------
T_LANGGRAPH_MEM_ACTOR_PROFILE
T_LANGGRAPH_MEM_MEMORY
T_LANGGRAPH_MEM_MESSAGE
T_LANGGRAPH_MEM_ORACLEAGENTMEMORY_SCHEMA_META
T_LANGGRAPH_MEM_RECORD_CHUNKS
T_LANGGRAPH_MEM_THREAD

6 rows selected.

T_LANGGRAPH_MEM_ から始まる以下のテーブルが作成されていることを確認できました。

テーブル名 内容
T_LANGGRAPH_MEM_ACTOR_PROFILE actor profile 管理用
T_LANGGRAPH_MEM_MEMORY durable memory 保存用
T_LANGGRAPH_MEM_MESSAGE thread message 保存用
T_LANGGRAPH_MEM_ORACLEAGENTMEMORY_SCHEMA_META schema metadata 管理用
T_LANGGRAPH_MEM_RECORD_CHUNKS search 用 record chunk 管理用
T_LANGGRAPH_MEM_THREAD thread 管理用

確認後、SQL*Plus を終了します。

exit

9. クリーンアップ

コンテナを停止します。

docker stop oracle-free-lite
docker rm oracle-free-lite

DB ファイルも削除する場合は Volume も削除します。

docker volume rm OracleDBData

まとめ

本記事では、Oracle AI Database Free Lite、Oracle Agent Memory、LangGraph を組み合わせて、durable memory を使う AI Agent flow を試しました。

今回できたことは以下です。

  • Oracle AI Database Free Lite をローカルで起動
  • Agent Memory 用の DB ユーザーを作成
  • OCI Cohere Embed v4 を Embedding Model として利用
  • OCI Grok 4.20 Reasoning を LLM として利用
  • LangGraph の prebuilt ReAct agent から Agent Memory を検索
  • LangGraph の StateGraph(MessagesState) custom flow から Agent Memory を検索
  • 1回目の会話で durable memory を保存
  • 2回目の会話で Oracle Database から durable memory を検索
  • Agent の応答に memory を反映
  • extract_memories=False による手動 memory 追加も確認
  • Agent Memory 用のテーブルが Oracle Database 側に作成されていることを確認

今回のポイントは、LangGraph で利用する ChatLiteLLM に OCI 認証情報を渡す部分です。

return ChatLiteLLM(
    model=llm_model,
    temperature=0,
    model_kwargs=oci_common_args,
)

Embedder や Llm には **oci_common_args を渡しますが、ChatLiteLLM では model_kwargs 経由で LiteLLM に渡す必要がありました。

Agent Memory と LangGraph を組み合わせると、単なる会話履歴ではなく、ユーザーごとの重要な情報を Oracle Database に永続化し、後続の AI Agent flow で再利用できます。

参考記事

本記事の作成にあたり、以下の公式ドキュメントを参考にしました。

3
1
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
3
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?