631 lines
32 KiB
Python
631 lines
32 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, AgentRequestMetadata, 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, HandoverTicket, RequestIdempotency
|
||
from app.repository.agent_run_repository import AgentRunRepository
|
||
from app.service.agent.bootstrap import (
|
||
get_agent_factory,
|
||
get_memory_cache_adapter,
|
||
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.customer_service_session_memory_service import (
|
||
CustomerServiceSessionMemory,
|
||
CustomerServiceSessionTurn,
|
||
build_customer_service_session_memory,
|
||
)
|
||
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.customer_profile_candidate_worker import CustomerProfileCandidateWorker
|
||
from app.worker.episode_worker import (
|
||
EpisodeConsumptionResult,
|
||
EpisodeExtractionConsumer,
|
||
EpisodeWorker,
|
||
)
|
||
from app.worker.memory_extraction_worker import MemoryExtractionWorker
|
||
from app.worker.outbox_worker import 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"
|
||
|
||
|
||
@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,
|
||
session_memory: CustomerServiceSessionMemory | None = None,
|
||
) -> 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 删除客户端：当前组装层没有提供（bootstrap 只装配召回用的读适配器），
|
||
# 因此默认 None = 投影清理显式降级并留痕，绝不写成"删除成功"。
|
||
self.projection_cleaner = projection_cleaner
|
||
# 该组件是客服当前会话 Redis List，不是长期记忆召回缓存或客户画像。
|
||
self.session_memory = (
|
||
session_memory
|
||
if session_memory is not None
|
||
else build_customer_service_session_memory()
|
||
)
|
||
# episode 聚合是低频批处理，按轮次节流而不是每轮都查。
|
||
self._episode_rounds = 0
|
||
|
||
async def restore_context(
|
||
self, *, actor_type: str, actor_id: str, trace_id: str
|
||
) -> RequestContext:
|
||
"""按已验证的内部事件身份恢复执行上下文。"""
|
||
identity = RequestContext(user_id=actor_id, trace_id=trace_id)
|
||
if actor_type == "visitor":
|
||
return identity.model_copy(update={
|
||
"roles": ("visitor",),
|
||
# 与访客 JWT 对齐，只恢复公开 Agent 和公开知识的最小权限。
|
||
"permissions": ("agent:run", "knowledge:query"),
|
||
"data_scope": "public",
|
||
})
|
||
return await self.resolve_identity(identity)
|
||
|
||
@staticmethod
|
||
def should_request_memory_extraction(
|
||
*, agent_type: str, context: RequestContext, message: str, result: AgentResult,
|
||
business_events: tuple[str, ...] | list[str],
|
||
) -> bool:
|
||
"""只允许非客服 Agent 的已登录用户业务事实进入客户记忆抽取队列。"""
|
||
# 客服只使用未来受控的会话短期上下文，绝不沉淀为长期画像或客户偏好。
|
||
if agent_type == "customer_service" or "visitor" in context.roles:
|
||
return False
|
||
return MemoryService.should_extract_memory(
|
||
conversation_content=message,
|
||
role="user",
|
||
tool_result=any(call.status == "succeeded" for call in result.result.tool_calls),
|
||
event_type=business_events[0] if business_events else None,
|
||
signals=MemoryService.detect_memory_signals(message),
|
||
)
|
||
|
||
@staticmethod
|
||
def should_request_profile_candidate(
|
||
*, agent_type: str, context: RequestContext, message: str,
|
||
) -> bool:
|
||
"""客服只为已登录客户生成候选，不为访客生成正式画像数据。"""
|
||
if agent_type != "customer_service" or "visitor" in context.roles:
|
||
return False
|
||
if not {"customer", "authenticated_user"}.intersection(context.roles):
|
||
return False
|
||
if context.data_scope != "self":
|
||
return False
|
||
# 只有明确的长期偏好、约束或目标才进入候选队列，普通问答不触发模型抽取。
|
||
return bool(MemoryService.detect_memory_signals(message))
|
||
|
||
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 ValueError("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 ValueError("memory extraction payload is incomplete")
|
||
# 幂等键只认事件 id，由 worker 自己按 payload 回查，避免调用方漏传。
|
||
# 注入召回缓存适配器：写入生效后立即失效该客户的热缓存。
|
||
await MemoryExtractionWorker(
|
||
session, extractor=self.memory_extraction, cache=self.memory_cache
|
||
).handle(payload)
|
||
|
||
async def dispatch_profile_candidate(payload: dict[str, Any]) -> None:
|
||
if "message_id" not in payload or "customer_id" not in payload:
|
||
raise ValueError("profile candidate payload is incomplete")
|
||
await CustomerProfileCandidateWorker(
|
||
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 ValueError("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_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 ValueError("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", "")),
|
||
)
|
||
|
||
async def dispatch_handover_queue_ready(payload: dict[str, Any]) -> None:
|
||
# 事件成功消费只表示管理员队列可查看，不能修改 pending 状态，更不能向用户
|
||
# 表述人工已经接单、处理或承诺时效。
|
||
ticket_no = str(payload.get("ticket_no", "")).strip()
|
||
if not ticket_no:
|
||
raise ValueError("conversation.transfer_requested payload is incomplete")
|
||
ticket = await session.scalar(
|
||
select(HandoverTicket).where(HandoverTicket.ticket_no == ticket_no)
|
||
)
|
||
if ticket is None:
|
||
raise ValueError("handover ticket not found")
|
||
session.add(InteractionAudit(
|
||
actor_type="system",
|
||
actor_id=None,
|
||
target_customer_id=ticket.customer_id,
|
||
session_id=ticket.session_id,
|
||
portal="worker",
|
||
action_type="handover.queue_ready",
|
||
detail={
|
||
"ticket_no": ticket.ticket_no,
|
||
"source_agent": ticket.source_agent,
|
||
"reason_code": ticket.reason_code,
|
||
"ticket_status": ticket.status,
|
||
},
|
||
created_at=datetime.now(UTC).replace(tzinfo=None),
|
||
))
|
||
await session.flush()
|
||
|
||
handlers: dict[str, Callable[[dict[str, Any]], Awaitable[None]]] = {
|
||
"agent.run_requested": dispatch,
|
||
"memory.extraction_requested": dispatch_memory_extraction,
|
||
"customer_profile.candidate_requested": dispatch_profile_candidate,
|
||
"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,
|
||
"conversation.transfer_requested": dispatch_handover_queue_ready,
|
||
}
|
||
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()
|
||
|
||
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
|
||
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 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")
|
||
metadata = event.payload.get("metadata", {}) if event else {}
|
||
request = AgentRequest(
|
||
agent_type=run.agent_type, message=message.content, session_id=run.session_id,
|
||
idempotency_key=idem.idempotency_key,
|
||
metadata=AgentRequestMetadata.model_validate(metadata),
|
||
)
|
||
actor_type = (
|
||
str(event.payload.get("actor_type", "authenticated"))
|
||
if event else "authenticated"
|
||
)
|
||
actor_id = str(run.user_id)
|
||
trace_id = run.trace_id
|
||
context = await self.restore_context(
|
||
actor_type=actor_type, actor_id=actor_id, trace_id=trace_id
|
||
)
|
||
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=self.should_request_memory_extraction(
|
||
agent_type=run.agent_type, context=context,
|
||
message=request.message,
|
||
result=result,
|
||
business_events=business_events,
|
||
),
|
||
profile_candidate_requested=self.should_request_profile_candidate(
|
||
agent_type=run.agent_type, context=context, message=request.message,
|
||
),
|
||
)
|
||
# 只有数据库成功保存完整用户/助手轮次后才写短期 Redis；Redis 故障不应让已完成
|
||
# 的客服回答回滚或重试。短期组件会再次脱敏，形成持久化链路的第二道保护。
|
||
await self._append_customer_service_session_memory(
|
||
agent_type=run.agent_type, actor_id=str(run.user_id), session_id=run.session_id,
|
||
request_message=request.message, response_message=result.result.text,
|
||
)
|
||
|
||
async def _append_customer_service_session_memory(
|
||
self, *, agent_type: str, actor_id: str, session_id: str,
|
||
request_message: str, response_message: str,
|
||
) -> None:
|
||
"""将一轮成功客服会话写入 Redis；此派生写入永不影响已提交的主事务。"""
|
||
if agent_type != "customer_service":
|
||
return
|
||
try:
|
||
await self.session_memory.append(
|
||
actor_id=actor_id, session_id=session_id,
|
||
turns=(
|
||
CustomerServiceSessionTurn(role="user", content=request_message),
|
||
CustomerServiceSessionTurn(role="assistant", content=response_message),
|
||
),
|
||
)
|
||
except Exception:
|
||
# 注入的第三方 Redis 适配器也可能违反“不抛异常”约定，Worker 仍必须隔离故障。
|
||
logger.warning("客服短期会话写入降级，不影响已完成的客服运行", exc_info=True)
|
||
|
||
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,
|
||
))
|