Integrate AgentScope with Long-Term Memory

Step 1: Install dependencies

1. Create a project directory and a Python virtual environment.

   mkdir -p agentscope-adbpg-demo && cd agentscope-adbpg-demo
   python3 -m venv .venv
   source .venv/bin/activate

2. Install the required packages.

   pip install agentscope python-dotenv

   Important

   The agentscope version must be earlier than 2.0.

Step 2: Configure environment variables

Create a .env file with the following environment variables.

DASHSCOPE_API_KEY=<your-dashscope-api-key>
ADBPG_BASE_URL=https://api-longmemory-cn-chengdu.opentrust.net
ADBPG_API_KEY=<your-adbpg-longmemory-api-key>

Step 3: Write the long-term memory adapter module

Create a file named adbpg_long_term_memory.py. This module extends the LongTermMemoryBase base class of AgentScope and connects to the AnalyticDB for PostgreSQL long-term memory service through REST APIs.

# -*- coding: utf-8 -*-
"""Long-term memory implementation backed by AnalyticDB for PostgreSQL."""
from typing import Any

import aiohttp

from agentscope.memory._long_term_memory._long_term_memory_base import (
 LongTermMemoryBase,
)
from agentscope.message import Msg, TextBlock
from agentscope.tool import ToolResponse


class ADBPGLongTermMemory(LongTermMemoryBase):
 """Connect to AnalyticDB for PostgreSQL long-term memory service via REST API."""

 def __init__(
     self,
     base_url: str,
     api_key: str,
     user_name: str,
     agent_name: str | None = None,
 ) -> None:
     super().__init__()
     self.base_url = base_url.rstrip("/")
     self.api_key = api_key
     self.user_id = user_name
     self.agent_id = agent_name

 def _headers(self) -> dict:
     return {
         "Authorization": f"Token {self.api_key}",
         "Content-Type": "application/json",
     }

 async def _post(self, path: str, data: dict) -> dict:
     async with aiohttp.ClientSession() as session:
         async with session.post(
             f"{self.base_url}{path}", json=data, headers=self._headers()
         ) as resp:
             resp.raise_for_status()
             return await resp.json()

 async def record(self, msgs: list[Msg | None], **kwargs: Any) -> Any:
     messages = [
         {"role": m.role, "content": str(m.content)} for m in msgs if m
     ]
     data = {"messages": messages, "user_id": self.user_id}
     if self.agent_id:
         data["agent_id"] = self.agent_id
     return await self._post("/v3/memories/add/", data)

 async def retrieve(
     self, msg: Msg | list[Msg] | None, limit: int = 5, **kwargs: Any
 ) -> str:
     if isinstance(msg, Msg):
         msg = [msg]
     query = " ".join(m.get_text_content() for m in msg if m)
     data = {
         "query": query,
         "filters": {"user_id": self.user_id},
         "top_k": limit,
     }
     if self.agent_id:
         data["filters"]["agent_id"] = self.agent_id
     result = await self._post("/v3/memories/search/", data)
     return "\n".join(
         r.get("memory", "") for r in result.get("results", [])
     )

 async def record_to_memory(
     self, thinking: str, content: list[str], **kwargs: Any
 ) -> ToolResponse:
     text = "\n".join(([thinking] if thinking else []) + content)
     data = {
         "messages": [{"role": "user", "content": text}],
         "user_id": self.user_id,
     }
     if self.agent_id:
         data["agent_id"] = self.agent_id
     result = await self._post("/v3/memories/add/", data)
     return ToolResponse(
         content=[TextBlock(type="text", text=f"Recorded: {result}")]
     )

 async def retrieve_from_memory(
     self, keywords: list[str], limit: int = 5, **kwargs: Any
 ) -> ToolResponse:
     results = []
     for kw in keywords:
         data = {
             "query": kw,
             "filters": {"user_id": self.user_id},
             "top_k": limit,
         }
         if self.agent_id:
             data["filters"]["agent_id"] = self.agent_id
         r = await self._post("/v3/memories/search/", data)
         results.extend(
             item.get("memory", "") for item in r.get("results", [])
         )
     return ToolResponse(
         content=[
             TextBlock(
                 type="text",
                 text="\n".join(results) or "No relevant memories found",
             )
         ]
     )

Parameters

API reference

Step 4: Write the demo code

Create a file named memory_example.py.

# -*- coding: utf-8 -*-
"""AgentScope + AnalyticDB for PostgreSQL long-term memory demo"""
import asyncio
import os

from dotenv import load_dotenv

from adbpg_long_term_memory import ADBPGLongTermMemory
from agentscope.agent import ReActAgent
from agentscope.formatter import DashScopeChatFormatter
from agentscope.memory import InMemoryMemory
from agentscope.message import Msg
from agentscope.model import DashScopeChatModel
from agentscope.tool import Toolkit

load_dotenv()


async def main() -> None:
 """Run the memory demo."""

 # ── 1. Initialize long-term memory ────────────────────────
 long_term_memory = ADBPGLongTermMemory(
     base_url=os.environ["ADBPG_BASE_URL"],
     api_key=os.environ["ADBPG_API_KEY"],
     user_name="user_123",
     agent_name="Friday",
 )

 print("=== AgentScope + ADBPG Long-Term Memory Demo ===\n")

 # ── 2. Basic memory operations: record and retrieve ───────
 print("1. Record conversation to long-term memory")
 print("-" * 40)
 results = await long_term_memory.record(
     msgs=[
         Msg(role="user", content="Book a hotel for me, preferably a B&B", name="user"),
     ],
 )
 print(f"Record result: {results}\n")

 print("2. Retrieve from long-term memory")
 print("-" * 40)
 memories = await long_term_memory.retrieve(
     msg=[
         Msg(role="user", content="What type of hotel do I like?", name="user"),
     ],
 )
 print(f"Retrieval result: {memories}\n")

 # ── 3. ReActAgent with long-term memory ───────────────────
 print("3. ReActAgent + Long-Term Memory")
 print("-" * 40)

 agent = ReActAgent(
     name="Friday",
     sys_prompt=(
         "You are an assistant named Friday. "
         "If you identify important information about user preferences, "
         "use the `record_to_memory` tool to save it to long-term memory. "
         "If you need to retrieve information from long-term memory, "
         "use the `retrieve_from_memory` tool."
     ),
     model=DashScopeChatModel(
         model_name="qwen-plus",
         api_key=os.environ["DASHSCOPE_API_KEY"],
         stream=False,
     ),
     formatter=DashScopeChatFormatter(),
     toolkit=Toolkit(),
     memory=InMemoryMemory(),
     long_term_memory=long_term_memory,
     long_term_memory_mode="both",
 )

 await agent.memory.clear()

 # Conversation 1: Share preference → Agent records automatically
 msg = Msg(
     role="user",
     content="When I travel to Hangzhou, I prefer staying at B&Bs",
     name="user",
 )
 response = await agent(msg)
 print(f"Agent response: {response.get_text_content()}\n")

 # Conversation 2: Ask about preferences → Agent retrieves automatically
 msg = Msg(role="user", content="What are my preferences?", name="user")
 response = await agent(msg)
 print(f"Agent response: {response.get_text_content()}\n")

 # Conversation 3: Add preference → Agent continues recording
 msg = Msg(role="user", content="I enjoy visiting West Lake", name="user")
 response = await agent(msg)
 print(f"Agent response: {response.get_text_content()}\n")


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

long_term_memory_mode parameter description

Step 5: Run the demo

python memory_example.py

Expected output:

=== AgentScope + ADBPG Long-Term Memory Demo ===

1. Record conversation to long-term memory
----------------------------------------
Record result: {'results': [{'message': 'Memory processing has been queued for background execution', 'status': 'PENDING', ...}]}

2. Retrieve from long-term memory
----------------------------------------
Retrieval result: Prefers staying at B&Bs

3. ReActAgent + Long-Term Memory
----------------------------------------
Agent response: Recorded: You prefer staying at B&Bs when traveling to Hangzhou. Feel free to share more preferences or ask for travel recommendations!

Agent response: Based on your long-term memory, you have one recorded preference:
  You prefer staying at B&Bs when traveling to Hangzhou.

Agent response: Recorded: You enjoy visiting West Lake.
  Combined with previous information, your Hangzhou travel preferences include:
  - Prefer staying at B&Bs
  - Enjoy visiting West Lake