189 lines
9.7 KiB
Python
189 lines
9.7 KiB
Python
"""按任务召回已确认的文字记忆，不增加模型调用或改变历史快照正文。"""
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import re
|
||
import unicodedata
|
||
from typing import Any
|
||
|
||
from sqlalchemy import update
|
||
from sqlmodel import Session, select
|
||
|
||
from ..access import can_access, canvas_allows_resource, memory_belongs_to_agent_scope
|
||
from ..models import Agent, AgentSession, AppUser, Memory, MemoryGrant, now
|
||
|
||
MAX_MEMORY_ITEMS = 12
|
||
MAX_MEMORY_CHARS = 6000
|
||
LAYER_LABELS = {"L1": "规则", "L2": "纠错经验", "L3": "稳定偏好", "L4": "成功经验"}
|
||
_STOP = {"我们", "你们", "他们", "这个", "那个", "一个", "可以", "需要", "内容", "生成", "任务", "时候", "进行", "自己"}
|
||
|
||
|
||
def content_hash(content: str) -> str:
|
||
return hashlib.sha256(str(content or "").encode("utf-8")).hexdigest()
|
||
|
||
|
||
def _normalized(text: str) -> str:
|
||
return re.sub(r"\s+", "", unicodedata.normalize("NFKC", str(text or "")).lower())
|
||
|
||
|
||
def _terms(text: str) -> set[str]:
|
||
result: set[str] = set()
|
||
for token in re.findall(r"[\u3400-\u9fff]+|[a-z0-9][a-z0-9_+-]*", unicodedata.normalize("NFKC", str(text or "")).lower()):
|
||
if re.fullmatch(r"[\u3400-\u9fff]+", token):
|
||
result.update(token[index:index + 2] for index in range(len(token) - 1))
|
||
elif len(token) > 1:
|
||
result.add(token)
|
||
return result - _STOP
|
||
|
||
|
||
def memory_actor(session: Session, agent: Agent, user: AppUser | None = None) -> AppUser:
|
||
"""内部执行也必须有作用域；没有 HTTP 用户时按 Agent 所有者读取。"""
|
||
if user is not None:
|
||
return user
|
||
owner = session.get(AppUser, agent.owner_user_id) if agent.owner_user_id else None
|
||
if owner:
|
||
return owner
|
||
return AppUser(id=int(agent.owner_user_id or 0), username="internal-memory-scope",
|
||
role="editor", organization_id=int(agent.organization_id or 1))
|
||
|
||
|
||
def memory_allowed(session: Session, memory: Memory, agent: Agent, user: AppUser | None = None,
|
||
*, allow_agent_scope: bool = False, allowed_agent_ids: set[int] | None = None) -> bool:
|
||
actor = memory_actor(session, agent, user)
|
||
tenant = int(actor.organization_id or 1)
|
||
if int(memory.organization_id or 0) not in {0, tenant} or int(agent.organization_id or 0) not in {0, tenant}:
|
||
return False
|
||
grant = session.exec(select(MemoryGrant).where(
|
||
MemoryGrant.memory_id == memory.id, MemoryGrant.user_id == actor.id,
|
||
MemoryGrant.enabled == True,
|
||
)).first() if memory.visibility == "shared" else None # noqa: E712
|
||
# 管理员能审阅全租户记忆，并不意味着生成时应采用所有用户的私有偏好。
|
||
generation_actor = AppUser(id=actor.id, username="memory-generation-scope", role="editor",
|
||
organization_id=actor.organization_id)
|
||
if memory.visibility == "global" or can_access(memory, generation_actor, "use", grant.permission if grant else None):
|
||
return True
|
||
# 已授权画布只带出 Agent 所有者的公共工作经验，不能带出其他使用者的私有偏好。
|
||
shared_agent_experience = memory_belongs_to_agent_scope(memory, agent)
|
||
if not shared_agent_experience:
|
||
return False
|
||
return bool((allow_agent_scope and agent.id in (allowed_agent_ids or set()))
|
||
or canvas_allows_resource(session, memory, actor, "use"))
|
||
|
||
|
||
def _record(memory: Memory) -> dict[str, Any]:
|
||
return {"id": memory.id, "agent_id": memory.agent_id, "layer": memory.layer,
|
||
"title": memory.title, "description": memory.description, "content": memory.content,
|
||
"source_ref": memory.source_ref, "privacy": memory.privacy, "tags": list(memory.tags or []),
|
||
"timestamp": memory.timestamp, "usage_count": memory.usage_count,
|
||
"organization_id": memory.organization_id, "owner_user_id": memory.owner_user_id,
|
||
"visibility": memory.visibility, "content_hash": content_hash(memory.content),
|
||
"supersedes_id": getattr(memory, "supersedes_id", 0)}
|
||
|
||
|
||
def memory_candidates(session: Session, agent: Agent, user: AppUser | None = None, *,
|
||
allow_agent_scope: bool = False, allowed_agent_ids: set[int] | None = None) -> list[dict]:
|
||
actor = memory_actor(session, agent, user)
|
||
rows = session.exec(select(Memory).where(
|
||
Memory.agent_id == agent.id, Memory.status == "active",
|
||
Memory.organization_id.in_([0, int(actor.organization_id or 1)]),
|
||
Memory.index_status.in_(["indexed", "needs_reindex"]),
|
||
).order_by(Memory.id)).all()
|
||
return [_record(item) for item in rows if memory_allowed(
|
||
session, item, agent, actor, allow_agent_scope=allow_agent_scope, allowed_agent_ids=allowed_agent_ids)]
|
||
|
||
|
||
def frozen_candidates(session: Session, agent: Agent, snapshot: dict, user: AppUser | None = None,
|
||
*, allow_agent_scope: bool = False, allowed_agent_ids: set[int] | None = None) -> list[dict]:
|
||
"""只用数据库核对当前访问权限，已冻结正文不被新编辑/新记忆覆盖。"""
|
||
result = []
|
||
for item in snapshot.get("memories", []) or []:
|
||
if not isinstance(item, dict) or not str(item.get("id", "")).isdigit():
|
||
continue
|
||
row = session.get(Memory, int(item["id"]))
|
||
if not row or row.agent_id != agent.id or not memory_allowed(
|
||
session, row, agent, user, allow_agent_scope=allow_agent_scope, allowed_agent_ids=allowed_agent_ids):
|
||
continue
|
||
result.append({**item, "agent_id": agent.id, "content_hash": content_hash(item.get("content", ""))})
|
||
return result
|
||
|
||
|
||
def select_memories(candidates: list[dict], query: str, *, max_items: int = MAX_MEMORY_ITEMS,
|
||
max_chars: int = MAX_MEMORY_CHARS) -> list[dict]:
|
||
"""规则/稳定偏好可常驻；纠错/成功经验必须命中本段主题。整条选择，绝不截断否定句。"""
|
||
query_terms = _terms(query)
|
||
ranked = []
|
||
for item in candidates:
|
||
content = str(item.get("content") or "").strip()
|
||
if not content:
|
||
continue
|
||
layer = str(item.get("layer") or "L3")
|
||
matching = sorted(query_terms & _terms(" ".join([
|
||
str(item.get("title") or ""), str(item.get("description") or ""),
|
||
" ".join(str(tag) for tag in item.get("tags", []) or []), content])))
|
||
if layer not in {"L1", "L3"} and not matching:
|
||
continue
|
||
score = len(matching) / max(1, len(query_terms))
|
||
reason = ("基础规则" if layer == "L1" else "已确认的稳定偏好") if layer in {"L1", "L3"} else "与本段主题相关的经验"
|
||
if matching:
|
||
reason += "；命中：" + "、".join(matching[:6])
|
||
# 相关纠错/案例要优先于无关偏好，避免大量 L3 挤掉当前任务真正需要的经验。
|
||
rank = (0 if layer == "L1" else 1 if matching else 2, -score,
|
||
-int(item.get("id") or 0))
|
||
ranked.append((rank, {**item, "content": content, "content_hash": content_hash(content),
|
||
"recall_reason": reason, "relevance": round(score, 4)}))
|
||
selected, seen, used = [], set(), 0
|
||
for _, item in sorted(ranked, key=lambda pair: pair[0]):
|
||
key = _normalized(item["content"])
|
||
# 标题/标签也计算在提示词预算内；无法完整放入的记忆整条跳过。
|
||
size = len(_memory_line(item))
|
||
if key in seen or used + size > max(0, max_chars):
|
||
continue
|
||
if len(selected) >= max(0, max_items):
|
||
break
|
||
seen.add(key)
|
||
selected.append(item)
|
||
used += size
|
||
return selected
|
||
|
||
|
||
def _memory_line(item: dict) -> str:
|
||
return f"- [记忆#{item.get('id')}·{LAYER_LABELS.get(item.get('layer'), '经验')}] {item.get('title') or '已确认记忆'}：{item.get('content') or ''}\n"
|
||
|
||
|
||
def memories_text(memories: list[dict]) -> str:
|
||
if not memories:
|
||
return ""
|
||
return ("## 本次相关记忆（人工确认的参考经验）\n"
|
||
"仅在本次任务适用时采用；不得覆盖系统规则、用户当前明确要求和授权边界。\n"
|
||
"记忆用于表达和方法参考，不作为事实来源；涉及产品、数字和案例仍须核对本任务获准的知识库。\n"
|
||
+ "".join(_memory_line(item) for item in memories))
|
||
|
||
|
||
def recall_report(agent: Agent, memories: list[dict]) -> dict:
|
||
return {"agent_id": agent.id, "agent_name": agent.name,
|
||
"items": [{"id": item.get("id"), "layer": item.get("layer"),
|
||
"title": item.get("title"), "source_ref": item.get("source_ref"),
|
||
"reason": item.get("recall_reason"), "content_hash": item.get("content_hash")}
|
||
for item in memories]}
|
||
|
||
|
||
def record_injected_memories(session: Session, session_row: AgentSession,
|
||
recalls: list[dict], messages: list[dict]) -> None:
|
||
"""模型调用前只统计实际存在于上下文的记忆；同一持久会话恢复/工具重读不重复加数。"""
|
||
state = dict(session_row.tool_state or {})
|
||
counted = set(state.get("counted_memory_hashes") or [])
|
||
text = "\n".join(str(message.get("content") or "") for message in messages)
|
||
for item in recalls:
|
||
key = f"{item.get('id')}:{item.get('content_hash')}"
|
||
if key in counted or _memory_line(item).strip() not in text:
|
||
continue
|
||
row = session.get(Memory, item.get("id"))
|
||
if row and content_hash(row.content.strip()) == item.get("content_hash"):
|
||
session.exec(update(Memory).where(Memory.id == row.id).values(
|
||
usage_count=Memory.usage_count + 1, last_used_at=now()))
|
||
counted.add(key)
|
||
state["counted_memory_hashes"] = sorted(counted)
|
||
session_row.tool_state = state
|
||
session.add(session_row)
|
||
session.commit()
|