239 lines
12 KiB
Python
239 lines
12 KiB
Python
import copy
|
|
|
|
import pytest
|
|
from sqlmodel import SQLModel, Session, create_engine, select
|
|
|
|
from app.agents.memory import (content_hash, frozen_candidates, memories_text,
|
|
memory_candidates, recall_report, record_injected_memories,
|
|
select_memories)
|
|
from app.agents.runner import (_generation_memories, _memory_model_call,
|
|
_runtime_segment_tools, human_review)
|
|
from app.agents.runtime import run_agent
|
|
from app.agents.snapshot import freeze_agent_snapshot
|
|
from app.models import Agent, AgentSession, AppUser, Campaign, Memory, MemoryGrant, PendingMemory, Segment
|
|
|
|
|
|
@pytest.fixture
|
|
def db():
|
|
engine = create_engine("sqlite://")
|
|
SQLModel.metadata.create_all(engine)
|
|
with Session(engine) as session:
|
|
yield session
|
|
engine.dispose()
|
|
|
|
|
|
def _item(identifier, content, layer="L2", **extra):
|
|
return {"id": identifier, "content": content, "layer": layer, **extra}
|
|
|
|
|
|
def _scope(db):
|
|
owner = AppUser(username="owner", role="editor", organization_id=1)
|
|
user = AppUser(username="learner", role="editor", organization_id=1)
|
|
other = AppUser(username="other", role="editor", organization_id=1)
|
|
db.add_all([owner, user, other])
|
|
db.commit()
|
|
agent = Agent(name="训练营主笔", owner_user_id=owner.id, organization_id=1)
|
|
db.add(agent)
|
|
db.commit()
|
|
return agent, owner, user, other
|
|
|
|
|
|
def test_chinese_relevance_keeps_foundations_but_not_unrelated_experiences():
|
|
rows = [_item(1, "不要编造事实", "L1"), _item(2, "用口语表达", "L3"),
|
|
_item(3, "先介绍分数的分母再讲约分", "L2"), _item(4, "售后沟通要先核对订单", "L4")]
|
|
selected = select_memories(rows, "数学分数约分练习")
|
|
assert {item["id"] for item in selected} == {1, 2, 3}
|
|
related = next(item for item in selected if item["id"] == 3)
|
|
assert "约分" in related["recall_reason"]
|
|
assert related["content_hash"] == content_hash(rows[2]["content"])
|
|
assert [item["id"] for item in select_memories(rows, "")] == [1, 2]
|
|
|
|
|
|
def test_whole_memory_budget_preserves_negation_and_deduplicates():
|
|
oversized = "示例说明" * 100 + "，以上案例不能当作真实用户经历。"
|
|
rows = [_item(1, oversized, "L1"), _item(2, "禁止编造事实", "L1"),
|
|
_item(3, " 禁止编造事实 ", "L1"), _item(4, "讲解使用大白话", "L3")]
|
|
selected = select_memories(rows, "", max_chars=130, max_items=2)
|
|
assert len(selected) == 2
|
|
assert all(item["content"] != oversized for item in selected)
|
|
assert len({item["content"] for item in selected}) == 2
|
|
assert any(item["content"] == "禁止编造事实" for item in selected)
|
|
assert len(memories_text(selected).split("\n", 3)[-1]) <= 130
|
|
assert select_memories(rows, "", max_items=0) == []
|
|
|
|
|
|
def test_candidates_enforce_user_tenant_grants_and_canvas_agent_owner(db):
|
|
agent, owner, user, other = _scope(db)
|
|
rows = [Memory(agent_id=agent.id, owner_user_id=person.id, organization_id=1,
|
|
content=content, visibility=visibility)
|
|
for person, content, visibility in [(user, "本人偏好", "private"),
|
|
(owner, "主笔公共经验", "private"), (other, "他人私有偏好", "private"),
|
|
(other, "明确授权", "shared"), (other, "同组织全局", "global")]]
|
|
foreign = Memory(agent_id=agent.id, owner_user_id=other.id, organization_id=2,
|
|
content="其他组织", visibility="global")
|
|
db.add_all(rows + [foreign])
|
|
db.commit()
|
|
db.add(MemoryGrant(memory_id=rows[3].id, user_id=user.id, permission="use"))
|
|
db.commit()
|
|
direct = memory_candidates(db, agent, user)
|
|
assert {item["content"] for item in direct} == {"本人偏好", "明确授权", "同组织全局"}
|
|
shared = memory_candidates(db, agent, user, allow_agent_scope=True, allowed_agent_ids={agent.id})
|
|
assert {item["content"] for item in shared} == {"本人偏好", "明确授权", "同组织全局", "主笔公共经验"}
|
|
user.role = "admin"
|
|
assert "其他组织" not in {item["content"] for item in memory_candidates(db, agent, user)}
|
|
assert "他人私有偏好" not in {item["content"] for item in memory_candidates(db, agent, user)}
|
|
|
|
|
|
def test_read_only_grant_does_not_authorize_generation(db):
|
|
agent, owner, user, _ = _scope(db)
|
|
memory = Memory(agent_id=agent.id, owner_user_id=owner.id, content="只能查看", visibility="shared")
|
|
db.add(memory)
|
|
db.commit()
|
|
db.add(MemoryGrant(memory_id=memory.id, user_id=user.id, permission="read"))
|
|
db.commit()
|
|
assert memory_candidates(db, agent, user) == []
|
|
|
|
|
|
def test_old_index_state_is_readable_and_freezing_does_not_count_usage(db):
|
|
agent, owner, _, _ = _scope(db)
|
|
memory = Memory(agent_id=agent.id, owner_user_id=owner.id, content="不要夸大效果", index_status="needs_reindex")
|
|
db.add(memory)
|
|
db.commit()
|
|
frozen = freeze_agent_snapshot(db, agent, user=owner)
|
|
assert [item["id"] for item in frozen["memories"]] == [memory.id]
|
|
assert memory.usage_count == 0
|
|
assert memory.last_used_at == ""
|
|
|
|
|
|
def test_frozen_content_unchanged_after_edit_and_legacy_private_memory_filtered(db):
|
|
agent, owner, user, other = _scope(db)
|
|
original = Memory(agent_id=agent.id, owner_user_id=owner.id, content="原版本：不得报低价")
|
|
private = Memory(agent_id=agent.id, owner_user_id=other.id, content="其他用户的秘密")
|
|
db.add_all([original, private])
|
|
db.commit()
|
|
frozen = {"memories": [{"id": original.id, "layer": "L3", "content": original.content},
|
|
{"id": private.id, "layer": "L3", "content": private.content}]}
|
|
unchanged = copy.deepcopy(frozen)
|
|
original.content = "新版本：只说已核实的价格"
|
|
db.add(original)
|
|
db.commit()
|
|
result = frozen_candidates(db, agent, frozen, user, allow_agent_scope=True, allowed_agent_ids={agent.id})
|
|
assert [item["content"] for item in result] == ["原版本：不得报低价"]
|
|
assert frozen == unchanged
|
|
|
|
|
|
def test_usage_counts_actual_injection_once_per_session(db):
|
|
agent, owner, _, _ = _scope(db)
|
|
first = Memory(agent_id=agent.id, owner_user_id=owner.id, content="先解释分数")
|
|
unused = Memory(agent_id=agent.id, owner_user_id=owner.id, content="不要夸大效果")
|
|
db.add_all([first, unused])
|
|
db.commit()
|
|
selected = select_memories(memory_candidates(db, agent, owner), "分数")
|
|
row = AgentSession(agent_id=agent.id)
|
|
db.add(row)
|
|
db.commit()
|
|
rendered = memories_text([item for item in selected if item["id"] == first.id])
|
|
record_injected_memories(db, row, selected, [{"content": rendered}])
|
|
record_injected_memories(db, row, selected, [{"content": rendered}])
|
|
db.refresh(first)
|
|
db.refresh(unused)
|
|
assert first.usage_count == 1 and first.last_used_at
|
|
assert unused.usage_count == 0
|
|
new_row = AgentSession(agent_id=agent.id)
|
|
db.add(new_row)
|
|
db.commit()
|
|
record_injected_memories(db, new_row, selected, [{"content": rendered}])
|
|
db.refresh(first)
|
|
assert first.usage_count == 2
|
|
|
|
|
|
def test_generation_primary_fusion_tools_share_selected_memories_without_mutating_snapshot(db):
|
|
agent, owner, user, _ = _scope(db)
|
|
fused = Agent(name="融合讲师", owner_user_id=owner.id)
|
|
db.add(fused)
|
|
db.commit()
|
|
agent.fusion_agent_ids = [fused.id]
|
|
primary_mem = Memory(agent_id=agent.id, owner_user_id=owner.id, layer="L2", content="分数讲清分母")
|
|
fused_mem = Memory(agent_id=fused.id, owner_user_id=owner.id, layer="L2", content="分数先做图示")
|
|
wrong = Memory(agent_id=fused.id, owner_user_id=owner.id, layer="L2", content="售后先核对订单")
|
|
db.add_all([agent, primary_mem, fused_mem, wrong])
|
|
db.commit()
|
|
campaign = Campaign(name="数学", owner_user_id=user.id, canvas_snapshot={"nodes": [
|
|
{"agent_id": agent.id}, {"agent_id": fused.id}]})
|
|
db.add(campaign)
|
|
db.commit()
|
|
segment = Segment(campaign_id=campaign.id, agent_id=agent.id, title="分数入门")
|
|
db.add(segment)
|
|
db.commit()
|
|
frozen = freeze_agent_snapshot(db, agent, user=user, allow_agent_scope=True,
|
|
allowed_agent_ids={agent.id, fused.id})
|
|
unchanged = copy.deepcopy(frozen)
|
|
selected, reports = _generation_memories(db, agent, segment, campaign, frozen)
|
|
assert [item["id"] for item in selected["memories"]] == [primary_mem.id]
|
|
assert [item["id"] for item in selected["fusions"][0]["memories"]] == [fused_mem.id]
|
|
assert reports[1]["items"][0]["reason"]
|
|
assert frozen == unchanged
|
|
tools, _ = _runtime_segment_tools(db, segment, agent, selected, [], [], [], [])
|
|
read = next(tool for tool in tools if tool.name == "read_agent_memory")
|
|
payload = read.handler(None, {})
|
|
assert payload["memories"][0]["content"] == selected["memories"][0]["content"]
|
|
assert len(payload["memories"]) == 1
|
|
|
|
|
|
def test_model_call_records_usage_and_trace_report_without_additional_ai_call(db, monkeypatch):
|
|
agent, owner, _, _ = _scope(db)
|
|
memory = Memory(agent_id=agent.id, owner_user_id=owner.id, content="讲解用大白话")
|
|
db.add(memory)
|
|
db.commit()
|
|
selected = select_memories(memory_candidates(db, agent, owner), "课程")
|
|
calls = []
|
|
def model_call(*args, **kwargs):
|
|
calls.append(1)
|
|
return {"status": "ok", "text": "正文", "tool_calls": []}
|
|
monkeypatch.setattr("app.agents.runtime.chat", model_call)
|
|
result = run_agent(db, run_id=0, campaign_id=0, segment_id=0, node_id="", agent=agent,
|
|
provider_id=1, model="test", phase="production",
|
|
messages=[{"role": "user", "content": memories_text(selected)}],
|
|
model_call=_memory_model_call(selected, "production"),
|
|
prompt_meta={"memory_recall": [recall_report(agent, selected)]})
|
|
db.refresh(memory)
|
|
assert len(calls) == 1 and memory.usage_count == 1
|
|
row = db.get(AgentSession, result["session_id"])
|
|
assert row.result_meta["prompt"]["memory_recall"][0]["items"][0]["id"] == memory.id
|
|
|
|
|
|
def test_existing_snapshot_does_not_acquire_new_fusion_memories(db):
|
|
agent, owner, _, _ = _scope(db)
|
|
fused = Agent(name="后新增融合 Agent", owner_user_id=owner.id)
|
|
db.add(fused)
|
|
db.commit()
|
|
agent.fusion_agent_ids = [fused.id]
|
|
db.add_all([agent, Memory(agent_id=fused.id, owner_user_id=owner.id, content="新的稳定偏好")])
|
|
db.commit()
|
|
campaign = Campaign(name="课程", owner_user_id=owner.id)
|
|
segment = Segment(campaign_id=1, agent_id=agent.id, title="段落")
|
|
selected, _ = _generation_memories(db, agent, segment, campaign, {"id": agent.id, "memories": []})
|
|
assert selected["fusions"][0]["memories"] == []
|
|
|
|
|
|
def test_rejection_pending_memory_dedup_is_user_scoped(db):
|
|
agent, owner, user, _ = _scope(db)
|
|
for person in [owner, user]:
|
|
campaign = Campaign(name="课程", owner_user_id=person.id)
|
|
db.add(campaign)
|
|
db.commit()
|
|
segment = Segment(campaign_id=campaign.id, agent_id=agent.id, title="段落", status="human_review")
|
|
db.add(segment)
|
|
db.commit()
|
|
human_review(db, segment.id, "reject", "不要用术语")
|
|
pending = db.exec(select(PendingMemory)).all()
|
|
assert {item.owner_user_id for item in pending} == {owner.id, user.id}
|
|
|
|
|
|
def test_relevant_correction_is_not_crowded_out_by_many_unrelated_preferences():
|
|
rows = [_item(i, f"口播习惯偏好编号{i}", "L3") for i in range(1, 80)]
|
|
rows.append(_item(100, "分数约分要先检查分母", "L2"))
|
|
selected = select_memories(rows, "数学分数约分", max_items=12)
|
|
assert len(selected) <= 12
|
|
assert selected[0]["id"] == 100
|