df76edcf55
Some checks failed
CI / test (push) Failing after 33s
架构保养候选 3：探索生命周期的账本规则与状态机从 HTTP 层落入 exploration/lifecycle.py（open/conduct/close）与 patrol.py（巡检读模型）， 违规改用类型化领域异常（NotFound/Guardrail/Channel），router 瘦回纯 HTTP 翻译（404/409/502 映射），领域层不再依赖 fastapi，可脱离 TestClient 直测（新增 12 个单元测试）。
188 lines
7.3 KiB
Python
188 lines
7.3 KiB
Python
"""Exploration session lifecycle (探索会话生命周期).
|
||
|
||
会话三态的领域操作：open_session / conduct_turn / close_session。
|
||
平台账本（预算、状态机、轮次记账）全部在此执行，违规抛类型化领域
|
||
异常（errors.py），由 router 映射 HTTP 状态码；通道收发沿用
|
||
ChannelFactory 既有接缝。
|
||
"""
|
||
|
||
import json
|
||
from typing import Any, Optional
|
||
|
||
from sqlmodel import Session
|
||
|
||
from agenteval.channels.factory import ChannelFactory
|
||
from agenteval.config import get_settings
|
||
from agenteval.exploration.errors import (
|
||
ExplorationChannelError,
|
||
ExplorationGuardrailError,
|
||
ExplorationNotFoundError,
|
||
)
|
||
from agenteval.exploration.judge import start_judge_review
|
||
from agenteval.exploration.models import (
|
||
ExplorationBudget,
|
||
ExplorationMessage,
|
||
ExplorationSession,
|
||
ExplorationSessionStatus,
|
||
ExplorationTrigger,
|
||
normalize_experience,
|
||
resolve_budget,
|
||
)
|
||
from agenteval.models import CampaignStatus
|
||
from agenteval.storage.db import as_utc, utc_now
|
||
from agenteval.storage.repository import (
|
||
CampaignRepository,
|
||
ExplorationMessageRepository,
|
||
ExplorationSessionRepository,
|
||
TargetRepository,
|
||
)
|
||
|
||
|
||
def coerce_reply_text(content: Any) -> str:
|
||
"""Flatten a reply payload to text; tutu returns msgBody as a parsed object,
|
||
and str(dict) would leak a Python repr into the view."""
|
||
if isinstance(content, str):
|
||
return content
|
||
if isinstance(content, dict):
|
||
for key in ("content", "text", "message"):
|
||
value = content.get(key)
|
||
if isinstance(value, str) and value:
|
||
return value
|
||
if content is None:
|
||
return ""
|
||
return json.dumps(content, ensure_ascii=False)
|
||
|
||
|
||
def _check_creation_guardrails(
|
||
campaign,
|
||
triggered_by: ExplorationTrigger,
|
||
budget: ExplorationBudget,
|
||
repo: ExplorationSessionRepository,
|
||
) -> None:
|
||
if campaign.status != CampaignStatus.RUNNING:
|
||
raise ExplorationGuardrailError("活动不在进行中，无法创建探索会话")
|
||
if campaign.time_scale != 1 and triggered_by != ExplorationTrigger.MANUAL:
|
||
raise ExplorationGuardrailError("加速调试线仅允许手动创建探索会话（时间压缩与拟真相冲突）")
|
||
sessions = repo.list_by_campaign(campaign.id)
|
||
if len(sessions) >= budget.max_sessions:
|
||
raise ExplorationGuardrailError(f"探索会话数超出预算：本活动窗口最多 {budget.max_sessions} 个会话")
|
||
if sessions:
|
||
latest = max(s.created_at for s in sessions if s.created_at)
|
||
elapsed = (utc_now() - as_utc(latest)).total_seconds()
|
||
if elapsed < budget.min_interval_seconds:
|
||
wait_minutes = budget.min_interval_seconds // 60
|
||
raise ExplorationGuardrailError(f"相邻探索会话间隔不足：最小间隔 {wait_minutes} 分钟，请稍后再试")
|
||
|
||
|
||
def open_session(
|
||
db_session: Session,
|
||
*,
|
||
campaign_id: str,
|
||
persona: dict[str, Any],
|
||
goal: str,
|
||
seed_ref: Optional[dict[str, Any]] = None,
|
||
triggered_by: ExplorationTrigger = ExplorationTrigger.AUTO,
|
||
) -> ExplorationSession:
|
||
"""按账本规则开一个 running 会话；违规抛领域异常。"""
|
||
campaign = CampaignRepository(db_session).get(campaign_id)
|
||
if not campaign:
|
||
raise ExplorationNotFoundError("campaign not found")
|
||
if not TargetRepository(db_session).get(campaign.target_id):
|
||
raise ExplorationNotFoundError("campaign target not found")
|
||
|
||
budget = resolve_budget(campaign)
|
||
repo = ExplorationSessionRepository(db_session)
|
||
_check_creation_guardrails(campaign, triggered_by, budget, repo)
|
||
|
||
return repo.create(ExplorationSession(
|
||
campaign_id=campaign.id,
|
||
target_id=campaign.target_id,
|
||
persona=persona,
|
||
goal=goal,
|
||
seed_ref=seed_ref,
|
||
triggered_by=triggered_by,
|
||
))
|
||
|
||
|
||
async def conduct_turn(db_session: Session, *, session_id: str, content: str) -> dict[str, Any]:
|
||
"""一轮完整问答：状态机拒收 → 轮次预算记账 → 通道往返 → 双条落库。
|
||
|
||
消息已送达即消耗一轮预算（平台账本）：先落用户消息，再等回复。
|
||
"""
|
||
repo = ExplorationSessionRepository(db_session)
|
||
session_obj = repo.get(session_id)
|
||
if not session_obj:
|
||
raise ExplorationNotFoundError("exploration session not found")
|
||
if session_obj.status != ExplorationSessionStatus.RUNNING:
|
||
raise ExplorationGuardrailError("探索会话不在进行中，拒收消息")
|
||
|
||
campaign = CampaignRepository(db_session).get(session_obj.campaign_id)
|
||
budget = resolve_budget(campaign) if campaign else ExplorationBudget()
|
||
if session_obj.turn_count >= budget.max_turns:
|
||
raise ExplorationGuardrailError(f"会话轮数超出预算：单会话最多 {budget.max_turns} 轮")
|
||
|
||
target = TargetRepository(db_session).get(session_obj.target_id)
|
||
if not target:
|
||
raise ExplorationNotFoundError("session target not found")
|
||
|
||
channel = ChannelFactory.create(target)
|
||
sent_at = utc_now()
|
||
try:
|
||
send_result = await channel.send(content)
|
||
except Exception as exc: # channel adapters raise transport-specific errors
|
||
raise ExplorationChannelError(f"评测对象通道发送失败: {exc}") from exc
|
||
if not send_result.ok:
|
||
raise ExplorationChannelError(f"评测对象通道发送失败: {send_result.error}")
|
||
|
||
message_repo = ExplorationMessageRepository(db_session)
|
||
round_index = session_obj.turn_count + 1
|
||
message_repo.save_message(
|
||
ExplorationMessage(
|
||
session_id=session_obj.id, round_index=round_index, role="user", content=content, created_at=sent_at
|
||
)
|
||
)
|
||
session_obj.turn_count = round_index
|
||
repo.update(session_obj)
|
||
|
||
try:
|
||
reply = await channel.poll_reply(
|
||
send_result.question_msg_id,
|
||
timeout=get_settings().poll_reply_timeout,
|
||
)
|
||
except Exception as exc:
|
||
raise ExplorationChannelError(f"等待评测对象回复失败: {exc}") from exc
|
||
if reply is None:
|
||
raise ExplorationChannelError("等待评测对象回复超时")
|
||
|
||
received_at = utc_now()
|
||
latency_ms = int((received_at - sent_at).total_seconds() * 1000)
|
||
reply_text = coerce_reply_text(reply.content)
|
||
message_repo.save_message(
|
||
ExplorationMessage(
|
||
session_id=session_obj.id,
|
||
round_index=round_index,
|
||
role="assistant",
|
||
content=reply_text,
|
||
latency_ms=latency_ms,
|
||
created_at=received_at,
|
||
)
|
||
)
|
||
return {"reply": reply_text, "latency_ms": latency_ms, "turn_count": round_index}
|
||
|
||
|
||
def close_session(db_session: Session, *, session_id: str, experience: dict[str, Any]) -> ExplorationSession:
|
||
"""关闭会话：体验记录归一化 → 状态迁移 → 触发 judge 抽样复核。"""
|
||
repo = ExplorationSessionRepository(db_session)
|
||
session_obj = repo.get(session_id)
|
||
if not session_obj:
|
||
raise ExplorationNotFoundError("exploration session not found")
|
||
if session_obj.status != ExplorationSessionStatus.RUNNING:
|
||
raise ExplorationGuardrailError("探索会话不在进行中，无法关闭")
|
||
|
||
session_obj.experience = normalize_experience(experience)
|
||
session_obj.status = ExplorationSessionStatus.COMPLETED
|
||
session_obj.closed_at = utc_now()
|
||
updated = repo.update(session_obj)
|
||
start_judge_review(session_obj.id)
|
||
return updated
|