wangjianlong_0626
57f56bbcbe
## 1. 独立缺陷：`profile_snapshots` 被两个 ORM 类重复映射
排查 `memory_sync_outbox` 中 `target_store='neo4j'` 那行 `last_error='InvalidRequestError'`
时发现，根因不在图库，而在模型层：
- `app/model/profile.py` → `ProfileSnapshot` 映射 `profile_snapshots`
- `app/model/risk_questionnaire.py` → **另一个** `ProfileSnapshot` 也映射 `profile_snapshots`
SQLAlchemy 不允许两个类映射同一张表。实测：
| 场景 | 结果 |
|---|---|
| 单独导入 `app.main` / `app.worker.runtime` | 正常 |
| 单独导入 `profile_assembly_service` / `risk_questionnaire_service` | 正常 |
| **两者同时导入** | `InvalidRequestError: Table 'profile_snapshots' is already defined` |
Worker 在同一进程里既要处理 `profile.rebuild_requested`（走 `app.model.profile`），
又要处理投顾风险问卷（走 `risk_questionnaire.py`）——所以这是**会打挂 Worker 的缺陷**，
不是理论风险。
修法：`app/model/risk_questionnaire.py` 不再重复定义，改为从 `app.model.profile`
转出（re-export），既有 4 处 `from app.model.risk_questionnaire import ProfileSnapshot`
无需改动。原定义多映射的 `current_customer_id` 经全仓核查无人使用，故不保留
（`app.model.profile` 明确注明该列由数据库维护、故意不映射）。
## 2. 消费端兜底 `memory_sources`
`memory_sources` 是本线新增的投影入参，而投顾线两处生产者的 payload
（`{customer_id, profile_uuid, version, profile}`）没有这个键，原样会导致它们每次画像
变更都 `memory_sources is invalid` → 重试至死信。
新增 `WorkerRuntime._with_memory_sources()`：**键缺失或为 None** 时回退查询该客户
`memory_unit` 中 `status='active'` 的记忆，并记 warning（使"谁没提供"保持可见）。
语义成立：长期记忆是**客户级**而非画像版本级的，每条记忆自带 `version`，
适配器按 `memory_uuid + version` 幂等，故"用的是哪一版"仍确定。
**兜底不掩盖真错误**：键**存在但格式不对**时**不兜底**，原样交给适配器失败关闭。
实现上用键存在性判断而非 `isinstance`——后者会把"缺失"与"格式错"混为一谈，
那是初版实现里的一个真 bug，被新测试抓出后修正。
## 3. 补上此前欠缺的消费端全路径验证
此前"整合验证"是直接调适配器，跳过了 outbox 的领取→分派→状态更新。
- **失败分支**（Milvus 断开时实测）：行被领取、按 target_store 分派、异常被捕获、
`status`/`retry_count`/`last_error`/`next_retry_at` 正确落库。
- **成功分支**（注入替身向量客户端）：outbox 行 → `processed`、`processed_at` 已写；
不可投影的 `constraint:` 被跳过（只写 1 行）；维度 1024；字符串客户号转 int；
**手机号脱敏生效**（`稳健型投资者，手机号 [手机号已隐藏] 请勿外泄`）。
- **兜底实证**：历史行 `id=5`（payload 无 `memory_sources`）经兜底后成功投递为 `processed`。
- **修复实证**：`id=6` 的 `last_error` 从 `InvalidRequestError` 变为
`RecoverableAgentError`（图库不可用）——证明重复定义缺陷确已消除，剩下的是环境问题。
## 4. 测试与验证
- 新增 `tests/unit/worker/test_runtime_profile_projection.py`（5 用例：已提供原样透传、
缺失兜底、空记忆给空列表而非删键、无客户号不兜底、格式错不兜底）
- 全量：`2 failed, 1312 passed, 2 skipped`（2 个失败为既有环境项，非本次引入）
- mypy：`Success: no issues found in 227 source files`
- 表结构审计：89 张业务表无缺失/意外（未改动任何表结构）
- 文档守卫：41 份文档无编号冲突
## 5. 文档
`docs/32-记忆投影链路实现说明.md` 增补 §6.1（兜底）、§6.2（重复定义缺陷）、
§7.1（消费端全路径验证）并更新验证表与文件清单；
`AGENTS.md` 新增"一张表只能有一个 ORM 类"易错点、校正测试基线数字。
## 未做
未改 `docs/00` 基线、未动数据库迁移、未改投顾线生产者代码、未启动常驻 Worker。
遗留：投顾线两处生产者的 payload 仍缺 `memory_sources`（已有兜底，不再死信，
但根治应由投顾线确认）；Milvus/Neo4j 容器本轮不可用（Docker Desktop 崩溃），
`id=6` 停在 failed 属环境不可用、非代码缺陷。
803 lines
43 KiB
Python
803 lines
43 KiB
Python
import asyncio
|
||
import contextlib
|
||
import logging
|
||
from collections.abc import Awaitable, Callable
|
||
from dataclasses import dataclass
|
||
from datetime import UTC, datetime, timedelta
|
||
from typing import Any, Protocol, cast
|
||
from uuid import uuid4
|
||
|
||
from sqlalchemy import select, update
|
||
|
||
from app.core.config import Settings, get_settings
|
||
from app.core.contracts import AgentRequest, AgentResult, RequestContext
|
||
from app.core.errors import AgentError, RecoverableAgentError, RunLeaseLostError
|
||
from app.infrastructure.db import SessionFactory
|
||
from app.model.audit import InteractionAudit
|
||
from app.model.conversation import ConversationMessage
|
||
from app.model.platform import AgentRun, DomainEventOutbox, RequestIdempotency
|
||
from app.repository.agent_run_repository import AgentRunRepository
|
||
from app.service.agent.bootstrap import (
|
||
get_agent_factory,
|
||
get_memory_cache_adapter,
|
||
get_memory_embedding_service,
|
||
get_milvus_knowledge_writer,
|
||
get_milvus_profile_vector_client,
|
||
get_model_service,
|
||
)
|
||
from app.service.agent.executor import AgentExecutor
|
||
from app.service.agent.factory import AgentFactory
|
||
from app.service.agent_persistence_service import AgentPersistenceService
|
||
from app.service.identity_service import IdentityService
|
||
from app.service.memory_extraction_service import (
|
||
ExtractionEndpointResolver,
|
||
MemoryExtractionService,
|
||
)
|
||
from app.service.memory_lifecycle_service import MemoryLifecycleService, Mode
|
||
from app.service.memory_recall_service import MemoryRecallService
|
||
from app.service.memory_service import CacheDeleteAdapter, MemoryService
|
||
from app.service.memory_taxonomy import BUSINESS_EVENT_TYPES
|
||
from app.service.model_gateway import DatabaseModelEndpointResolver, ModelGenerationService
|
||
from app.worker.episode_worker import (
|
||
EpisodeConsumptionResult,
|
||
EpisodeExtractionConsumer,
|
||
EpisodeWorker,
|
||
)
|
||
from app.worker.knowledge_vector_worker import build_knowledge_handlers
|
||
from app.worker.memory_extraction_worker import MemoryExtractionWorker
|
||
from app.worker.outbox_worker import OutboxHandlerError, OutboxWorker
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# episode 聚合按轮次节流：每 N 轮顺带处理一次已静默的会话片段，
|
||
# 避免每轮轮询都做一次客户维度的 distinct 查询。
|
||
EPISODE_INTERVAL_ROUNDS = 30
|
||
|
||
# 一轮内最多消费的事件条数。`OutboxWorker.publish_one` 每次只领一条（单条事务让幂等与
|
||
# skip_locked 锁语义保持简单），所以清空速度原先等于轮询速度：262 条积压要 262 轮。
|
||
OUTBOX_DISPATCH_LIMIT = 10
|
||
# 单轮消费的片段条数上限：批处理必须可中断，不能一次吃完整库。
|
||
EPISODE_CONSUME_LIMIT = 20
|
||
PROJECTION_AUDIT_ACTION = "memory.projection_cleanup"
|
||
|
||
#: 组装层默认值的哨兵。知识写路径的三个依赖都要区分"没传（取生产装配）"与"显式
|
||
#: `None`（显式降级）"：`None` 若同时表示两者，测试里就无法在不改环境变量的前提下
|
||
#: 构造"Milvus 未配置"的场景，降级行为也就无法被固定。
|
||
_UNSET: Any = object()
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class ProjectionCleanupOutcome:
|
||
"""投影清理结果；`cleaned=False` 时 `detail` 必须说明真实缺口。"""
|
||
|
||
cleaned: bool
|
||
detail: str = ""
|
||
|
||
|
||
class ProjectionCleaner(Protocol):
|
||
"""Milvus/Neo4j 投影清理边界：由组装层注入，未注入即显式降级（绝不伪造成功）。"""
|
||
|
||
async def cleanup(self, *, memory_uuid: str, operation: str) -> ProjectionCleanupOutcome: ...
|
||
|
||
|
||
class WorkerRuntime:
|
||
def __init__(
|
||
self, factory: AgentFactory | None = None, settings: Settings | None = None,
|
||
resolve_identity: Callable[[RequestContext], Awaitable[RequestContext]] | None = None,
|
||
model_service: ModelGenerationService | None = None,
|
||
endpoint_resolver: ExtractionEndpointResolver | None = None,
|
||
memory_cache: CacheDeleteAdapter | None = None,
|
||
projection_cleaner: ProjectionCleaner | None = None,
|
||
relationships: Any = None,
|
||
knowledge_writer: Any = _UNSET,
|
||
knowledge_embedder: Any = _UNSET,
|
||
knowledge_endpoint_resolver: Any = _UNSET,
|
||
profile_vector_client: Any = _UNSET,
|
||
) -> None:
|
||
self.factory = factory if factory is not None else get_agent_factory()
|
||
self.settings = settings or get_settings()
|
||
self.resolve_identity = resolve_identity or IdentityService().resolve
|
||
# 记忆抽取必须走与业务 Agent 相同的模型路由入口。默认取生产装配
|
||
# （app/service/agent/bootstrap.py），允许构造参数注入替身模型与端点解析器，
|
||
# 使验收探针不依赖库中真实模型端点配置。
|
||
self.model_service = model_service if model_service is not None else get_model_service()
|
||
self.memory_extraction = MemoryExtractionService(
|
||
self.model_service,
|
||
endpoint_resolver if endpoint_resolver is not None else DatabaseModelEndpointResolver(),
|
||
)
|
||
# 召回热缓存适配器：任何记忆写入（事件消费、片段消费）都必须让该客户的
|
||
# 召回热缓存失效，否则新记忆在 TTL 内召回不到。默认取生产装配，可注入替身。
|
||
self.memory_cache: CacheDeleteAdapter | None = (
|
||
memory_cache if memory_cache is not None else get_memory_cache_adapter()
|
||
)
|
||
# Milvus/Neo4j 删除客户端：由**组装层**（app/worker/__main__.py）注入生产实现，
|
||
# 这里不兜底。组件内部给默认实现会把"尚未装配"这一事实悄悄盖住——而"未注入即显式
|
||
# 降级并留痕"是本模块刻意保留的语义（有单测守着），因此默认值保持 None。
|
||
self.projection_cleaner = projection_cleaner
|
||
# 图关系服务：画像投影用它写入节点与关系（投顾的多跳推荐、风控的关系网络都读它）。
|
||
# 默认取生产装配；图库不可用时该值为 None，投影如实降级而不是失败。
|
||
if relationships is not None:
|
||
self.relationships = relationships
|
||
else:
|
||
# 延迟导入：bootstrap 会间接导入本模块，模块级导入会形成循环依赖
|
||
from app.service.agent.bootstrap import get_relationship_service
|
||
|
||
self.relationships = get_relationship_service()
|
||
# 知识向量同步：Milvus 写适配器 + 嵌入服务 + 端点解析器全部由组装层注入
|
||
# （`app/service/agent/bootstrap.py`）。写适配器构造是**惰性**的（不连 Milvus），
|
||
# 所以这里取默认值不会让 worker 起不来；真连不上时在写入时抛
|
||
# `RecoverableAgentError`，交给 OutboxWorker 的退避重试与死信机制。
|
||
self.knowledge_writer = (
|
||
get_milvus_knowledge_writer() if knowledge_writer is _UNSET else knowledge_writer
|
||
)
|
||
self.knowledge_embedder = (
|
||
get_memory_embedding_service() if knowledge_embedder is _UNSET
|
||
else knowledge_embedder
|
||
)
|
||
self.knowledge_endpoint_resolver = (
|
||
DatabaseModelEndpointResolver() if knowledge_endpoint_resolver is _UNSET
|
||
else knowledge_endpoint_resolver
|
||
)
|
||
# 降级告警只打一次：dispatch 是轮询热路径，每轮一条 warning 会把日志淹掉。
|
||
self._knowledge_degraded_logged = False
|
||
# 画像投影（`memory_sync_outbox`）消费装配。
|
||
#
|
||
# 与 `knowledge_writer` 同一取向：客户端构造**惰性**（不连 Milvus），
|
||
# `milvus_uri` 未配置时显式降级为不注册 handler（事件留 pending、可观测、可重放），
|
||
# 绝不伪造同步成功。
|
||
self.profile_vector_client = (
|
||
get_milvus_profile_vector_client()
|
||
if profile_vector_client is _UNSET
|
||
else profile_vector_client
|
||
)
|
||
self.profile_endpoint_resolver = self.knowledge_endpoint_resolver
|
||
self._profile_degraded_logged = False
|
||
# episode 聚合是低频批处理，按轮次节流而不是每轮都查。
|
||
self._episode_rounds = 0
|
||
|
||
async def dispatch_one(self, *, run_id: str | None = None) -> bool:
|
||
# Outbox acknowledges a durable SQL queue entry, not an in-memory task.
|
||
async with SessionFactory() as session:
|
||
async def dispatch(payload: dict[str, Any]) -> None:
|
||
run = await AgentRunRepository(session).get(str(payload["run_id"]))
|
||
if run is None:
|
||
raise OutboxHandlerError("run not found")
|
||
|
||
async def dispatch_memory_extraction(payload: dict[str, Any]) -> None:
|
||
if "message_id" not in payload or "customer_id" not in payload:
|
||
raise OutboxHandlerError("memory extraction payload is incomplete")
|
||
# 幂等键只认事件 id，由 worker 自己按 payload 回查，避免调用方漏传。
|
||
# 注入召回缓存适配器：写入生效后立即失效该客户的热缓存。
|
||
await MemoryExtractionWorker(
|
||
session, extractor=self.memory_extraction, cache=self.memory_cache
|
||
).handle(payload)
|
||
|
||
async def dispatch_run_completed(payload: dict[str, Any]) -> None:
|
||
# 结果消息与审计已由 complete_run 同事务落库，此事件只承担
|
||
# "运行已完成"的对外通知职责。当前没有独立外部消费者，
|
||
# 这里显式消费以免事件永久滞留；接入推送链路时在此处扩展。
|
||
if not str(payload.get("run_id", "")):
|
||
raise OutboxHandlerError("agent.run_completed payload is incomplete")
|
||
|
||
async def dispatch_cache_invalidate(payload: dict[str, Any]) -> None:
|
||
await self._invalidate_config_cache(payload)
|
||
|
||
async def dispatch_profile_rebuild(payload: dict[str, Any]) -> None:
|
||
"""画像重建 + 图投影，由记忆写入后发出的事件驱动。
|
||
|
||
为什么绕一层事件而不在记忆抽取处直接调用：抽取时那条记忆还在**未提交**的
|
||
事务里，另开 session 去重建画像看不到它（实测：快照加了、事实没进、图里
|
||
也没多出关系）。事件只可能在本事务提交之后被消费，届时数据一定可见。
|
||
"""
|
||
customer_id = payload.get("customer_id")
|
||
if not customer_id:
|
||
raise OutboxHandlerError("profile.rebuild_requested payload is incomplete")
|
||
# 延迟导入：bootstrap 会间接导入本模块，模块级导入会形成循环依赖
|
||
from app.service.profile_assembly_service import ProfileAssemblyService
|
||
from app.service.profile_graph_projection_service import (
|
||
ProfileGraphProjectionService,
|
||
)
|
||
|
||
outcome = await ProfileAssemblyService(session).rebuild(int(customer_id))
|
||
if outcome.get("profile") is None:
|
||
# 客户尚未开户：画像行由开户流程创建（trade_account 等字段属注册侧所有），
|
||
# 这里不代建。事实已提升进 user_facts，开户后再重建即可。
|
||
logger.info("profile rebuild skipped (not opened) customer_id=%s", customer_id)
|
||
return
|
||
projection = ProfileGraphProjectionService(session, self.relationships)
|
||
result = await projection.project_customer(int(customer_id))
|
||
if result.degraded:
|
||
logger.warning("graph projection degraded customer_id=%s reason=%s",
|
||
customer_id, result.reason)
|
||
|
||
async def dispatch_projection_cleanup(payload: dict[str, Any]) -> None:
|
||
# memory.invalidated / memory.deleted 由 MemoryLifecycleService 按
|
||
# memory_uuid 写入，这里做幂等的投影清理（Milvus 向量、Neo4j 关系）。
|
||
await self._cleanup_projection(payload, session=session)
|
||
|
||
async def dispatch_memory_deletion(payload: dict[str, Any]) -> None:
|
||
# 客户级联失效/删除的公共入口：业务侧（用户注销、合规删除令）
|
||
# 只写 memory.deletion_requested 事件，不直接操作记忆表。
|
||
customer_id = payload.get("customer_id")
|
||
if not customer_id:
|
||
raise ValueError("memory.deletion_requested payload is incomplete")
|
||
mode = str(payload.get("mode", "invalidate"))
|
||
if mode not in {"invalidate", "delete"}:
|
||
raise OutboxHandlerError("memory.deletion_requested mode is invalid")
|
||
await MemoryLifecycleService(session).run(
|
||
int(customer_id),
|
||
mode=cast("Mode", mode),
|
||
reason=str(payload.get("reason", "customer_lifecycle")),
|
||
trace_id=str(payload.get("trace_id", "")),
|
||
)
|
||
|
||
handlers: dict[str, Callable[[dict[str, Any]], Awaitable[None]]] = {
|
||
"agent.run_requested": dispatch,
|
||
"memory.extraction_requested": dispatch_memory_extraction,
|
||
"agent.run_completed": dispatch_run_completed,
|
||
"config.cache_invalidate_requested": dispatch_cache_invalidate,
|
||
"memory.deletion_requested": dispatch_memory_deletion,
|
||
# 投影清理：这两类事件此前没有消费者，永久 pending。
|
||
"memory.invalidated": dispatch_projection_cleanup,
|
||
"memory.deleted": dispatch_projection_cleanup,
|
||
# 画像重建：记忆写入后自动触发，使「记忆 → 画像 → 图」全链路无需手工介入
|
||
"profile.rebuild_requested": dispatch_profile_rebuild,
|
||
}
|
||
# 知识向量同步/删除：Task 5 交付了 handler 与写适配器，但先前没有任何生产装配
|
||
# 调用它们 —— 事件类型不在上面的白名单里，`OutboxWorker.publish_one` 的
|
||
# `event_type.in_(tuple(self.handlers))` 就永远领不到这些行，现库 408 条
|
||
# `knowledge.vector_sync_requested` 因此永久 pending、Milvus 零向量、
|
||
# 检索永远返回空。这里把两个 handler 合并进同一个字典（**同一个 `session`**，
|
||
# handler 不得 commit，事务仍归 OutboxWorker）。
|
||
if self.knowledge_writer is not None:
|
||
handlers.update(build_knowledge_handlers(
|
||
session,
|
||
writer=self.knowledge_writer,
|
||
embedder=self.knowledge_embedder,
|
||
endpoint_resolver=self.knowledge_endpoint_resolver,
|
||
))
|
||
elif not self._knowledge_degraded_logged:
|
||
# 显式降级 + 留痕：不注册 handler，知识事件保持 pending（库里可查、可重放），
|
||
# 绝不伪造"已同步"。与 `projection_cleaner` 的降级口径一致。
|
||
self._knowledge_degraded_logged = True
|
||
logger.warning(
|
||
"knowledge vector handlers not registered: milvus writer unavailable "
|
||
"(check settings.milvus_uri); knowledge.vector_sync_requested / "
|
||
"knowledge.vector_delete_requested stay pending"
|
||
)
|
||
return await OutboxWorker(session, handlers).publish_one(aggregate_id=run_id)
|
||
|
||
async def dispatch_batch(self, *, limit: int = OUTBOX_DISPATCH_LIMIT) -> int:
|
||
"""一轮内尽量多消费事件，返回实际消费条数。
|
||
|
||
单条领取的语义与幂等边界都不变，只是不再让"每轮一条"限制清空速度：
|
||
队列空时提前退出，因此稳态下与原来的一次调用开销相同，积压时才提速。
|
||
"""
|
||
consumed = 0
|
||
for _ in range(max(1, limit)):
|
||
if not await self.dispatch_one():
|
||
break
|
||
consumed += 1
|
||
return consumed
|
||
|
||
async def _invalidate_config_cache(self, payload: dict[str, Any]) -> None:
|
||
"""删除发布配置与记忆召回热缓存键；Redis 不可用时只记录告警，不阻塞事件消费。
|
||
|
||
记忆召回热缓存的键必须由 `MemoryRecallService.cache_keys` 枚举：此前这里手写的
|
||
`mem:mid:hot:{customer_id}` 并不是召回缓存的真实前缀（真实前缀是 `mem:recall`），
|
||
失效动作一直打在并不存在的键上。事件 payload 里没有查询摘要与 limit，客户级
|
||
枚举是唯一可靠的失效方式（召回热缓存本身也只服务常见参数组合）。
|
||
"""
|
||
release_id = payload.get("release_id")
|
||
raw_customer_id = payload.get("customer_id")
|
||
keys: list[str] = []
|
||
if raw_customer_id:
|
||
keys.extend(MemoryRecallService.cache_keys(int(raw_customer_id)))
|
||
if release_id is not None:
|
||
keys.append(f"config:release:{release_id}")
|
||
if not keys:
|
||
return
|
||
client = await self._redis_client()
|
||
if client is None:
|
||
return
|
||
try:
|
||
await client.delete(*keys)
|
||
except Exception:
|
||
logger.warning("config cache invalidation degraded keys=%s", ",".join(keys))
|
||
finally:
|
||
await client.aclose()
|
||
|
||
@staticmethod
|
||
async def _conversation_history(
|
||
session: Any, *, session_id: str, user_id: int, before_message_id: int | None,
|
||
limit: int = 10,
|
||
) -> tuple[Any, ...]:
|
||
"""取该会话最近若干轮对话，按时间正序（旧 → 新）返回。
|
||
|
||
以 MySQL 的会话消息为**唯一来源**，不引入 Redis 双写：消息在受理时已经落库，
|
||
再同步一份到 Redis 只会带来不一致与 TTL 管理成本，换来的仅是一次索引查询的节省。
|
||
方案 §2.2 设想的是 Redis 列表，这里取等价语义（同样"最近若干轮、超出即截断"）
|
||
而不复制存储。
|
||
|
||
`before_message_id` 排除本轮请求消息本身：它刚写入库，若也算进历史，
|
||
模型会在上下文里看到自己的问题被重复一遍。
|
||
|
||
截断按**条数**而非 token：这里没有与模型一致的分词器，按 token 截断只能靠估算、
|
||
边界会随实现漂移；按条数是确定性的，宁可少给几轮，也不给一个不稳定的边界。
|
||
"""
|
||
from app.core.contracts import ConversationTurn
|
||
from app.repository.conversation_repository import ConversationRepository
|
||
|
||
rows = await ConversationRepository(session).messages(
|
||
session_id, user_id, limit + 1, before=before_message_id
|
||
)
|
||
# repository 按 id DESC 返回（最新在前），这里翻正为旧 → 新
|
||
ordered = list(reversed(rows))[-limit:]
|
||
turns: list[Any] = []
|
||
for row in ordered:
|
||
content = str(row.content or "").strip()
|
||
if not content:
|
||
continue
|
||
turns.append(ConversationTurn(
|
||
role="assistant" if str(row.role) == "assistant" else "user",
|
||
content=content,
|
||
))
|
||
return tuple(turns)
|
||
|
||
async def _cleanup_projection(self, payload: dict[str, Any], *, session: Any) -> None:
|
||
"""幂等清理一条记忆的派生投影（Milvus 向量、Neo4j 关系）。
|
||
|
||
为什么选择"显式降级 + 留痕"而不是假装成功：当前组装层没有提供任何删除客户端
|
||
（`bootstrap` 只装配了召回用的 Milvus 读适配器，Neo4j 连读适配器都没有），
|
||
伪造"已删除"会让合规删除令在投影侧静默失效。因此无客户端时只做两件事：
|
||
记录告警，并写一条 `interaction_audit`（`status=skipped_no_client`），事件本身
|
||
照常标记为已消费——权威库（MySQL）状态已经正确，投影是可重建的派生数据，
|
||
让事件永久 pending 只会阻塞队列里其它事件。
|
||
注入 `projection_cleaner` 后同一入口执行真实删除，审计只记录适配器返回的真实
|
||
结论（`cleaned` 为假即写 `skipped`），不做任何"假定成功"的兜底。
|
||
"""
|
||
memory_uuid = str(payload.get("memory_uuid", "")).strip()
|
||
if not memory_uuid:
|
||
raise ValueError("memory projection cleanup payload is incomplete")
|
||
operation = str(payload.get("operation", "invalidate"))
|
||
raw_customer_id = payload.get("customer_id")
|
||
customer_id = int(raw_customer_id) if raw_customer_id else None
|
||
cleaner = self.projection_cleaner
|
||
if cleaner is None:
|
||
logger.warning(
|
||
"projection cleanup degraded: milvus/neo4j delete client not configured "
|
||
"memory_uuid=%s operation=%s", memory_uuid, operation,
|
||
)
|
||
await self._audit_projection(
|
||
session, memory_uuid, customer_id, operation,
|
||
status="skipped_no_client", reason="projection delete client not configured",
|
||
)
|
||
return
|
||
outcome = await cleaner.cleanup(memory_uuid=memory_uuid, operation=operation)
|
||
if outcome.cleaned:
|
||
logger.info("projection cleanup done memory_uuid=%s operation=%s",
|
||
memory_uuid, operation)
|
||
else:
|
||
logger.warning("projection cleanup degraded memory_uuid=%s reason=%s",
|
||
memory_uuid, outcome.detail)
|
||
await self._audit_projection(
|
||
session, memory_uuid, customer_id, operation,
|
||
status="cleaned" if outcome.cleaned else "skipped", reason=outcome.detail,
|
||
)
|
||
|
||
async def _audit_projection(
|
||
self, session: Any, memory_uuid: str, customer_id: int | None, operation: str,
|
||
*, status: str, reason: str,
|
||
) -> None:
|
||
"""投影清理留痕：审计行与事件同事务提交，降级状态在库里可查。"""
|
||
session.add(InteractionAudit(
|
||
actor_type="system",
|
||
actor_id=None,
|
||
target_customer_id=customer_id,
|
||
session_id=None,
|
||
portal=None,
|
||
action_type=PROJECTION_AUDIT_ACTION,
|
||
detail={
|
||
"memory_uuid": memory_uuid,
|
||
"operation": operation,
|
||
"status": status,
|
||
"reason": reason,
|
||
},
|
||
created_at=datetime.now(UTC).replace(tzinfo=None),
|
||
))
|
||
await session.flush()
|
||
|
||
async def _redis_client(self) -> Any:
|
||
try:
|
||
from redis.asyncio import Redis
|
||
except ImportError:
|
||
logger.warning("redis client unavailable; cache invalidation skipped")
|
||
return None
|
||
client: Any = Redis.from_url(
|
||
self.settings.redis_url,
|
||
socket_connect_timeout=self.settings.redis_connect_timeout_seconds,
|
||
socket_timeout=self.settings.redis_connect_timeout_seconds,
|
||
)
|
||
return client
|
||
|
||
async def run_once(self) -> bool:
|
||
dispatched = await self.dispatch_batch() > 0
|
||
# 画像投影消费：与领域事件同一轮次内处理。失败只告警，不影响 run 的处理与
|
||
# 轮询节奏——事件仍在库里，下一轮照常重试（退避由 worker 自己记在 next_retry_at）。
|
||
try:
|
||
projected = await self.consume_profile_projections() > 0
|
||
except Exception:
|
||
logger.warning("profile projection consumption failed", exc_info=True)
|
||
projected = False
|
||
dispatched = dispatched or projected
|
||
self._episode_rounds += 1
|
||
if self._episode_rounds % EPISODE_INTERVAL_ROUNDS == 0:
|
||
# 会话片段聚合：内部幂等（content_hash 唯一键），失败只告警，
|
||
# 不得影响 run 的处理与轮询节奏。
|
||
try:
|
||
await self.aggregate_episodes()
|
||
except Exception:
|
||
logger.warning("episode aggregation failed", exc_info=True)
|
||
# 聚合之后立刻消费：片段只有被消费才会变成记忆，否则"待提取"永久滞留。
|
||
try:
|
||
await self.consume_episodes()
|
||
except Exception:
|
||
logger.warning("episode extraction failed", exc_info=True)
|
||
now = datetime.now(UTC).replace(tzinfo=None)
|
||
async with SessionFactory() as session:
|
||
run_id = await session.scalar(select(AgentRun.run_id).where(
|
||
AgentRun.status.in_(("queued", "running", "cancel_requested")),
|
||
(AgentRun.locked_until.is_(None) | (AgentRun.locked_until < now)),
|
||
).order_by(AgentRun.created_at).limit(1))
|
||
if run_id is None:
|
||
return dispatched
|
||
return await self.execute(run_id) or dispatched
|
||
|
||
async def consume_profile_projections(self, *, limit: int = 20) -> int:
|
||
"""消费 `memory_sync_outbox` 的画像投影事件，返回本次处理条数。
|
||
|
||
两个目标存储的分工（**方案 A**：以架构师主干为主线，不引入第二套 Neo4j 投影）：
|
||
|
||
- `milvus` → 写入长期记忆向量集合 `user_long_term_memory_v1`
|
||
（此前**完全没有消费者**，事件永久滞留）；
|
||
- `neo4j` → 复用主干 `ProfileGraphProjectionService`。主干已由
|
||
`profile.rebuild_requested` 事件驱动同一条链，图投影是 `MERGE` 幂等的，
|
||
因此这里再投一次不产生重复节点/关系，只用于把 outbox 行的投递状态收敛掉。
|
||
|
||
为什么不让 handler 自己 commit：事务边界与 `dispatch_batch` 一致，
|
||
由本方法按条提交；单条失败由 `MemorySyncOutboxWorker` 内部转成
|
||
`failed`+退避或死信，不冒泡打断本轮其余事件。
|
||
"""
|
||
if self.profile_vector_client is None:
|
||
# 显式降级：不注册 handler 就交给 worker 判死信是**错的**（那是把配置缺失
|
||
# 伪装成投递失败）。这里直接不消费，事件保持 pending，由启动日志提示。
|
||
if not self._profile_degraded_logged:
|
||
self._profile_degraded_logged = True
|
||
logger.warning(
|
||
"profile projection disabled: milvus profile vector client unavailable; "
|
||
"memory_sync_outbox events stay pending"
|
||
)
|
||
return 0
|
||
# 收窄到局部变量：闭包内访问 self 属性时 mypy 无法保留上面的 None 判定。
|
||
vector_client = self.profile_vector_client
|
||
|
||
from app.infrastructure.milvus_profile_projection import MilvusProfileProjection
|
||
from app.worker.memory_sync_outbox_worker import MemorySyncOutboxWorker
|
||
|
||
async def project_milvus(payload: dict[str, Any]) -> None:
|
||
effective = await self._with_memory_sources(payload)
|
||
projection = MilvusProfileProjection(vector_client, self._profile_embed)
|
||
await projection.upsert(effective)
|
||
|
||
async def project_neo4j(payload: dict[str, Any]) -> None:
|
||
raw_customer_id = payload.get("customer_id")
|
||
# 显式 isinstance 而不是 `in (None, "")`：后者不做类型收窄，mypy 无法确认
|
||
# int() 的入参类型；同时也把"客户号必须是数字"这一契约写在类型检查里。
|
||
if not isinstance(raw_customer_id, (int, str)) or raw_customer_id == "":
|
||
raise RecoverableAgentError("profile projection payload has no customer_id")
|
||
customer_id = int(raw_customer_id)
|
||
# 延迟导入：与 dispatch_profile_rebuild 同一理由，避免模块级循环依赖。
|
||
from app.service.profile_graph_projection_service import (
|
||
ProfileGraphProjectionService,
|
||
)
|
||
|
||
async with SessionFactory() as session:
|
||
outcome = await ProfileGraphProjectionService(
|
||
session, self.relationships
|
||
).project_customer(customer_id)
|
||
if outcome.degraded:
|
||
# 图库不可用：如实抛出，让 worker 走失败/退避，而不是记成已投递。
|
||
raise RecoverableAgentError(f"graph projection degraded: {outcome.reason}")
|
||
|
||
worker = MemorySyncOutboxWorker(
|
||
{"milvus": project_milvus, "neo4j": project_neo4j}
|
||
)
|
||
handled = 0
|
||
for _ in range(max(1, limit)):
|
||
if not await worker.run_once():
|
||
break
|
||
handled += 1
|
||
return handled
|
||
|
||
async def _with_memory_sources(self, payload: dict[str, Any]) -> dict[str, Any]:
|
||
"""保证 payload 带 `memory_sources`；缺失时回退为查询当前有效记忆。
|
||
|
||
为什么需要这个兜底：`memory_sources` 是本仓新增的投影入参，而**投顾线两处
|
||
生产者**（`profile_governance_service` / `risk_questionnaire_service`）发的 payload
|
||
是 `{customer_id, profile_uuid, version, profile}`，**没有**这个键。若不兜底，
|
||
它们每次画像变更都会因 `memory_sources is invalid` 失败重试直至死信
|
||
（本仓 `memory_sync_outbox` 已有这种 `ValueError` 行留痕）。
|
||
|
||
为什么不是"缺失就报错"：缺失与"格式错"性质不同——缺失表示该生产者不知道要提供，
|
||
属契约演进期的正常情况；格式错（不是列表、字段不合法）仍由适配器**失败关闭**，
|
||
不会被这里掩盖。
|
||
|
||
回退查的是 `memory_unit` 中 `status='active'` 的行，即"该客户当前有效的长期记忆"。
|
||
这在语义上成立：长期记忆是**客户级**的，不是画像版本级的；且每条记忆自带
|
||
`version`，适配器按 `memory_uuid + version` 做幂等，所以"用的是哪一版"仍然确定。
|
||
|
||
每次兜底都记一条 warning，使"谁没提供 memory_sources"保持可见，而不是静默兼容。
|
||
"""
|
||
sources = payload.get("memory_sources")
|
||
# 只对"**键不存在或为 None**"兜底。若键存在但格式不对（例如字符串），
|
||
# 原样放行交给适配器报错——那是真错误，兜底会把它悄悄修好、线上永远看不见。
|
||
# （用 isinstance 判断会把这两种情况混为一谈，故用键存在性判断。）
|
||
if sources is not None:
|
||
return payload
|
||
|
||
raw_customer_id = payload.get("customer_id")
|
||
if not isinstance(raw_customer_id, (int, str)) or raw_customer_id == "":
|
||
# 没有客户号就无法兜底；交给适配器按原 payload 失败关闭。
|
||
return payload
|
||
customer_id = int(raw_customer_id)
|
||
|
||
from app.repository.profile_repository import ProfileRepository
|
||
|
||
async with SessionFactory() as session:
|
||
rows = await ProfileRepository(session).active_memories(customer_id)
|
||
logger.warning(
|
||
"profile projection payload has no memory_sources (customer_id=%s); "
|
||
"fell back to %s active memories from memory_unit",
|
||
customer_id,
|
||
len(rows),
|
||
)
|
||
return {
|
||
**payload,
|
||
"memory_sources": [
|
||
{
|
||
"memory_uuid": str(row["memory_uuid"]),
|
||
"memory_key": str(row["memory_key"]),
|
||
"content": str(row["content"]),
|
||
"memory_type": str(row["memory_type"]),
|
||
"confidence": float(row["confidence"]),
|
||
"version": int(row["version"]),
|
||
"valid_until": (
|
||
row["valid_until"].isoformat() if row["valid_until"] else None
|
||
),
|
||
}
|
||
for row in rows
|
||
],
|
||
}
|
||
|
||
async def _profile_embed(self, text: str) -> list[float]:
|
||
"""向量化一条记忆正文；端点走与知识向量化同一套已批准端点解析。
|
||
|
||
不复用 `bootstrap._embed_text`：那是模块私有函数，跨模块引用私有名会把
|
||
两处的耦合藏起来。这里用同一组公开装配（端点解析器 + embedding 服务）。
|
||
"""
|
||
endpoints = await self.profile_endpoint_resolver.resolve(
|
||
agent_type="memory_recall", task_type="embedding"
|
||
)
|
||
if not endpoints:
|
||
raise RecoverableAgentError("没有可用的 embedding 端点，无法投影长期记忆")
|
||
execution = await self.knowledge_embedder.embed(endpoints, text)
|
||
return list(execution.vector)
|
||
|
||
async def aggregate_episodes(self, *, customer_limit: int = 50) -> int:
|
||
"""把已静默的会话片段聚合为 episode，返回新写入的片段数。
|
||
|
||
可重复调用：同一片段的 `content_hash` 命中唯一键时直接跳过，
|
||
不需要额外的幂等表。
|
||
"""
|
||
# 必须由调用方开启并提交事务：EpisodeWorker 只做 add/flush（savepoint 幂等），
|
||
# 自己不提交。漏掉提交会同时出现"返回已插入"与"表里没有行"。
|
||
async with SessionFactory() as session, session.begin():
|
||
# customer_id 列可空：SQL 的 is_not(None) 不足以让类型检查器收窄，
|
||
# 这里显式再过滤一次。
|
||
raw_customers = list(await session.scalars(
|
||
select(ConversationMessage.customer_id)
|
||
.where(ConversationMessage.customer_id.is_not(None))
|
||
.distinct()
|
||
.limit(max(1, customer_limit))
|
||
))
|
||
customers = [int(item) for item in raw_customers if item is not None]
|
||
worker = EpisodeWorker(session)
|
||
inserted = 0
|
||
for customer_id in customers:
|
||
result = await worker.aggregate(customer_id)
|
||
inserted += len(result.inserted)
|
||
return inserted
|
||
|
||
async def consume_episodes(
|
||
self, *, limit: int = EPISODE_CONSUME_LIMIT
|
||
) -> EpisodeConsumptionResult:
|
||
"""消费一批待提取片段（低频批处理），把片段提升为长期记忆。
|
||
|
||
必须由调用方开启并提交事务：消费者只做 flush 与状态更新，自己不提交，
|
||
否则会出现"返回已处理但表里状态没变"。单个片段失败不冒泡（在消费者内部
|
||
转换为 `失败` + 重试计数），因此本方法只在数据库层面失败时抛错。
|
||
"""
|
||
async with SessionFactory() as session, session.begin():
|
||
return await EpisodeExtractionConsumer(
|
||
session,
|
||
extractor=self.memory_extraction,
|
||
cache=self.memory_cache,
|
||
limit=limit,
|
||
).consume_pending()
|
||
|
||
async def execute(self, run_id: str) -> bool:
|
||
# A new fencing token for each claim also fences restarts of the same process.
|
||
worker_id = str(uuid4())
|
||
async with SessionFactory() as session, session.begin():
|
||
run = await session.scalar(select(AgentRun).where(
|
||
AgentRun.run_id == run_id).with_for_update())
|
||
if run is None:
|
||
return False
|
||
if run.status == "cancel_requested":
|
||
# 文档 §6.4：运行真正落到 cancelled 时，原请求在 request_idempotency 中
|
||
# 以 failed + RUN_CANCELLED 结束（HTTP 层取消受理时可能已写过一次，
|
||
# 这里幂等覆盖，保证 worker 抢先落终态的场景也不漏）。
|
||
run.status = "cancelled"
|
||
run.completed_at = datetime.now(UTC).replace(tzinfo=None)
|
||
run.locked_until = None
|
||
await session.execute(update(RequestIdempotency).where(
|
||
RequestIdempotency.id == run.idempotency_id
|
||
).values(status="failed", error_code="RUN_CANCELLED", updated_at=run.completed_at))
|
||
return True
|
||
claimed = await AgentRunRepository(session).claim(
|
||
run_id, worker_id, self.settings.worker_lease_seconds)
|
||
if not claimed:
|
||
return False
|
||
task = asyncio.create_task(self._execute_claimed(run_id, worker_id))
|
||
heartbeat = asyncio.create_task(self._heartbeat(run_id, worker_id, task))
|
||
try:
|
||
await task
|
||
except asyncio.CancelledError:
|
||
await self._failure(run_id, worker_id, "RUN_INTERRUPTED", retryable=True)
|
||
# Runtime cancellation is shutdown; a lost lease only cancels the child.
|
||
current = asyncio.current_task()
|
||
if current is not None and current.cancelling():
|
||
raise
|
||
except RunLeaseLostError:
|
||
await self._failure(run_id, worker_id, "RUN_LEASE_LOST", retryable=True)
|
||
except Exception as exc:
|
||
code = exc.code if isinstance(exc, AgentError) else "AGENT_INTERNAL_ERROR"
|
||
await self._failure(run_id, worker_id, code,
|
||
retryable=isinstance(exc, RecoverableAgentError))
|
||
# 记录堆栈：只记录异常类型会让线上排障无从下手（run 的 error_code 只有
|
||
# AGENT_INTERNAL_ERROR，看不到真实原因）。
|
||
logger.warning("run failed run_id=%s error_type=%s", run_id, type(exc).__name__,
|
||
exc_info=True)
|
||
finally:
|
||
heartbeat.cancel()
|
||
with contextlib.suppress(asyncio.CancelledError):
|
||
await heartbeat
|
||
return True
|
||
|
||
async def _heartbeat(
|
||
self, run_id: str, worker_id: str, task: asyncio.Task[None]
|
||
) -> None:
|
||
try:
|
||
while True:
|
||
await asyncio.sleep(self.settings.worker_lease_seconds / 3)
|
||
async with SessionFactory() as session, session.begin():
|
||
renewed = await AgentRunRepository(session).renew(
|
||
run_id, worker_id, self.settings.worker_lease_seconds)
|
||
if not renewed:
|
||
task.cancel()
|
||
return
|
||
except Exception:
|
||
task.cancel()
|
||
logger.warning("lease renewal failed run_id=%s", run_id)
|
||
|
||
async def _execute_claimed(self, run_id: str, worker_id: str) -> None:
|
||
async with SessionFactory() as session:
|
||
run = await AgentRunRepository(session).get(run_id)
|
||
if run is None:
|
||
raise ValueError("run not found")
|
||
message = await session.get(ConversationMessage, run.request_message_id)
|
||
idem = await session.get(RequestIdempotency, run.idempotency_id)
|
||
event = await session.scalar(select(DomainEventOutbox).where(
|
||
DomainEventOutbox.aggregate_id == run_id,
|
||
DomainEventOutbox.event_type == "agent.run_requested").limit(1))
|
||
if message is None or idem is None:
|
||
raise ValueError("run input missing")
|
||
# 短期会话记忆：加载本次之前的对话，模型靠它解析指代。
|
||
history = await self._conversation_history(
|
||
session, session_id=run.session_id, user_id=int(run.user_id),
|
||
before_message_id=run.request_message_id,
|
||
)
|
||
request = AgentRequest(
|
||
agent_type=run.agent_type, message=message.content, session_id=run.session_id,
|
||
idempotency_key=idem.idempotency_key,
|
||
metadata=event.payload.get("metadata", {}) if event else {},
|
||
history=history,
|
||
)
|
||
identity = RequestContext(user_id=str(run.user_id), trace_id=run.trace_id)
|
||
# Re-check account and permissions at execution time, including delayed jobs.
|
||
context = await self.resolve_identity(identity)
|
||
result: AgentResult | None = None
|
||
async for event_data in AgentExecutor(self.factory).execute(
|
||
request.agent_type, request, context, run_id
|
||
):
|
||
if event_data.event_type == "done":
|
||
result = AgentResult.model_validate(event_data.payload["result"])
|
||
if result is None:
|
||
raise ValueError("Agent produced no terminal result")
|
||
# 业务事件查询必须用独立 Session：在 complete_run 的 Session 上先跑 SELECT 会触发
|
||
# SQLAlchemy 的 autobegin，使 complete_run 内部的 session.begin() 抛
|
||
# "A transaction is already begun on this Session"，运行直接失败。
|
||
async with SessionFactory() as events_session:
|
||
business_events = list(await events_session.scalars(
|
||
select(DomainEventOutbox.event_type).where(
|
||
DomainEventOutbox.aggregate_id == run_id,
|
||
DomainEventOutbox.event_type.in_(tuple(BUSINESS_EVENT_TYPES)),
|
||
)
|
||
))
|
||
async with SessionFactory() as session:
|
||
await AgentPersistenceService(session).complete_run(
|
||
run_id, result, worker_id=worker_id,
|
||
memory_extraction_requested=MemoryService.should_extract_memory(
|
||
conversation_content=request.message,
|
||
role="user",
|
||
# 工具产出的权威事实同样构成持久记忆（工具调用记录来自终态结果）。
|
||
tool_result=any(
|
||
call.status == "succeeded" for call in result.result.tool_calls
|
||
),
|
||
# 本 run 落库的业务事件（风险评估完成、交易完成等）。
|
||
event_type=business_events[0] if business_events else None,
|
||
# 用户明确陈述的偏好/约束/身份/目标，命中才触发抽取。
|
||
signals=MemoryService.detect_memory_signals(request.message),
|
||
),
|
||
)
|
||
|
||
async def _failure(
|
||
self, run_id: str, worker_id: str, error_code: str, *, retryable: bool
|
||
) -> None:
|
||
now = datetime.now(UTC).replace(tzinfo=None)
|
||
async with SessionFactory() as session, session.begin():
|
||
run = await session.scalar(select(AgentRun).where(
|
||
AgentRun.run_id == run_id).with_for_update())
|
||
if run is None or run.worker_id != worker_id:
|
||
return
|
||
if run.status not in {"running", "cancel_requested"}:
|
||
return
|
||
if run.status == "cancel_requested":
|
||
run.status = "cancelled"
|
||
elif retryable and run.attempt_count < self.settings.worker_retry_limit:
|
||
run.status = "queued"
|
||
else:
|
||
run.status = "failed"
|
||
run.error_code, run.updated_at = error_code, now
|
||
run.locked_until = (
|
||
now + timedelta(seconds=min(60, 2**run.attempt_count))
|
||
if run.status == "queued" else None
|
||
)
|
||
if run.status != "queued":
|
||
run.completed_at = now
|
||
# 取消导致的终态属于“原请求已终止”，按文档 §6.4 在 request_idempotency
|
||
# 中写 failed + RUN_CANCELLED；`agent_run.error_code` 仍保留真实触发原因
|
||
#（如 RUN_INTERRUPTED），两者层级不同。
|
||
await session.execute(update(RequestIdempotency).where(
|
||
RequestIdempotency.id == run.idempotency_id
|
||
).values(
|
||
status="failed",
|
||
error_code="RUN_CANCELLED" if run.status == "cancelled" else error_code,
|
||
updated_at=now,
|
||
))
|
||
session.add(InteractionAudit(
|
||
actor_type="agent", actor_id=run.user_id, session_id=run.session_id,
|
||
portal="api", action_type=f"agent.run_{run.status}",
|
||
detail={"run_id": run_id, "error_code": error_code}, created_at=now,
|
||
))
|