455a48409d
Add Memori (https://github.com/MemoriLabs/Memori) integration for persistent cross-session memory capabilities in both create_agent and create_deep_agent. ## New Files - agent/memori_config.py: MemoriConfig dataclass for configuration - agent/memori_manager.py: MemoriManager for connection and instance management - agent/memori_middleware.py: MemoriMiddleware for memory recall/storage - tests/: Unit tests for Memori components ## Modified Files - agent/agent_config.py: Added enable_memori, memori_semantic_search_top_k, etc. - agent/deep_assistant.py: Integrated MemoriMiddleware into init_agent() - utils/settings.py: Added MEMORI_* environment variables - pyproject.toml: Added memori>=3.1.0 dependency ## Features - Semantic memory search with configurable top-k and threshold - Multi-tenant isolation (entity_id=user, process_id=bot, session_id) - Memory injection into system prompt - Background asynchronous memory augmentation - Graceful degradation when Memori is unavailable 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
121 lines
3.8 KiB
Python
121 lines
3.8 KiB
Python
"""
|
|
Memori 中间件测试
|
|
"""
|
|
|
|
import pytest
|
|
from unittest.mock import Mock, AsyncMock, patch
|
|
from langchain.agents.middleware import AgentState
|
|
from langgraph.runtime import Runtime
|
|
|
|
from agent.memori_middleware import MemoriMiddleware, create_memori_middleware
|
|
from agent.memori_config import MemoriConfig
|
|
|
|
|
|
class TestMemoriMiddleware:
|
|
"""测试 MemoriMiddleware 类"""
|
|
|
|
def test_extract_user_query_empty(self):
|
|
"""测试空状态"""
|
|
config = MemoriConfig(
|
|
enabled=True,
|
|
entity_id="user_123",
|
|
process_id="bot_456",
|
|
)
|
|
manager = Mock()
|
|
middleware = MemoriMiddleware(manager, config)
|
|
|
|
state: AgentState = {"messages": []}
|
|
query = middleware._extract_user_query(state)
|
|
assert query == ""
|
|
|
|
def test_extract_user_query_from_message(self):
|
|
"""测试从消息中提取查询"""
|
|
config = MemoriConfig(
|
|
enabled=True,
|
|
entity_id="user_123",
|
|
process_id="bot_456",
|
|
)
|
|
manager = Mock()
|
|
middleware = MemoriMiddleware(manager, config)
|
|
|
|
# 创建模拟消息
|
|
mock_message = Mock()
|
|
mock_message.content = "What is the weather today?"
|
|
|
|
state: AgentState = {"messages": [mock_message]}
|
|
query = middleware._extract_user_query(state)
|
|
assert query == "What is the weather today?"
|
|
|
|
def test_extract_user_query_from_dict(self):
|
|
"""测试从字典消息中提取查询"""
|
|
config = MemoriConfig(
|
|
enabled=True,
|
|
entity_id="user_123",
|
|
process_id="bot_456",
|
|
)
|
|
manager = Mock()
|
|
middleware = MemoriMiddleware(manager, config)
|
|
|
|
state: AgentState = {"messages": [{"content": "Hello, world!"}]}
|
|
query = middleware._extract_user_query(state)
|
|
assert query == "Hello, world!"
|
|
|
|
def test_format_memories(self):
|
|
"""测试记忆格式化"""
|
|
config = MemoriConfig(
|
|
enabled=True,
|
|
entity_id="user_123",
|
|
process_id="bot_456",
|
|
)
|
|
manager = Mock()
|
|
middleware = MemoriMiddleware(manager, config)
|
|
|
|
memories = [
|
|
{"content": "User likes coffee", "similarity": 0.9, "fact_type": "preference"},
|
|
{"content": "User lives in Tokyo", "similarity": 0.8, "fact_type": "fact"},
|
|
]
|
|
|
|
formatted = middleware._format_memories(memories)
|
|
assert "User likes coffee" in formatted
|
|
assert "User lives in Tokyo" in formatted
|
|
assert "[preference]" in formatted
|
|
assert "[fact]" in formatted
|
|
|
|
def test_format_memories_empty(self):
|
|
"""测试空记忆列表"""
|
|
config = MemoriConfig(
|
|
enabled=True,
|
|
entity_id="user_123",
|
|
process_id="bot_456",
|
|
)
|
|
manager = Mock()
|
|
middleware = MemoriMiddleware(manager, config)
|
|
|
|
formatted = middleware._format_memories([])
|
|
assert formatted == ""
|
|
|
|
def test_create_memori_middleware_disabled(self):
|
|
"""测试创建中间件（禁用状态）"""
|
|
middleware = create_memori_middleware(
|
|
bot_id="bot_456",
|
|
user_identifier="user_123",
|
|
session_id="session_789",
|
|
enabled=False,
|
|
)
|
|
assert middleware is None
|
|
|
|
def test_create_memori_middleware_enabled(self):
|
|
"""测试创建中间件（启用状态）"""
|
|
middleware = create_memori_middleware(
|
|
bot_id="bot_456",
|
|
user_identifier="user_123",
|
|
session_id="session_789",
|
|
enabled=True,
|
|
semantic_search_top_k=10,
|
|
semantic_search_threshold=0.8,
|
|
)
|
|
assert middleware is not None
|
|
assert isinstance(middleware, MemoriMiddleware)
|
|
assert middleware.config.semantic_search_top_k == 10
|
|
assert middleware.config.semantic_search_threshold == 0.8
|