ai-agent/MemoryAgents/backend/tests/test_text2mem.py

import pytest

import app.adapters.text2mem as module
from app.adapters.text2mem import Text2MemAgent
from app.db import MemoryRepository
from app.llm import LLMUnavailable


class OfflineLLM:
    async def chat(self, *_args, **_kwargs):
        raise LLMUnavailable("test offline")


class DisabledEmbedding:
    configured = False


class FailingLLM:
    async def chat(self, *_args, **_kwargs):
        raise RuntimeError("temporary provider failure")


def build_agent(monkeypatch, tmp_path) -> tuple[Text2MemAgent, MemoryRepository]:
    repo = MemoryRepository()
    repo.path = tmp_path / "memory.json"
    monkeypatch.setattr(module, "embedding_client", DisabledEmbedding())
    agent = Text2MemAgent()
    agent.repo = repo
    agent.llm = OfflineLLM()
    return agent, repo


@pytest.mark.asyncio
async def test_text2mem_explicit_write(monkeypatch, tmp_path):
    agent, repo = build_agent(monkeypatch, tmp_path)

    response = await agent.chat("记住：我偏好使用 Python 和 FastAPI")

    assert response.system == "text2mem"
    assert response.memory_events[0]["status"] == "created"
    memories = await repo.list_memories("text2mem")
    assert memories[0].content.startswith("记住")
    assert memories[0].memory_type == "semantic"


@pytest.mark.asyncio
async def test_text2mem_writes_all_four_memory_types_and_does_not_store_question(
    monkeypatch,
    tmp_path,
):
    agent, repo = build_agent(monkeypatch, tmp_path)
    statements = [
        "记住：我偏好使用 Python 和 FastAPI，代码要求严格类型注解。",
        "我正在开发一个企业内部智能知识库问答项目，使用通义千问和 RAG。",
        "更新一下：检索重排功能已经完成 75%，下一步补充效果评测集。",
        "上次部署测试环境时遇到向量索引构建超时，最后通过分批导入文档解决。",
        "以后修复代码时按这个流程：先复现问题，再定位原因，修改后运行测试。",
    ]
    for statement in statements:
        await agent.chat(statement)

    memories = await repo.list_memories("text2mem")
    assert len(memories) == 5
    assert [item.memory_type for item in memories].count("semantic") == 2
    assert {item.memory_type for item in memories} == {
        "semantic",
        "episodic",
        "procedural",
        "task-state",
    }

    response = await agent.chat("上次部署遇到了什么问题？")

    assert len(await repo.list_memories("text2mem")) == 5
    assert response.memory_events == []
    assert any("向量索引构建超时" in item.content for item in response.memory_context)
    assert all(item.memory_type == "episodic" for item in response.memory_context)


def test_text2mem_write_classifier_rejects_normal_questions():
    assert Text2MemAgent._should_encode("我的项目使用什么数据库？") is False
    assert Text2MemAgent._should_encode("上次部署遇到了什么问题？") is False
    assert Text2MemAgent._should_encode("记住：项目数据库使用 PostgreSQL。") is True


@pytest.mark.asyncio
async def test_text2mem_falls_back_when_configured_model_call_fails(monkeypatch, tmp_path):
    agent, _ = build_agent(monkeypatch, tmp_path)
    agent.llm = FailingLLM()

    response = await agent.chat("记住：我偏好使用 Python 和 FastAPI。")

    assert response.mode == "local-fallback"
    assert "模型调用失败" in response.answer
    assert response.memory_events[0]["status"] == "created"