Across sessions and conversations
Long-term memory 建立在Lang Graph stores,存储的形式:JSON documents organized by namespace and key
Usage#
为了给一个agent添加memory功能,在使用create_agent时create a store and pass it to agent
1
2
3
4
5
6
7
8
9
10
11
12
| from langchain.agents import create_agent
from langchain_core.runnables import Runnable
from langgraph.store.memory import InMemoryStore
# InMemoryStore saves data to an in-memory dictionary. Use a DB-backed store in production use.
store = InMemoryStore()
agent: Runnable = create_agent(
"claude-sonnet-4-6",
tools=[],
store=store,
)
|
1
2
3
4
5
6
7
8
9
10
11
12
13
| from langchain.agents import create_agent
from langchain_core.runnables import Runnable
from langgraph.store.postgres import PostgresStore # type: ignore[import-not-found]
DB_URI = "postgresql://postgres:postgres@localhost:5432/postgres?sslmode=disable"
with PostgresStore.from_conn_string(DB_URI) as store:
store.setup()
agent: Runnable = create_agent(
"claude-sonnet-4-6",
tools=[],
store=store,
)
|
tool可以通过runtime.store来访问并且写入store
Memory storage#
每个记忆都被组织在自定义的namespace(类似于文件夹)和一个独特的key(类似于文件名)之下
随后可以通过内容过滤器实现跨命名空间的搜索
InmemoryStore
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
| from collections.abc import Sequence
from langgraph.store.base import IndexConfig
from langgraph.store.memory import InMemoryStore
def embed(texts: Sequence[str]) -> list[list[float]]:
# Replace with an actual embedding function or LangChain embeddings object
return [[1.0, 2.0] for _ in texts]
# InMemoryStore saves data to an in-memory dictionary. Use a DB-backed store in production use.
store = InMemoryStore(index=IndexConfig(embed=embed, dims=2))
user_id = "my-user"
application_context = "chitchat"
namespace = (user_id, application_context)
store.put(
namespace,
"a-memory",
{
"rules": [
"User likes short, direct language",
"User only speaks English & python",
],
"my-key": "my-value",
},
)
# get the "memory" by ID
item = store.get(namespace, "a-memory")
# search for "memories" within this namespace, filtering on content equivalence, sorted by vector similarity
items = store.search(
namespace, filter={"my-key": "my-value"}, query="language preferences"
)
|
PostgreSQL
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
| from collections.abc import Sequence
from langgraph.store.base import IndexConfig
from langgraph.store.postgres import PostgresStore # type: ignore[import-not-found]
def embed(texts: Sequence[str]) -> list[list[float]]:
# Replace with an actual embedding function or LangChain embeddings object
return [[1.0, 2.0] for _ in texts]
DB_URI = "postgresql://postgres:postgres@localhost:5432/postgres?sslmode=disable"
with PostgresStore.from_conn_string(
DB_URI,
index=IndexConfig(embed=embed, dims=2), # type: ignore[arg-type]
) as store:
store.setup()
user_id = "my-user"
application_context = "chitchat"
namespace = (user_id, application_context)
store.put(
namespace,
"a-memory",
{
"rules": [
"User likes short, direct language",
"User only speaks English & python",
],
"my-key": "my-value",
},
)
item = store.get(namespace, "a-memory")
items = store.search(
namespace, filter={"my-key": "my-value"}, query="language preferences"
)
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
| from dataclasses import dataclass
from langchain.agents import create_agent
from langchain.tools import ToolRuntime, tool
from langchain_core.runnables import Runnable
from langgraph.store.postgres import PostgresStore # type: ignore[import-not-found]
@dataclass
class Context:
user_id: str
DB_URI = "postgresql://postgres:postgres@localhost:5432/postgres?sslmode=disable"
with PostgresStore.from_conn_string(DB_URI) as store:
store.setup()
store.put(("users",), "user_123", {"name": "John Smith", "language": "English"})
@tool
def get_user_info(runtime: ToolRuntime[Context]) -> str:
"""Look up user info."""
assert runtime.store is not None
user_info = runtime.store.get(("users",), runtime.context.user_id)
return str(user_info.value) if user_info else "Unknown user"
agent: Runnable = create_agent(
"claude-sonnet-4-6",
tools=[get_user_info],
store=store,
context_schema=Context,
)
result = agent.invoke(
{"messages": [{"role": "user", "content": "look up user information"}]},
context=Context(user_id="user_123"),
)
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
| from dataclasses import dataclass
from langchain.agents import create_agent
from langchain.tools import ToolRuntime, tool
from langchain_core.runnables import Runnable
from langgraph.store.postgres import PostgresStore # type: ignore[import-not-found]
from typing_extensions import TypedDict
@dataclass
class Context:
user_id: str
class UserInfo(TypedDict):
name: str
@tool
def save_user_info(user_info: UserInfo, runtime: ToolRuntime[Context]) -> str:
"""Save user info."""
assert runtime.store is not None
runtime.store.put(("users",), runtime.context.user_id, dict(user_info))
return "Successfully saved user info."
DB_URI = "postgresql://postgres:postgres@localhost:5432/postgres?sslmode=disable"
with PostgresStore.from_conn_string(DB_URI) as store:
store.setup()
agent: Runnable = create_agent(
"claude-sonnet-4-6",
tools=[save_user_info],
store=store,
context_schema=Context,
)
agent.invoke(
{"messages": [{"role": "user", "content": "My name is John Smith"}]},
context=Context(user_id="user_123"),
)
|