diff --git a/apps/api/deps.py b/apps/api/deps.py index 1648c5a..4af8947 100644 --- a/apps/api/deps.py +++ b/apps/api/deps.py @@ -15,6 +15,7 @@ from fastapi import Depends, Request from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer from sqlalchemy.ext.asyncio import AsyncSession +from packages.agent.context import PROFILE_SUMMARY_MAX_CHARS from packages.agent.service import AgentService from packages.analytics.service import AnalyticsService from packages.billing.service import BillingService @@ -34,6 +35,7 @@ from packages.llm.providers.deepseek import DeepSeekProvider from packages.llm.providers.zhipu import ZhipuGLMProvider from packages.llm.recording import PrometheusAICallRecorder, SqlAlchemyAICallRecorder from packages.llm.routing import SceneRouter, build_scene_configs +from packages.report.profile import latest_profile_summary from packages.shared.config import Settings, get_settings from packages.shared.context import bind_context from packages.shared.db import get_session_factory @@ -210,8 +212,22 @@ def agent_service_dep( session: SessionDep, gateway: GatewayDep, billing: Annotated[BillingService, Depends(billing_service_dep)], + settings: SettingsDep, ) -> AgentService: - return AgentService(session, gateway=gateway, billing=billing) + # agent 与 report 在 import-linter 分层里同层、不允许直接互调，画像摘要 + # 以闭包在 apps 层接线；report 域内部仍只有 report.repo 触 rp_report。 + async def profile_provider(user_id: int) -> str | None: + return await latest_profile_summary( + session, user_id=user_id, max_chars=PROFILE_SUMMARY_MAX_CHARS + ) + + return AgentService( + session, + gateway=gateway, + billing=billing, + profile_provider=profile_provider, + history_turns=settings.agent_history_turns, + ) IdentityServiceDep = Annotated[IdentityService, Depends(identity_service_dep)] diff --git a/apps/api/main.py b/apps/api/main.py index 8de88f8..bebd51b 100644 --- a/apps/api/main.py +++ b/apps/api/main.py @@ -21,6 +21,7 @@ from apps.api.routers import ( kb, quota, reports, + reservations, tasks, uploads, ) @@ -90,6 +91,7 @@ def create_app(settings: Settings | None = None) -> FastAPI: uploads.router, kb.router, reports.router, + reservations.router, ) for router in routers: app.include_router(router, prefix=cfg.api_prefix) diff --git a/apps/api/routers/agent.py b/apps/api/routers/agent.py index 3389ac3..edd195a 100755 --- a/apps/api/routers/agent.py +++ b/apps/api/routers/agent.py @@ -1,9 +1,12 @@ -"""``POST /agent/chat`` — the vertical slice. +"""``POST /agent/chat`` + 会话持久化接口（FR-AGT-09 / T-P4-01）。 One request exercises: JWT auth → quota reserve → LLM gateway (routing, prompt version, cache-prefix check, daily budget) → ``ob_ai_call`` cost row → quota -commit/refund → unified envelope. Streaming (SSE, FR-AGT-01) is intentionally -out of scope for the scaffold. +commit/refund → ``ag_message`` 双行落库 → unified envelope. Streaming (SSE, +FR-AGT-01) is intentionally out of scope for the scaffold. + +会话接口：创建 / 列表 / 历史。chat 不带 conversation_id 时隐式建会话并在 +响应里带回；带则做归属校验（他人会话与不存在同口径 404，防 ID 遍历）。 """ from __future__ import annotations @@ -13,7 +16,13 @@ import uuid from fastapi import APIRouter from apps.api.deps import AgentServiceDep, CurrentUserId -from apps.api.schemas import ChatRequest, ChatResponse, ChatUsage +from apps.api.schemas import ( + ChatMessageItem, + ChatRequest, + ChatResponse, + ChatUsage, + ConversationResponse, +) from packages.llm.errors import ( AllProvidersFailedError, CachePrefixViolationError, @@ -22,11 +31,62 @@ from packages.llm.errors import ( StructuredOutputError, ) from packages.shared.errors import AppError, ErrorCode +from packages.shared.models import AgConversation from packages.shared.response import ok router = APIRouter(prefix="/agent", tags=["agent"]) +def _conversation_payload(conversation: AgConversation) -> ConversationResponse: + return ConversationResponse( + conversation_id=str(conversation.conversation_id), + title=conversation.title, + message_count=conversation.message_count, + last_message_at=conversation.last_message_at, + created_at=conversation.created_at, + ) + + +@router.post("/conversations", summary="创建空会话（FR-AGT-09）") +async def create_conversation( + user_id: CurrentUserId, + agent: AgentServiceDep, +) -> dict[str, object]: + conversation = await agent.create_conversation(user_id=user_id) + return ok(_conversation_payload(conversation).model_dump(mode="json")) + + +@router.get("/conversations", summary="会话列表（最近活跃倒序）") +async def list_conversations( + user_id: CurrentUserId, + agent: AgentServiceDep, +) -> dict[str, object]: + conversations = await agent.list_conversations(user_id=user_id) + return ok([_conversation_payload(c).model_dump(mode="json") for c in conversations]) + + +@router.get("/conversations/{conversation_id}/messages", summary="会话历史（时间正序）") +async def list_messages( + conversation_id: int, + user_id: CurrentUserId, + agent: AgentServiceDep, +) -> dict[str, object]: + messages = await agent.list_messages(user_id=user_id, conversation_id=conversation_id) + return ok( + [ + ChatMessageItem( + message_id=str(m.message_id), + role=m.role, + content=m.content, + model=m.model, + call_id=str(m.call_id) if m.call_id is not None else None, + created_at=m.created_at, + ).model_dump(mode="json") + for m in messages + ] + ) + + @router.post("/chat", summary="Single-turn styling chat (non-streaming)") async def chat( payload: ChatRequest, @@ -35,7 +95,12 @@ async def chat( ) -> dict[str, object]: request_key = payload.request_key or uuid.uuid4().hex try: - result = await agent.chat(user_id=user_id, message=payload.message, request_key=request_key) + result = await agent.chat( + user_id=user_id, + message=payload.message, + request_key=request_key, + conversation_id=payload.conversation_id, + ) except DailyBudgetExceededError as exc: raise AppError(ErrorCode.LLM_DAILY_BUDGET_EXCEEDED) from exc except SceneDegradedError as exc: @@ -50,6 +115,7 @@ async def chat( return ok( ChatResponse( + conversation_id=str(result.conversation_id), reply=result.reply, provider=result.provider, model=result.model, diff --git a/apps/api/routers/reports.py b/apps/api/routers/reports.py index ecf55e9..2313ff8 100644 --- a/apps/api/routers/reports.py +++ b/apps/api/routers/reports.py @@ -17,7 +17,11 @@ from fastapi.responses import FileResponse from apps.api.deps import BudgetGuardDep, CurrentUserId, SessionDep, SettingsDep from packages.report.export_service import ReportExportService +from packages.report.feedback_service import FeedbackService from packages.report.schemas import ( + FeedbackCreateRequest, + FeedbackItem, + FeedbackUpsertResponse, ReportCreateRequest, ReportExportRequest, ReportExportResponse, @@ -88,6 +92,59 @@ def report_export_service_dep( ReportExportServiceDep = Annotated[ReportExportService, Depends(report_export_service_dep)] +def feedback_service_dep(session: SessionDep) -> FeedbackService: + return FeedbackService(session) + + +FeedbackServiceDep = Annotated[FeedbackService, Depends(feedback_service_dep)] + + +@router.post( + "/{report_id}/feedback", + summary="报告维度结论反馈（T-P3-21：同维度重复 = 覆盖更新，upsert 语义）", +) +async def upsert_report_feedback( + report_id: int, + payload: FeedbackCreateRequest, + user_id: CurrentUserId, + service: FeedbackServiceDep, +) -> dict[str, object]: + upsert = await service.upsert_feedback( + user_id=user_id, + report_id=report_id, + dimension_key=payload.dimension_key, + verdict=payload.verdict, + correction=payload.correction, + ) + return ok( + FeedbackUpsertResponse( + dimension_key=upsert.feedback.dimension_key, + verdict=upsert.feedback.verdict, + updated=upsert.updated, + ).model_dump() + ) + + +@router.get("/{report_id}/feedback", summary="报告已提交的维度反馈列表（归属校验）") +async def list_report_feedback( + report_id: int, + user_id: CurrentUserId, + service: FeedbackServiceDep, +) -> dict[str, object]: + rows = await service.list_feedback(user_id=user_id, report_id=report_id) + return ok( + [ + FeedbackItem( + dimension_key=row.dimension_key, + verdict=row.verdict, + correction=row.correction, + created_at=row.created_at, + ).model_dump() + for row in rows + ] + ) + + @router.post("", summary="生成第一份美学报告（终身一次，走 report_basic 配额）") async def create_report( payload: ReportCreateRequest, diff --git a/apps/api/routers/reservations.py b/apps/api/routers/reservations.py new file mode 100644 index 0000000..8c3fc86 --- /dev/null +++ b/apps/api/routers/reservations.py @@ -0,0 +1,55 @@ +"""``POST /reservations``、``GET /reservations``（T-P3-22 预约留资）。 + +topic 预留通用性，首期只有 full_report（全面报告上线提醒）。 +幂等口径：重复预约返回已有记录（created=false，HTTP 仍 200）。 +""" + +from __future__ import annotations + +from typing import Annotated + +from fastapi import APIRouter, Depends, Query + +from apps.api.deps import CurrentUserId, SessionDep +from packages.report.reservation_service import ReservationService +from packages.report.schemas import ( + ReservationCreateRequest, + ReservationListResponse, + ReservationResponse, +) +from packages.shared.response import ok + +router = APIRouter(prefix="/reservations", tags=["reservations"]) + + +def reservation_service_dep(session: SessionDep) -> ReservationService: + return ReservationService(session) + + +ReservationServiceDep = Annotated[ReservationService, Depends(reservation_service_dep)] + + +@router.post("", summary="预约留资（幂等：重复预约 created=false）") +async def create_reservation( + payload: ReservationCreateRequest, + user_id: CurrentUserId, + service: ReservationServiceDep, +) -> dict[str, object]: + result = await service.reserve(user_id=user_id, topic=payload.topic) + return ok( + ReservationResponse( + topic=result.reservation.topic, + created=result.created, + created_at=result.reservation.created_at, + ).model_dump() + ) + + +@router.get("", summary="我的预约主题列表（topic 可选，缺省返回全部）") +async def list_reservations( + user_id: CurrentUserId, + service: ReservationServiceDep, + topic: str | None = Query(default=None, max_length=32), +) -> dict[str, object]: + topics = await service.list_topics(user_id=user_id, topic=topic) + return ok(ReservationListResponse(topics=topics).model_dump()) diff --git a/apps/api/schemas.py b/apps/api/schemas.py index 8003cac..19cb888 100755 --- a/apps/api/schemas.py +++ b/apps/api/schemas.py @@ -30,6 +30,12 @@ class DevTokenResponse(BaseModel): class ChatRequest(BaseModel): message: str = Field(min_length=1, max_length=2000) + conversation_id: str | None = Field( + default=None, + max_length=32, + description="会话 ID（字符串；雪花 ID 超 2^53，JS Number 会丢精度）。" + "缺省时新建会话并在响应中返回 (FR-AGT-09)", + ) request_key: str | None = Field( default=None, max_length=128, @@ -37,6 +43,25 @@ class ChatRequest(BaseModel): ) +class ConversationResponse(BaseModel): + conversation_id: str = Field(description="雪花 ID 以字符串返回，避免 JS 精度丢失") + title: str + message_count: int + last_message_at: datetime | None + created_at: datetime + + +class ChatMessageItem(BaseModel): + message_id: str + role: str + content: str + model: str | None + call_id: str | None = Field( + description="assistant 消息关联的 ob_ai_call 行（复盘锚点）；user 消息为 null" + ) + created_at: datetime + + class ChatUsage(BaseModel): tokens_in: int tokens_out: int @@ -46,6 +71,7 @@ class ChatUsage(BaseModel): class ChatResponse(BaseModel): + conversation_id: str reply: str provider: str model: str diff --git a/migrations/versions/0007_agent_conversations.py b/migrations/versions/0007_agent_conversations.py new file mode 100644 index 0000000..1665638 --- /dev/null +++ b/migrations/versions/0007_agent_conversations.py @@ -0,0 +1,78 @@ +"""Agent 会话与消息表（FR-AGT-09 / T-P4-01）。 + + ag_conversation — 会话主表；title 取首条用户消息截断，last_message_at 是 + 列表排序与 90 天保留期清理（合规方案 §6）的时间锚点 + ag_message — 消息表；retrieval_trace / rule_hits 是「每条回答可复盘」 + 的锚点，检索步骤（架构 §7.2 ④）落地前留空；assistant 行 + 以 call_id 关联 ob_ai_call，回复可一路追到成本账本 + +Reversible: 是。 +""" + +from __future__ import annotations + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op +from sqlalchemy.dialects import postgresql + +revision: str = "0007" +down_revision: str | None = "0006" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + op.create_table( + "ag_conversation", + sa.Column("conversation_id", sa.BigInteger(), nullable=False), + sa.Column("user_id", sa.BigInteger(), nullable=False), + sa.Column("title", sa.String(length=64), nullable=False, server_default=sa.text("''")), + sa.Column("message_count", sa.Integer(), nullable=False, server_default=sa.text("0")), + sa.Column("last_message_at", sa.DateTime(timezone=True), nullable=True), + sa.Column( + "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now() + ), + sa.PrimaryKeyConstraint("conversation_id"), + ) + op.create_index( + "ix_ag_conversation_user_active", "ag_conversation", ["user_id", "last_message_at"] + ) + + op.create_table( + "ag_message", + sa.Column("message_id", sa.BigInteger(), nullable=False), + sa.Column("conversation_id", sa.BigInteger(), nullable=False), + sa.Column("user_id", sa.BigInteger(), nullable=False), + sa.Column("role", sa.String(length=16), nullable=False), + sa.Column("content", sa.Text(), nullable=False), + sa.Column("call_id", sa.BigInteger(), nullable=True), + sa.Column("model", sa.String(length=64), nullable=True), + sa.Column("tokens_in", sa.Integer(), nullable=True), + sa.Column("tokens_out", sa.Integer(), nullable=True), + sa.Column("tokens_cached", sa.Integer(), nullable=True), + sa.Column("latency_ms", sa.Integer(), nullable=True), + sa.Column("cost_cents", sa.Numeric(14, 6), nullable=True), + sa.Column("retrieval_trace", postgresql.JSONB(), nullable=False), + sa.Column("rule_hits", postgresql.JSONB(), nullable=False), + sa.Column( + "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now() + ), + sa.PrimaryKeyConstraint("message_id"), + sa.ForeignKeyConstraint( + ["conversation_id"], ["ag_conversation.conversation_id"], ondelete="CASCADE" + ), + ) + op.create_index( + "ix_ag_message_conversation_created", "ag_message", ["conversation_id", "created_at"] + ) + op.create_index("ix_ag_message_user_created", "ag_message", ["user_id", "created_at"]) + + +def downgrade() -> None: + op.drop_index("ix_ag_message_user_created", table_name="ag_message") + op.drop_index("ix_ag_message_conversation_created", table_name="ag_message") + op.drop_table("ag_message") + op.drop_index("ix_ag_conversation_user_active", table_name="ag_conversation") + op.drop_table("ag_conversation") diff --git a/migrations/versions/0008_report_feedback.py b/migrations/versions/0008_report_feedback.py new file mode 100644 index 0000000..bd202a1 --- /dev/null +++ b/migrations/versions/0008_report_feedback.py @@ -0,0 +1,68 @@ +"""报告结论反馈 + 全面报告预约留资（T-P3-21 最小可用版 / T-P3-22）。 + + rp_feedback — 维度级结论反馈（准/不准 + 可选修正文本）； + UniqueConstraint(report_id, dimension_key) 承载 upsert 语义： + 同维度重复反馈 = 覆盖更新，不是堆行。 + 有意裁剪（T-P3-21 原文的「修正后置信度置 1.0 并即时写 L1 记忆」）： + L1 记忆系统未建，本迭代只落表，保证认同率/修正率两项指标 + 可从 SQL 直接算出，不回写报告结论。 + rp_reservation — 预约留资；topic 预留通用性（首期只有 full_report）。 + UniqueConstraint(user_id, topic) 承载幂等：重复预约返回已有 + 记录（created=false），不堆行。 + 指标口径：预约率 = rp_reservation(topic='full_report') 行数 ÷ + 埋点 full_report.card_view 去重曝光数；决策门槛 ≥8% + （产品方案 §2.2.3）。 + +Reversible: 是。 +""" + +from __future__ import annotations + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + +revision: str = "0008" +down_revision: str | None = "0007" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + op.create_table( + "rp_feedback", + sa.Column("feedback_id", sa.BigInteger(), nullable=False), + sa.Column("report_id", sa.BigInteger(), nullable=False), + sa.Column("user_id", sa.BigInteger(), nullable=False), + sa.Column("dimension_key", sa.String(length=64), nullable=False), + sa.Column("verdict", sa.String(length=16), nullable=False), + sa.Column("correction", sa.Text(), nullable=True), + sa.Column( + "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now() + ), + sa.Column( + "updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now() + ), + sa.PrimaryKeyConstraint("feedback_id"), + sa.UniqueConstraint("report_id", "dimension_key", name="uq_rp_feedback_report_dimension"), + ) + op.create_index("ix_rp_feedback_user_created", "rp_feedback", ["user_id", "created_at"]) + + op.create_table( + "rp_reservation", + sa.Column("reservation_id", sa.BigInteger(), nullable=False), + sa.Column("user_id", sa.BigInteger(), nullable=False), + sa.Column("topic", sa.String(length=32), nullable=False), + sa.Column( + "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now() + ), + sa.PrimaryKeyConstraint("reservation_id"), + sa.UniqueConstraint("user_id", "topic", name="uq_rp_reservation_user_topic"), + ) + + +def downgrade() -> None: + op.drop_table("rp_reservation") + op.drop_index("ix_rp_feedback_user_created", table_name="rp_feedback") + op.drop_table("rp_feedback") diff --git a/packages/agent/README.md b/packages/agent/README.md index 3e228a8..53e87af 100644 --- a/packages/agent/README.md +++ b/packages/agent/README.md @@ -14,15 +14,16 @@ | --- | --- | | ① 输入审核（TMS + 备案关键词库），fail-close | ❌ CR-06 | | ② 意图识别（含越界：医疗、身材评判） | ❌ FR-AGT-02 | -| ③ 上下文组装（token 预算严格分配） | 部分：只有固定前缀 + 当前输入 | +| ③ 上下文组装（token 预算严格分配） | 部分：画像摘要 + 最近 N 轮历史（`AGENT_HISTORY_TURNS`，默认 10）；检索知识与会话摘要未做 | | ④ 混合检索 | ❌ FR-AGT-04 | | ⑤ LLM 生成（流式 SSE） | 部分：非流式 | | ⑥ 生成后校验（禁忌违反 = 0、色值在板内） | ❌ FR-AGT-05 | | ⑦ 输出审核 | ❌ CR-06 | -| ⑧ 落库 message + retrieval_trace + rule_hits + 成本 | 部分：只有成本 | -| ⑨ 异步偏好抽取 → 写 L2 记忆 | ❌ FR-MEM-03 | +| ⑧ 落库 message + retrieval_trace + rule_hits + 成本 | 部分：ag_message 双行 + 成本已实现；retrieval_trace / rule_hits 列已建、待步骤④填充 | -表 `ag_conversation` / `ag_message` / `ag_feedback` 尚未创建。 +表 `ag_conversation` / `ag_message` 已建（迁移 0007，T-P4-01）；`ag_feedback` 尚未创建。 +画像摘要经 `report.profile.latest_profile_summary` 由 apps 层闭包注入 +（agent 与 report 同层，不允许直接互调）。 ## Token 预算（架构 §7.2，必须硬性限制） diff --git a/packages/agent/context.py b/packages/agent/context.py new file mode 100644 index 0000000..4f3f586 --- /dev/null +++ b/packages/agent/context.py @@ -0,0 +1,57 @@ +"""上下文组装（架构 §7.2 步骤③）的纯逻辑：无 IO、无 LLM 调用，可单测。 + +预算分配（packages/agent/README.md 的预算表）： + +- 画像摘要 ≤ 500 token —— 按中文约 1.6 字符/token 保守取 800 字符上限， + 摘要本身的生产侧（report 域）按 ``PROFILE_SUMMARY_MAX_CHARS`` 裁剪； +- 历史轮数默认 10 轮（``AGENT_HISTORY_TURNS``），一轮 = user + assistant 两条。 + +缓存纪律：画像与历史都是**逐用户逐轮变化**的内容，一律走 gateway 的 +``messages`` 通道排在固定 system 前缀之后，绝不进 ``system_prefix``—— +否则缓存命中单价 1/30 的红利会被画像差异完全打掉（README「两件容易踩的事」）。 +""" + +from __future__ import annotations + +from collections.abc import Sequence + +from packages.llm.types import Message + +#: 画像摘要载荷上限（字符），对应 ≤ 500 token 的预算行。 +PROFILE_SUMMARY_MAX_CHARS = 800 + +#: 画像摘要的消息壳。摘要是 JSON 素材（report 域 compose_prompt_material 的 +#: 产物），需要这句说明告诉模型怎么用它；措辞保持中性，不引入新规则。 +_PROFILE_MESSAGE_TEMPLATE = ( + "以下是该用户最近一份美学诊断报告的画像摘要（JSON），" + "回答时据此给出贴合其条件的建议；用户未问及时不要主动复述全文。\n" +) + +#: 单条历史消息的字符上限：用户输入已被 ChatRequest 卡在 2000 字符， +#: assistant 回复被提示词约束在 300 字以内，这里只做兜底截断防脏数据。 +_HISTORY_MESSAGE_MAX_CHARS = 2000 + + +def build_profile_message(profile_summary: str) -> Message: + return Message(role="system", content=_PROFILE_MESSAGE_TEMPLATE + profile_summary) + + +def build_context_messages( + *, + profile_summary: str | None, + history: Sequence[Message], + history_turns: int, +) -> list[Message]: + """组装 gateway ``messages`` 通道的内容（固定前缀与当前输入由 gateway 自己拼）。 + + - 画像摘要在最前（无画像用户为 None 即不注入，走通用建议降级路径 FR-AGT-10）； + - 历史按时间正序，只保留最近 ``history_turns`` 轮（一轮 = user+assistant 两条）， + 截尾不截头——越早的轮次对当前回答价值越低。 + """ + messages: list[Message] = [] + if profile_summary: + messages.append(build_profile_message(profile_summary)) + tail = list(history)[-(history_turns * 2) :] if history_turns > 0 else [] + for item in tail: + messages.append(Message(role=item.role, content=item.content[:_HISTORY_MESSAGE_MAX_CHARS])) + return messages diff --git a/packages/agent/repo.py b/packages/agent/repo.py new file mode 100644 index 0000000..e3d2b2b --- /dev/null +++ b/packages/agent/repo.py @@ -0,0 +1,108 @@ +"""ag_conversation / ag_message 的数据访问。**只有本模块可以触这两张表**。""" + +from __future__ import annotations + +from datetime import UTC, datetime +from decimal import Decimal + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from packages.shared.models import AgConversation, AgMessage + +ROLE_USER = "user" +ROLE_ASSISTANT = "assistant" + + +async def create_conversation( + session: AsyncSession, *, conversation_id: int, user_id: int +) -> AgConversation: + conversation = AgConversation(conversation_id=conversation_id, user_id=user_id) + session.add(conversation) + await session.flush() + return conversation + + +async def get_conversation(session: AsyncSession, conversation_id: int) -> AgConversation | None: + return await session.get(AgConversation, conversation_id) + + +async def list_conversations( + session: AsyncSession, *, user_id: int, limit: int = 20 +) -> list[AgConversation]: + """会话列表：按最近活跃倒序；从未发言的会话（last_message_at 为 NULL）排最后。""" + result = await session.execute( + select(AgConversation) + .where(AgConversation.user_id == user_id) + .order_by(AgConversation.last_message_at.desc().nulls_last()) + .limit(limit) + ) + return list(result.scalars().all()) + + +async def list_messages( + session: AsyncSession, *, conversation_id: int, limit: int = 100 +) -> list[AgMessage]: + """历史接口用：按时间正序返回。""" + result = await session.execute( + select(AgMessage) + .where(AgMessage.conversation_id == conversation_id) + .order_by(AgMessage.created_at.asc(), AgMessage.message_id.asc()) + .limit(limit) + ) + return list(result.scalars().all()) + + +async def list_recent_messages( + session: AsyncSession, *, conversation_id: int, limit: int +) -> list[AgMessage]: + """上下文组装用：取最近 ``limit`` 条，返回时按时间正序（旧→新）。""" + result = await session.execute( + select(AgMessage) + .where(AgMessage.conversation_id == conversation_id) + .order_by(AgMessage.created_at.desc(), AgMessage.message_id.desc()) + .limit(limit) + ) + return list(reversed(result.scalars().all())) + + +async def append_message( + session: AsyncSession, + *, + message_id: int, + conversation: AgConversation, + role: str, + content: str, + call_id: int | None = None, + model: str | None = None, + tokens_in: int | None = None, + tokens_out: int | None = None, + tokens_cached: int | None = None, + latency_ms: int | None = None, + cost_cents: Decimal | None = None, +) -> AgMessage: + """追加消息并推进会话的 message_count / last_message_at / title。 + + title 只在首条用户消息时落定（截断前 20 字），之后不再改。 + """ + message = AgMessage( + message_id=message_id, + conversation_id=conversation.conversation_id, + user_id=conversation.user_id, + role=role, + content=content, + call_id=call_id, + model=model, + tokens_in=tokens_in, + tokens_out=tokens_out, + tokens_cached=tokens_cached, + latency_ms=latency_ms, + cost_cents=cost_cents, + ) + session.add(message) + conversation.message_count += 1 + conversation.last_message_at = datetime.now(UTC) + if not conversation.title and role == ROLE_USER: + conversation.title = content[:20] + await session.flush() + return message diff --git a/packages/agent/service.py b/packages/agent/service.py index 1c4df42..8f90e0f 100755 --- a/packages/agent/service.py +++ b/packages/agent/service.py @@ -1,20 +1,23 @@ -"""Agent 对话编排 — the scaffold's thin vertical slice. - -It walks the full 架构文档 §7.2 spine but only the load-bearing parts: +"""Agent 对话编排（架构 §7.2）。 ① 输入审核 ← NOT implemented (compliance.moderate, CR-06) ② 意图识别 ← NOT implemented (FR-AGT-02) - ③ 上下文组装 ← only the fixed system prefix + current turn + ③ 上下文组装 ← IMPLEMENTED：画像摘要 + 最近 N 轮历史（context.py） ④ 混合检索 ← NOT implemented (retrieval-svc, FR-AGT-04) ⑤ LLM 生成 ← IMPLEMENTED (non-streaming) ⑥ 生成后校验 ← NOT implemented (FR-AGT-05) ⑦ 输出审核 ← NOT implemented (CR-06) - ⑧ 落库 + 成本 ← cost only; ag_message is not in this scaffold + ⑧ 落库 + 成本 ← IMPLEMENTED：ag_message 双行 + ob_ai_call 成本行 ⑨ 异步偏好抽取 ← NOT implemented (FR-MEM-03) -What *is* fully wired is the plumbing every future feature needs: JWT identity, -the quota three-phase protocol, the LLM gateway, cost attribution, and the -unified envelope. +会话持久化（FR-AGT-09 / T-P4-01）：会话与消息落 ``ag_conversation`` / +``ag_message``；assistant 行的 ``call_id`` 关联 ``ob_ai_call``，每条回答可 +复盘到模型、token 与成本。检索轨迹与规则命中列已建，等步骤④落地后填充。 + +画像注入：用户最近一份 succeeded 报告的画像摘要以独立 system 消息排在固定 +前缀之后（不进缓存前缀，理由见 context.py）。摘要的读取由 apps 层以 +``profile_provider`` 闭包注入——agent 与 report 在 import-linter 分层里是 +同层，不允许直接互调。 Transaction discipline: the reservation is committed **before** the LLM call. If the process dies mid-call, the reservation survives and can be reconciled; @@ -24,24 +27,46 @@ API layer's rollback cannot undo it. from __future__ import annotations +from collections.abc import Awaitable, Callable from dataclasses import dataclass from decimal import Decimal from sqlalchemy.ext.asyncio import AsyncSession +from packages.agent import repo +from packages.agent.context import build_context_messages from packages.billing.service import BillingService, Reservation +from packages.domain.ids import next_id from packages.llm.gateway import LLMGateway -from packages.llm.types import Scene +from packages.llm.types import Message, Scene +from packages.shared.errors import NotFoundError from packages.shared.logging import get_logger +from packages.shared.models import AgConversation, AgMessage logger = get_logger(__name__) QUOTA_KEY_AGENT_CHAT = "agent_chat" REF_TYPE = "agent_chat" +#: 画像提供者：user_id → 画像摘要素材（无报告用户返回 None）。由 apps 层接线。 +ProfileProvider = Callable[[int], Awaitable[str | None]] + + +def _parse_conversation_id(raw: str) -> int: + """wire 侧雪花 ID 是字符串（超 2^53,JS Number 会丢精度）。 + + 非法值与「会话不存在」同口径 40400：不区分「格式错」与「不存在」， + 避免给遍历者任何额外的探测信号。 + """ + try: + return int(raw) + except (TypeError, ValueError) as exc: + raise NotFoundError("会话不存在或已删除") from exc + @dataclass(frozen=True, slots=True) class ChatResult: + conversation_id: int reply: str provider: str model: str @@ -64,12 +89,55 @@ class AgentService: *, gateway: LLMGateway, billing: BillingService, + profile_provider: ProfileProvider | None = None, + history_turns: int = 10, ) -> None: self._session = session self._gateway = gateway self._billing = billing + self._profile_provider = profile_provider + self._history_turns = history_turns + + async def create_conversation(self, *, user_id: int) -> AgConversation: + conversation = await repo.create_conversation( + self._session, conversation_id=next_id(), user_id=user_id + ) + await self._session.commit() + return conversation + + async def list_conversations(self, *, user_id: int) -> list[AgConversation]: + return await repo.list_conversations(self._session, user_id=user_id) + + async def list_messages(self, *, user_id: int, conversation_id: int) -> list[AgMessage]: + await self.get_owned_conversation(user_id=user_id, conversation_id=conversation_id) + return await repo.list_messages(self._session, conversation_id=conversation_id) + + async def get_owned_conversation(self, *, user_id: int, conversation_id: int) -> AgConversation: + conversation = await repo.get_conversation(self._session, conversation_id) + if conversation is None: + raise NotFoundError("会话不存在或已删除") + if conversation.user_id != user_id: + # 不暴露「存在但属于别人」：与 NOT_FOUND 同口径，防会话 ID 遍历 + raise NotFoundError("会话不存在或已删除") + return conversation + + async def chat( + self, + *, + user_id: int, + message: str, + request_key: str, + conversation_id: str | None = None, + ) -> ChatResult: + if conversation_id is None: + conversation = await repo.create_conversation( + self._session, conversation_id=next_id(), user_id=user_id + ) + else: + conversation = await self.get_owned_conversation( + user_id=user_id, conversation_id=_parse_conversation_id(conversation_id) + ) - async def chat(self, *, user_id: int, message: str, request_key: str) -> ChatResult: reservation = await self._billing.reserve( user_id=user_id, quota_key=QUOTA_KEY_AGENT_CHAT, @@ -81,9 +149,13 @@ class AgentService: ) await self._session.commit() + # 预扣已 commit，自此到 commit/reserve 冲销之间的任何异常都必须退还—— + # 包括上下文组装（历史消息/画像查询同样是 DB 调用，也会抛）。 try: + context = await self._build_context(user_id=user_id, conversation=conversation) completion = await self._gateway.complete( scene=Scene.AGENT_CHAT, + messages=context, prompt_vars={"user_input": message}, user_id=user_id, ref_type=REF_TYPE, @@ -94,9 +166,31 @@ class AgentService: raise await self._billing.commit(reservation) + await repo.append_message( + self._session, + message_id=next_id(), + conversation=conversation, + role=repo.ROLE_USER, + content=message, + ) + await repo.append_message( + self._session, + message_id=next_id(), + conversation=conversation, + role=repo.ROLE_ASSISTANT, + content=completion.text, + call_id=completion.call_id, + model=completion.model, + tokens_in=completion.usage.tokens_in, + tokens_out=completion.usage.tokens_out, + tokens_cached=completion.usage.tokens_cached, + latency_ms=completion.latency_ms, + cost_cents=completion.cost_cents, + ) await self._session.commit() return ChatResult( + conversation_id=conversation.conversation_id, reply=completion.text, provider=completion.provider, model=completion.model, @@ -112,6 +206,29 @@ class AgentService: reserve_tx_id=reservation.tx_id, ) + async def _build_context(self, *, user_id: int, conversation: AgConversation) -> list[Message]: + # 画像取不到（无报告 / 读取异常）不阻断对话：降级为通用建议（FR-AGT-10）。 + # NFR-SEC-03：摘要内容本身不进日志，只记有无与长度。 + profile_summary: str | None = None + if self._profile_provider is not None: + try: + profile_summary = await self._profile_provider(user_id) + except Exception: + logger.warning("agent_profile_load_failed", user_id=user_id) + history: list[Message] = [] + if conversation.message_count > 0: + recent = await repo.list_recent_messages( + self._session, + conversation_id=conversation.conversation_id, + limit=self._history_turns * 2, + ) + history = [Message(role=item.role, content=item.content) for item in recent] + return build_context_messages( + profile_summary=profile_summary, + history=history, + history_turns=self._history_turns, + ) + async def _refund(self, reservation: Reservation, *, reason: str) -> None: try: await self._billing.refund(reservation, reason=reason) diff --git a/packages/report/README.md b/packages/report/README.md index d321f70..04a1e3e 100644 --- a/packages/report/README.md +++ b/packages/report/README.md @@ -3,6 +3,9 @@ **职责**（架构 §4.1）：触发判定、生成编排、版本管理、分享。 **不负责**：具体的视觉分析与文案生成（调 AI 能力层）。 +`profile.py` 对外提供「最近一份 succeeded 报告的画像摘要」（Agent 上下文 +注入用，P0-2）；rp_report / rp_input 仍只有本域 repo 可触，外部一律走这里。 + ## 待实现 - 表：`rp_report` `rp_input` `rp_share` @@ -17,3 +20,16 @@ （也就不需要用户重新上传）。 4. **生成后校验不可跳过**：禁忌违反、色值在板内、维度引用存在、无医疗/贬损表述。 最多重试 2 次，仍失败降级为规则直出模板文案，**绝不放行违规内容**。 + +## 结论反馈（T-P3-21 最小可用版，迭代 3 落地） + +`feedback_service.py` + `rp_feedback`：维度级「准/不准 + 可选修正」，同维度 +重复反馈 = 覆盖更新（upsert）。**有意裁剪**：原文要求的「修正后置信度置 1.0 +并即时写 L1 记忆」未做——L1 记忆系统未建，本迭代只落表，认同率/修正率 +两项指标可从 SQL 直接算出，不回写报告结论。 + +## 预约留资（T-P3-22，迭代 3 落地） + +`reservation_service.py` + `rp_reservation`：topic 预留通用性（首期 +full_report）。指标口径：**预约率 = rp_reservation(topic='full_report') 行数 +÷ 埋点 full_report.card_view 去重曝光数**；决策门槛 ≥8%（产品方案 §2.2.3）。 diff --git a/packages/report/feedback_service.py b/packages/report/feedback_service.py new file mode 100644 index 0000000..3443140 --- /dev/null +++ b/packages/report/feedback_service.py @@ -0,0 +1,108 @@ +"""报告结论反馈服务（T-P3-21 最小可用版）。 + +有意裁剪（防误解）：T-P3-21 原文要求「修正后置信度置 1.0 并即时写 L1 记忆」 +——L1 记忆系统未建，本迭代只落 rp_feedback 表，保证认同率/修正率两项指标 +可从 SQL 直接算出，**不回写报告结论**。 + +归属口径：他人报告与不存在同口径 40420（不透出报告是否存在的额外信息）。 +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + +from sqlalchemy.ext.asyncio import AsyncSession + +from packages.domain.ids import next_id +from packages.report import repo +from packages.report.schemas import FeedbackVerdict +from packages.shared.errors import AppError, ErrorCode +from packages.shared.models import RpFeedback, RpReport + +DIMENSION_STATUS_OK = "ok" + + +@dataclass(frozen=True, slots=True) +class FeedbackUpsert: + feedback: RpFeedback + updated: bool # true=同维度已有反馈被覆盖；false=新建 + + +def ok_dimension_keys(report_result: dict[str, Any]) -> set[str]: + """报告产物中可反馈的维度集合：存在于 diagnosis.dimensions 且 status=ok。 + + 纯函数（不碰 I/O），路由层校验与单测共用同一口径。 + """ + diagnosis = report_result.get("diagnosis") + if not isinstance(diagnosis, dict): + return set() + dimensions = diagnosis.get("dimensions") + if not isinstance(dimensions, list): + return set() + keys: set[str] = set() + for item in dimensions: + if ( + isinstance(item, dict) + and item.get("status") == DIMENSION_STATUS_OK + and isinstance(item.get("key"), str) + ): + keys.add(item["key"]) + return keys + + +class FeedbackService: + def __init__(self, session: AsyncSession) -> None: + self._session = session + + async def _get_owned_report(self, *, user_id: int, report_id: int) -> RpReport: + report = await repo.get_report(self._session, report_id) + # 归属校验沿用 get_owned 语义，但对外同口径 40420： + # 不存在与他人报告不可区分，不透出报告是否真实存在。 + if report is None or report.user_id != user_id: + raise AppError(ErrorCode.REPORT_NOT_FOUND) + return report + + async def upsert_feedback( + self, + *, + user_id: int, + report_id: int, + dimension_key: str, + verdict: FeedbackVerdict, + correction: str | None, + ) -> FeedbackUpsert: + """同维度重复反馈 = 覆盖更新（uq_rp_feedback_report_dimension 兜底）。""" + report = await self._get_owned_report(user_id=user_id, report_id=report_id) + if dimension_key not in ok_dimension_keys(report.result or {}): + raise AppError( + ErrorCode.VALIDATION_FAILED, + message="该维度没有可反馈的结论", + hint="只能对报告中已生成的维度结论反馈", + ) + correction = correction.strip() if correction is not None else None + if correction == "": + correction = None + existing = await repo.get_feedback( + self._session, report_id=report_id, dimension_key=dimension_key + ) + if existing is not None: + existing.verdict = verdict + existing.correction = correction + await self._session.flush() + return FeedbackUpsert(feedback=existing, updated=True) + row = RpFeedback( + feedback_id=next_id(), + report_id=report_id, + user_id=user_id, + dimension_key=dimension_key, + verdict=verdict, + correction=correction, + ) + self._session.add(row) + await self._session.flush() + return FeedbackUpsert(feedback=row, updated=False) + + async def list_feedback(self, *, user_id: int, report_id: int) -> list[RpFeedback]: + await self._get_owned_report(user_id=user_id, report_id=report_id) + return await repo.list_feedback(self._session, report_id=report_id) diff --git a/packages/report/profile.py b/packages/report/profile.py new file mode 100644 index 0000000..a9da3a1 --- /dev/null +++ b/packages/report/profile.py @@ -0,0 +1,50 @@ +"""对外（同层以上）提供「最近一份报告的画像摘要」，供 Agent 上下文注入。 + +只取结论摘要而非报告全文：素材走 ``copy.compose_prompt_material``——与 +REPORT_COPY 场景同一套裁剪纪律（skipped 维度不进素材防幻觉、超长按优先级 +裁、硬结论永不裁），只是预算从报告的 6000 字符收紧到调用方给的额度。 + +归属纪律不变：rp_report / rp_input 只有 report 域内部可触（repo.py 头部 +约定），外部要画像只能走这里。 +""" + +from __future__ import annotations + +from typing import Any + +from sqlalchemy.ext.asyncio import AsyncSession + +from packages.report import repo +from packages.report.copy import compose_prompt_material +from packages.shared.logging import get_logger + +logger = get_logger(__name__) + +#: 往前翻几份报告找可用的诊断产物。正常用户只有几份；留余量是为了容错 +#: 「最新一份 failed/processing 而上一份 succeeded」的场景。 +_SCAN_LIMIT = 5 + + +async def latest_profile_summary( + session: AsyncSession, *, user_id: int, max_chars: int +) -> str | None: + """该用户最近一份带诊断结论的 succeeded 报告 → 紧凑 JSON 素材；无则 None。 + + NFR-SEC-03：摘要内容不进日志，这里只记「命中第几份报告」。 + """ + reports = await repo.list_reports(session, user_id=user_id, limit=_SCAN_LIMIT) + for index, report in enumerate(reports): + if report.status != repo.STATUS_SUCCEEDED: + continue + diagnosis: Any = (report.result or {}).get("diagnosis") + if not isinstance(diagnosis, dict) or not diagnosis.get("dimensions"): + continue + material, dropped = compose_prompt_material(diagnosis, max_chars=max_chars) + logger.info( + "report_profile_summary_built", + report_id=report.report_id, + scan_depth=index, + dropped_keys=len(dropped), + ) + return material + return None diff --git a/packages/report/repo.py b/packages/report/repo.py index 4dcecef..cda8612 100644 --- a/packages/report/repo.py +++ b/packages/report/repo.py @@ -1,11 +1,14 @@ -"""rp_report / rp_input 的数据访问。**只有本模块可以触这两张表**（rp_share 归后续任务）。""" +"""rp_report / rp_input / rp_feedback / rp_reservation 的数据访问。 + +**只有本模块可以触这四张表**（rp_share 归后续任务）。 +""" from __future__ import annotations from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession -from packages.shared.models import RpInput, RpReport +from packages.shared.models import RpFeedback, RpInput, RpReport, RpReservation STATUS_SUCCEEDED = "succeeded" TIER_BASIC = "basic" @@ -45,3 +48,52 @@ async def list_reports(session: AsyncSession, *, user_id: int, limit: int = 20) async def get_input(session: AsyncSession, report_id: int) -> RpInput | None: result = await session.execute(select(RpInput).where(RpInput.report_id == report_id)) return result.scalar_one_or_none() + + +# -- rp_feedback（T-P3-21：同维度覆盖更新，upsert 语义） ----------------------- + + +async def get_feedback( + session: AsyncSession, *, report_id: int, dimension_key: str +) -> RpFeedback | None: + result = await session.execute( + select(RpFeedback).where( + RpFeedback.report_id == report_id, + RpFeedback.dimension_key == dimension_key, + ) + ) + return result.scalar_one_or_none() + + +async def list_feedback(session: AsyncSession, *, report_id: int) -> list[RpFeedback]: + result = await session.execute( + select(RpFeedback) + .where(RpFeedback.report_id == report_id) + .order_by(RpFeedback.created_at.asc(), RpFeedback.feedback_id.asc()) + ) + return list(result.scalars().all()) + + +# -- rp_reservation（T-P3-22：重复预约幂等） ------------------------------------ + + +async def get_reservation( + session: AsyncSession, *, user_id: int, topic: str +) -> RpReservation | None: + result = await session.execute( + select(RpReservation).where( + RpReservation.user_id == user_id, + RpReservation.topic == topic, + ) + ) + return result.scalar_one_or_none() + + +async def list_reservation_topics( + session: AsyncSession, *, user_id: int, topic: str | None = None +) -> list[str]: + stmt = select(RpReservation.topic).where(RpReservation.user_id == user_id) + if topic is not None: + stmt = stmt.where(RpReservation.topic == topic) + result = await session.execute(stmt.order_by(RpReservation.topic.asc())) + return [str(row) for row in result.scalars().all()] diff --git a/packages/report/reservation_service.py b/packages/report/reservation_service.py new file mode 100644 index 0000000..3f4977d --- /dev/null +++ b/packages/report/reservation_service.py @@ -0,0 +1,43 @@ +"""预约留资服务（T-P3-22）。 + +topic 预留通用性，首期只有 full_report（全面报告上线提醒）。 +幂等口径：UniqueConstraint(user_id, topic)——重复预约返回已有记录 +（created=false，HTTP 仍 200），绝不堆行。 + +指标口径：预约率 = rp_reservation(topic='full_report') 行数 ÷ 埋点 +full_report.card_view 去重曝光数；决策门槛 ≥8%（产品方案 §2.2.3）。 +""" + +from __future__ import annotations + +from dataclasses import dataclass + +from sqlalchemy.ext.asyncio import AsyncSession + +from packages.domain.ids import next_id +from packages.report import repo +from packages.shared.models import RpReservation + + +@dataclass(frozen=True, slots=True) +class ReservationResult: + reservation: RpReservation + created: bool # false=幂等命中已有记录 + + +class ReservationService: + def __init__(self, session: AsyncSession) -> None: + self._session = session + + async def reserve(self, *, user_id: int, topic: str) -> ReservationResult: + """幂等预约：同 user+topic 已存在即返回原记录。""" + existing = await repo.get_reservation(self._session, user_id=user_id, topic=topic) + if existing is not None: + return ReservationResult(reservation=existing, created=False) + row = RpReservation(reservation_id=next_id(), user_id=user_id, topic=topic) + self._session.add(row) + await self._session.flush() + return ReservationResult(reservation=row, created=True) + + async def list_topics(self, *, user_id: int, topic: str | None = None) -> list[str]: + return await repo.list_reservation_topics(self._session, user_id=user_id, topic=topic) diff --git a/packages/report/schemas.py b/packages/report/schemas.py index a18c4b5..f64305d 100644 --- a/packages/report/schemas.py +++ b/packages/report/schemas.py @@ -7,6 +7,7 @@ Wire 响应统一走 ok() 信封（架构文档 §9.2），这里的模型只描 from __future__ import annotations from datetime import datetime +from typing import Literal from pydantic import BaseModel, Field @@ -57,3 +58,49 @@ class ReportItem(BaseModel): ) created_at: datetime completed_at: datetime | None + + +# --------------------------------------------------------------------------- +# T-P3-21：维度级结论反馈 +# --------------------------------------------------------------------------- +FeedbackVerdict = Literal["accurate", "inaccurate"] + + +class FeedbackCreateRequest(BaseModel): + dimension_key: str = Field( + min_length=1, max_length=64, description="报告 result.diagnosis.dimensions 的 key" + ) + verdict: FeedbackVerdict = Field(description="准（accurate）/ 不准（inaccurate）") + correction: str | None = Field( + default=None, max_length=200, description="verdict=inaccurate 时的可选修正，≤200 字" + ) + + +class FeedbackUpsertResponse(BaseModel): + dimension_key: str + verdict: str + updated: bool = Field(description="true=同维度已有反馈被覆盖更新；false=新建") + + +class FeedbackItem(BaseModel): + dimension_key: str + verdict: str + correction: str | None + created_at: datetime + + +# --------------------------------------------------------------------------- +# T-P3-22：预约留资（topic 预留通用性，首期 full_report） +# --------------------------------------------------------------------------- +class ReservationCreateRequest(BaseModel): + topic: str = Field(min_length=1, max_length=32) + + +class ReservationResponse(BaseModel): + topic: str + created: bool = Field(description="false=已存在（幂等命中，HTTP 仍 200）") + created_at: datetime + + +class ReservationListResponse(BaseModel): + topics: list[str] diff --git a/packages/shared/config.py b/packages/shared/config.py index df11296..56219fe 100644 --- a/packages/shared/config.py +++ b/packages/shared/config.py @@ -126,6 +126,10 @@ class Settings(BaseSettings): # --- Auth: refresh sessions (FR-ACC-05) -------------------------------- refresh_token_ttl_seconds: int = 2_592_000 # 30 天 + # --- Agent 对话（FR-AGT-09）-------------------------------------------- + # 对话上下文携带的最近轮数（一轮 = user + assistant 两条），默认 10。 + agent_history_turns: int = 10 + # --- 任务奖励（FR-BILL-03/04，配置化）----------------------------------- # amount 全部是占位值 1：正式数值待产品在产品方案 §3.2 定稿后改 env，不改代码。 # claim_period 决定幂等粒度（同一任务同周期只能领一次）。 diff --git a/packages/shared/models.py b/packages/shared/models.py index ac40194..a4254be 100644 --- a/packages/shared/models.py +++ b/packages/shared/models.py @@ -1,9 +1,9 @@ """SQLAlchemy 2.0 ORM models. Scope of this scaffold: only the tables needed to prove the vertical slice -(架构文档 §6.2 identity / billing / task / 观测与成本). All remaining domains -(pf_ / rp_ / kb_ / ag_ / tr_ / cp_) are intentionally absent — see -packages/*/README.md for what still has to be built. +(架构文档 §6.2 identity / billing / task / 观测与成本) plus the Sprint batches +that have landed since (kb_ / rp_ / ag_). Remaining domains (pf_ / tr_ / cp_) +are intentionally absent — see packages/*/README.md for what still has to be built. Conventions enforced here (架构文档 §6.1): 1. 表名带域前缀 (`id_` / `bl_` / `tk_` / `ob_`)，为将来拆库预留边界。 @@ -541,3 +541,106 @@ class RpShare(Base): ) __table_args__ = (UniqueConstraint("share_token", name="uq_rp_share_token"),) + + +class RpFeedback(Base): + """维度级结论反馈（T-P3-21 最小可用版）：准/不准 + 可选修正文本。 + + 同维度重复反馈是覆盖更新（upsert），由 uq_rp_feedback_report_dimension + 兜底；认同率/修正率两项指标从本表 SQL 直接算出。 + 有意裁剪：不回写报告结论、不写 L1 记忆（记忆系统未建，见迁移 0008）。""" + + __tablename__ = "rp_feedback" + + feedback_id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=False) + report_id: Mapped[int] = mapped_column(BigInteger, nullable=False) + user_id: Mapped[int] = mapped_column(BigInteger, nullable=False) + dimension_key: Mapped[str] = mapped_column(String(64), nullable=False) + verdict: Mapped[str] = mapped_column(String(16), nullable=False) # accurate | inaccurate + correction: Mapped[str | None] = mapped_column(Text) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now() + ) + + __table_args__ = ( + UniqueConstraint("report_id", "dimension_key", name="uq_rp_feedback_report_dimension"), + Index("ix_rp_feedback_user_created", "user_id", "created_at"), + ) + + +class RpReservation(Base): + """预约留资（T-P3-22）：topic 预留通用性，首期只有 full_report。 + + uq_rp_reservation_user_topic 承载幂等：重复预约返回已有记录（created=false）。 + 指标口径：预约率 = topic='full_report' 行数 ÷ 埋点 full_report.card_view + 去重曝光数；决策门槛 ≥8%（产品方案 §2.2.3）。""" + + __tablename__ = "rp_reservation" + + reservation_id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=False) + user_id: Mapped[int] = mapped_column(BigInteger, nullable=False) + topic: Mapped[str] = mapped_column(String(32), nullable=False) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) + + __table_args__ = (UniqueConstraint("user_id", "topic", name="uq_rp_reservation_user_topic"),) + + +# --------------------------------------------------------------------------- +# agent 域（FR-AGT-09 / T-P4-01：会话与消息，含检索轨迹与规则命中） +# --------------------------------------------------------------------------- +class AgConversation(Base): + """对话会话。title 取首条用户消息的截断，列表页展示用。 + + 保留策略（合规方案 §6：对话原文 90 天）由后续清理任务按 + ``last_message_at`` 执行，表结构先把时间锚点留好。""" + + __tablename__ = "ag_conversation" + + conversation_id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=False) + user_id: Mapped[int] = mapped_column(BigInteger, nullable=False) + title: Mapped[str] = mapped_column(String(64), nullable=False, default="") + message_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + last_message_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) + + __table_args__ = (Index("ix_ag_conversation_user_active", "user_id", "last_message_at"),) + + +class AgMessage(Base): + """单条对话消息。``retrieval_trace`` / ``rule_hits`` 是 FR-AGT-09「每条回答 + 可复盘」的复盘锚点：检索与规则命中在对应步骤（架构 §7.2 ④）落地前留空。 + + ``call_id`` 关联 ``ob_ai_call``，一条 assistant 消息的模型/token/成本可 + 一路追到账本；user 消息没有 LLM 调用，这些字段恒 NULL。""" + + __tablename__ = "ag_message" + + message_id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=False) + conversation_id: Mapped[int] = mapped_column(BigInteger, nullable=False) + user_id: Mapped[int] = mapped_column(BigInteger, nullable=False) + role: Mapped[str] = mapped_column(String(16), nullable=False) # user / assistant + content: Mapped[str] = mapped_column(Text, nullable=False) + call_id: Mapped[int | None] = mapped_column(BigInteger) + model: Mapped[str | None] = mapped_column(String(64)) + tokens_in: Mapped[int | None] = mapped_column(Integer) + tokens_out: Mapped[int | None] = mapped_column(Integer) + tokens_cached: Mapped[int | None] = mapped_column(Integer) + latency_ms: Mapped[int | None] = mapped_column(Integer) + cost_cents: Mapped[Decimal | None] = mapped_column(Numeric(14, 6)) + retrieval_trace: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False, default=dict) + rule_hits: Mapped[list[Any]] = mapped_column(JSONB, nullable=False, default=list) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) + + __table_args__ = ( + Index("ix_ag_message_conversation_created", "conversation_id", "created_at"), + Index("ix_ag_message_user_created", "user_id", "created_at"), + ) diff --git a/tests/integration/test_report_feedback_reservations.py b/tests/integration/test_report_feedback_reservations.py new file mode 100644 index 0000000..4b4fd6f --- /dev/null +++ b/tests/integration/test_report_feedback_reservations.py @@ -0,0 +1,211 @@ +"""结论反馈 + 预约留资的端到端断言（T-P3-21 / T-P3-22，迭代 3）。 + +覆盖验收点：反馈 upsert 幂等（同维度覆盖更新不堆行）、归属校验他人报告 +40420、dimension_key 校验（不存在/skipped → 40001）、预约重复 created=false。 +报告行直接 INSERT（不走流水线）——反馈/预约断言的是路由与服务口径， +不是生成链路（生成链路由 test_report_flow 覆盖）。 +""" + +from __future__ import annotations + +import os +import uuid +from collections.abc import AsyncIterator +from typing import Any + +import pytest + + +def _infra_available() -> bool: + return bool(os.getenv("MEIDA_RUN_INTEGRATION")) + + +pytestmark = [ + pytest.mark.integration, + pytest.mark.skipif( + not _infra_available(), + reason="set MEIDA_RUN_INTEGRATION=1 with PostgreSQL and Redis running", + ), +] + +_OK_DIM = "body_shape" +_SKIPPED_DIM = "color_season" + + +@pytest.fixture +async def client() -> AsyncIterator[Any]: + import httpx + + from apps.api.main import create_app + from packages.shared.config import get_settings + + app = create_app(get_settings()) + transport = httpx.ASGITransport(app=app) + async with ( + httpx.AsyncClient(transport=transport, base_url="http://test") as http, + app.router.lifespan_context(app), + ): + yield http + + +async def _login(client: Any) -> tuple[dict[str, str], int]: + resp = await client.post( + "/api/v1/auth/dev-token", json={"device_key": f"fb-it-{uuid.uuid4().hex[:8]}"} + ) + assert resp.json()["code"] == 0 + data = resp.json()["data"] + return {"Authorization": f"Bearer {data['access_token']}"}, int(data["user_id"]) + + +async def _insert_report(user_id: int) -> int: + """直接落一份 succeeded 报告：两个维度，一个 ok 一个 skipped。""" + from packages.domain.ids import next_id + from packages.shared.db import session_scope + from packages.shared.models import RpReport + + report_id = next_id() + async with session_scope() as session: + session.add( + RpReport( + report_id=report_id, + user_id=user_id, + tier="basic", + status="succeeded", + version=1, + result={ + "diagnosis": { + "dimensions": [ + {"number": 5, "key": _OK_DIM, "status": "ok", "value": "X"}, + {"number": 19, "key": _SKIPPED_DIM, "status": "skipped", "value": None}, + ], + "coverage": {"basic_full": False, "total": 2}, + } + }, + ) + ) + return report_id + + +async def _count_feedback(report_id: int) -> int: + from sqlalchemy import func, select + + from packages.shared.db import session_scope + from packages.shared.models import RpFeedback + + async with session_scope() as session: + result = await session.execute( + select(func.count()).select_from(RpFeedback).where(RpFeedback.report_id == report_id) + ) + return int(result.scalar_one()) + + +# --- 反馈：upsert 幂等 -------------------------------------------------------- + + +async def test_feedback_upsert_overwrites_same_dimension(client: Any) -> None: + headers, user_id = await _login(client) + report_id = await _insert_report(user_id) + + first = await client.post( + f"/api/v1/reports/{report_id}/feedback", + headers=headers, + json={"dimension_key": _OK_DIM, "verdict": "accurate"}, + ) + assert first.status_code == 200 + assert first.json()["data"]["updated"] is False + + # 改判：同维度再反馈 = 覆盖更新（updated=true），不是堆行 + second = await client.post( + f"/api/v1/reports/{report_id}/feedback", + headers=headers, + json={"dimension_key": _OK_DIM, "verdict": "inaccurate", "correction": "其实我是梨形"}, + ) + assert second.status_code == 200 + assert second.json()["data"]["updated"] is True + assert await _count_feedback(report_id) == 1 + + listing = (await client.get(f"/api/v1/reports/{report_id}/feedback", headers=headers)).json()[ + "data" + ] + assert len(listing) == 1 + assert listing[0]["dimension_key"] == _OK_DIM + assert listing[0]["verdict"] == "inaccurate" + assert listing[0]["correction"] == "其实我是梨形" + + +# --- 反馈：归属与维度校验 ------------------------------------------------------ + + +async def test_feedback_on_others_report_returns_40420(client: Any) -> None: + headers, _ = await _login(client) + _, other_user_id = await _login(client) + report_id = await _insert_report(other_user_id) + + resp = await client.post( + f"/api/v1/reports/{report_id}/feedback", + headers=headers, + json={"dimension_key": _OK_DIM, "verdict": "accurate"}, + ) + assert resp.json()["code"] == 40420 + + get_resp = await client.get(f"/api/v1/reports/{report_id}/feedback", headers=headers) + assert get_resp.json()["code"] == 40420 + + +async def test_feedback_dimension_must_exist_and_be_ok(client: Any) -> None: + headers, user_id = await _login(client) + report_id = await _insert_report(user_id) + + # 不存在的维度 + resp = await client.post( + f"/api/v1/reports/{report_id}/feedback", + headers=headers, + json={"dimension_key": "no_such_dim", "verdict": "accurate"}, + ) + assert resp.json()["code"] == 40001 + + # skipped 维度没有结论，不可反馈 + resp2 = await client.post( + f"/api/v1/reports/{report_id}/feedback", + headers=headers, + json={"dimension_key": _SKIPPED_DIM, "verdict": "accurate"}, + ) + assert resp2.json()["code"] == 40001 + + # verdict 枚举外 → 40001（pydantic 422 → 统一信封 VALIDATION_FAILED） + resp3 = await client.post( + f"/api/v1/reports/{report_id}/feedback", + headers=headers, + json={"dimension_key": _OK_DIM, "verdict": "kind-of"}, + ) + assert resp3.json()["code"] == 40001 + + assert await _count_feedback(report_id) == 0 + + +# --- 预约：幂等 --------------------------------------------------------------- + + +async def test_reservation_duplicate_returns_created_false(client: Any) -> None: + headers, _ = await _login(client) + + first = await client.post( + "/api/v1/reservations", headers=headers, json={"topic": "full_report"} + ) + assert first.status_code == 200 + assert first.json()["data"]["created"] is True + + again = await client.post( + "/api/v1/reservations", headers=headers, json={"topic": "full_report"} + ) + assert again.status_code == 200 + assert again.json()["data"]["created"] is False + + listing = (await client.get("/api/v1/reservations?topic=full_report", headers=headers)).json()[ + "data" + ] + assert listing["topics"] == ["full_report"] + + # topic 参数缺省返回全部 + all_topics = (await client.get("/api/v1/reservations", headers=headers)).json()["data"] + assert "full_report" in all_topics["topics"] diff --git a/tests/unit/test_agent_context.py b/tests/unit/test_agent_context.py new file mode 100644 index 0000000..c774969 --- /dev/null +++ b/tests/unit/test_agent_context.py @@ -0,0 +1,67 @@ +"""上下文组装（架构 §7.2 步骤③）的纯逻辑测试：画像注入与历史截尾。""" + +from __future__ import annotations + +from packages.agent.context import ( + PROFILE_SUMMARY_MAX_CHARS, + build_context_messages, + build_profile_message, +) +from packages.llm.types import Message + + +class TestBuildContextMessages: + def test_profile_first_then_history(self) -> None: + history = [ + Message(role="user", content="第一句"), + Message(role="assistant", content="第一答"), + ] + messages = build_context_messages( + profile_summary='{"dimensions":[]}', history=history, history_turns=10 + ) + assert len(messages) == 3 + assert messages[0].role == "system" + assert '{"dimensions":[]}' in messages[0].content + assert [m.role for m in messages[1:]] == ["user", "assistant"] + assert [m.content for m in messages[1:]] == ["第一句", "第一答"] + + def test_no_profile_degrades_to_history_only(self) -> None: + """无画像用户（FR-AGT-10 降级）：不注入画像消息，历史照常携带。""" + history = [Message(role="user", content="hi")] + messages = build_context_messages(profile_summary=None, history=history, history_turns=10) + assert len(messages) == 1 + assert messages[0].role == "user" + + def test_empty_profile_string_treated_as_absent(self) -> None: + messages = build_context_messages(profile_summary="", history=[], history_turns=10) + assert messages == [] + + def test_history_tail_kept_by_turns(self) -> None: + """一轮 = user+assistant 两条；N=2 时只留最近 4 条，截头不截尾。""" + history: list[Message] = [] + for i in range(5): + history.append(Message(role="user", content=f"u{i}")) + history.append(Message(role="assistant", content=f"a{i}")) + messages = build_context_messages(profile_summary=None, history=history, history_turns=2) + assert [m.content for m in messages] == ["u3", "a3", "u4", "a4"] + + def test_zero_turns_drops_history(self) -> None: + history = [Message(role="user", content="u0")] + messages = build_context_messages(profile_summary=None, history=history, history_turns=0) + assert messages == [] + + def test_history_message_content_capped(self) -> None: + history = [Message(role="user", content="x" * 5000)] + messages = build_context_messages(profile_summary=None, history=history, history_turns=10) + assert len(messages[0].content) == 2000 + + +class TestBuildProfileMessage: + def test_wraps_summary_as_system_message(self) -> None: + message = build_profile_message("{}") + assert message.role == "system" + assert message.content.endswith("{}") + + def test_budget_constant_within_token_budget(self) -> None: + """画像摘要预算 ≤ 500 token（README 预算表），按 ~1.6 字符/token 折算。""" + assert PROFILE_SUMMARY_MAX_CHARS <= 500 * 1.6 diff --git a/tests/unit/test_agent_service.py b/tests/unit/test_agent_service.py new file mode 100644 index 0000000..624ef29 --- /dev/null +++ b/tests/unit/test_agent_service.py @@ -0,0 +1,331 @@ +"""AgentService 编排测试：repo 层打桩 + 假网关/假计费，不连库不发网。""" + +from __future__ import annotations + +from decimal import Decimal +from typing import Any + +import pytest + +from packages.agent import repo +from packages.agent.service import AgentService +from packages.billing.service import Reservation +from packages.llm.types import Completion, Scene, Usage +from packages.shared.errors import NotFoundError +from packages.shared.models import AgConversation + +USER_ID = 7001 +CONVERSATION_ID = 9001 + + +class FakeSession: + def __init__(self) -> None: + self.commits = 0 + self.rollbacks = 0 + + async def commit(self) -> None: + self.commits += 1 + + async def rollback(self) -> None: + self.rollbacks += 1 + + +class FakeGateway: + def __init__(self) -> None: + self.calls: list[dict[str, Any]] = [] + + async def complete(self, **kwargs: Any) -> Completion: + self.calls.append(kwargs) + return Completion( + text="可以试试白衬衫加深蓝直筒裤。", + scene=Scene.AGENT_CHAT, + provider="deepseek", + model="deepseek-v4-flash", + prompt_id="agent_chat", + prompt_version="v1", + usage=Usage(tokens_in=100, tokens_out=50, tokens_cached=0, tokens_reasoning=0), + latency_ms=800, + cost_cents=Decimal("0.05"), + call_id=555, + ) + + +class FailingGateway(FakeGateway): + async def complete(self, **kwargs: Any) -> Completion: + raise RuntimeError("provider exploded") + + +class FakeBilling: + def __init__(self) -> None: + self.refunds: list[str] = [] + + async def reserve(self, **kwargs: Any) -> Reservation: + return Reservation( + tx_id=1, + user_id=kwargs["user_id"], + quota_key=kwargs["quota_key"], + period_key="day:2026-09-11", + amount=1, + remaining=9, + idempotency_key="k", + ) + + async def commit(self, reservation: Reservation) -> None: + pass + + async def refund(self, reservation: Reservation, *, reason: str) -> None: + self.refunds.append(reason) + + +@pytest.fixture +def store(monkeypatch: pytest.MonkeyPatch) -> dict[str, Any]: + """repo 层全打桩：会话与消息落进内存 dict，断言编排逻辑而非 SQL。""" + state: dict[str, Any] = { + "conversations": { + CONVERSATION_ID: AgConversation( + conversation_id=str(CONVERSATION_ID), + user_id=USER_ID, + title="", + message_count=0, + last_message_at=None, + ) + }, + "messages": [], + "history": [], + "next_id": 100, + } + + def _next_id() -> int: + state["next_id"] += 1 + return state["next_id"] + + async def fake_create(session: Any, *, conversation_id: int, user_id: int) -> AgConversation: + conv = AgConversation( + conversation_id=conversation_id, + user_id=user_id, + title="", + message_count=0, + last_message_at=None, + ) + state["conversations"][conversation_id] = conv + return conv + + async def fake_get(session: Any, conversation_id: int) -> AgConversation | None: + return state["conversations"].get(conversation_id) + + async def fake_recent(session: Any, *, conversation_id: int, limit: int) -> list[Any]: + return state["history"] + + async def fake_append( + session: Any, + *, + message_id: int, + conversation: AgConversation, + role: str, + content: str, + **kwargs: Any, + ) -> Any: + state["messages"].append({"role": role, "content": content, **kwargs}) + conversation.message_count += 1 + if not conversation.title and role == "user": + conversation.title = content[:20] + return None + + monkeypatch.setattr(repo, "create_conversation", fake_create) + monkeypatch.setattr(repo, "get_conversation", fake_get) + monkeypatch.setattr(repo, "list_recent_messages", fake_recent) + monkeypatch.setattr(repo, "append_message", fake_append) + monkeypatch.setattr("packages.agent.service.next_id", _next_id) + return state + + +def _service(store: dict[str, Any], **overrides: Any) -> AgentService: + kwargs: dict[str, Any] = { + "gateway": FakeGateway(), + "billing": FakeBilling(), + "profile_provider": None, + "history_turns": 10, + } + kwargs.update(overrides) + return AgentService(FakeSession(), **kwargs) # type: ignore[arg-type] + + +class TestChat: + async def test_implicit_conversation_created_and_messages_persisted( + self, store: dict[str, Any] + ) -> None: + service = _service(store) + result = await service.chat(user_id=USER_ID, message="面试怎么穿", request_key="k1") + # 隐式建会话，响应带回会话 ID + assert result.conversation_id in store["conversations"] + assert result.conversation_id != CONVERSATION_ID + # 用户消息 + assistant 消息双行落库，assistant 行带复盘字段 + assert [m["role"] for m in store["messages"]] == ["user", "assistant"] + assistant = store["messages"][1] + assert assistant["call_id"] == 555 + assert assistant["model"] == "deepseek-v4-flash" + assert assistant["cost_cents"] == Decimal("0.05") + # 会话标题取首条用户消息 + conv = store["conversations"][result.conversation_id] + assert conv.title == "面试怎么穿" + assert result.quota_remaining == 9 + + async def test_existing_conversation_carries_history(self, store: dict[str, Any]) -> None: + from packages.llm.types import Message + + conv = store["conversations"][CONVERSATION_ID] + conv.message_count = 2 + store["history"] = [ + _msg("user", "之前的问题"), + _msg("assistant", "之前的回答"), + ] + service = _service(store) + await service.chat( + user_id=USER_ID, + message="那约会呢", + request_key="k2", + conversation_id=str(CONVERSATION_ID), + ) + sent = service._gateway.calls[0]["messages"] # type: ignore[attr-defined] + assert [m.content for m in sent] == ["之前的问题", "之前的回答"] + assert isinstance(sent[0], Message) + + async def test_profile_injected_as_leading_system_message(self, store: dict[str, Any]) -> None: + async def profile_provider(user_id: int) -> str | None: + assert user_id == USER_ID + return '{"dimensions":[{"key":"body_shape","value":"梨形"}]}' + + service = _service(store, profile_provider=profile_provider) + await service.chat( + user_id=USER_ID, + message="推荐裤子", + request_key="k3", + conversation_id=str(CONVERSATION_ID), + ) + sent = service._gateway.calls[0]["messages"] # type: ignore[attr-defined] + assert sent[0].role == "system" + assert "梨形" in sent[0].content + + async def test_profile_failure_degrades_silently(self, store: dict[str, Any]) -> None: + async def broken_provider(user_id: int) -> str | None: + raise RuntimeError("report table down") + + service = _service(store, profile_provider=broken_provider) + result = await service.chat( + user_id=USER_ID, + message="hi", + request_key="k4", + conversation_id=str(CONVERSATION_ID), + ) + assert result.reply # 画像失败不阻断对话 + sent = service._gateway.calls[0]["messages"] # type: ignore[attr-defined] + assert all(m.role != "system" for m in sent) + + async def test_other_users_conversation_not_found(self, store: dict[str, Any]) -> None: + """他人会话与不存在同口径 404，不暴露存在性。""" + service = _service(store) + with pytest.raises(NotFoundError): + await service.chat( + user_id=9999, + message="hi", + request_key="k5", + conversation_id=str(CONVERSATION_ID), + ) + + async def test_llm_failure_refunds_and_persists_nothing(self, store: dict[str, Any]) -> None: + billing = FakeBilling() + service = _service(store, gateway=FailingGateway(), billing=billing) + with pytest.raises(RuntimeError): + await service.chat( + user_id=USER_ID, + message="hi", + request_key="k6", + conversation_id=str(CONVERSATION_ID), + ) + assert billing.refunds == ["RuntimeError"] + assert store["messages"] == [] + + async def test_large_snowflake_id_as_string(self, store: dict[str, Any]) -> None: + """超 2^53 的雪花 ID 以字符串传入，无精度丢失地落到归属校验。""" + big = 223843017600012288 # > 2**53,JS Number 无法精确表示 + store["conversations"][big] = AgConversation( + conversation_id=big, + user_id=USER_ID, + title="", + message_count=0, + last_message_at=None, + ) + service = _service(store) + result = await service.chat( + user_id=USER_ID, + message="hi", + request_key="k7", + conversation_id=str(big), + ) + assert result.conversation_id == big + + async def test_invalid_conversation_id_same_404(self, store: dict[str, Any]) -> None: + """非法字符串与「不存在」同口径 40400，不给遍历者额外信号。""" + service = _service(store) + with pytest.raises(NotFoundError): + await service.chat( + user_id=USER_ID, + message="hi", + request_key="k8", + conversation_id="not-a-number", + ) + + async def test_context_build_failure_refunds( + self, store: dict[str, Any], monkeypatch: pytest.MonkeyPatch + ) -> None: + """预扣 commit 之后，上下文组装（历史查询）抛异常也必须退还额度。""" + + async def broken_recent(session: Any, *, conversation_id: int, limit: int) -> list[Any]: + raise RuntimeError("db hiccup") + + monkeypatch.setattr(repo, "list_recent_messages", broken_recent) + conv = store["conversations"][CONVERSATION_ID] + conv.message_count = 2 # 触发历史查询路径 + billing = FakeBilling() + service = _service(store, billing=billing) + with pytest.raises(RuntimeError): + await service.chat( + user_id=USER_ID, + message="hi", + request_key="k9", + conversation_id=str(CONVERSATION_ID), + ) + assert billing.refunds == ["RuntimeError"] + assert store["messages"] == [] + + +class TestConversationApis: + async def test_create_and_get_owned(self, store: dict[str, Any]) -> None: + service = _service(store) + conversation = await service.create_conversation(user_id=USER_ID) + assert conversation.user_id == USER_ID + owned = await service.get_owned_conversation( + user_id=USER_ID, conversation_id=conversation.conversation_id + ) + assert owned.conversation_id == conversation.conversation_id + with pytest.raises(NotFoundError): + await service.get_owned_conversation(user_id=USER_ID, conversation_id=404) + + async def test_list_conversations_scoped_to_user( + self, store: dict[str, Any], monkeypatch: pytest.MonkeyPatch + ) -> None: + async def fake_list(session: Any, *, user_id: int, limit: int = 20) -> list[Any]: + return [c for c in store["conversations"].values() if c.user_id == user_id] + + monkeypatch.setattr(repo, "list_conversations", fake_list) + service = _service(store) + mine = await service.list_conversations(user_id=USER_ID) + other = await service.list_conversations(user_id=9999) + assert len(mine) == 1 + assert other == [] + + +def _msg(role: str, content: str) -> Any: + from packages.llm.types import Message + + return Message(role=role, content=content) diff --git a/tests/unit/test_report_feedback.py b/tests/unit/test_report_feedback.py new file mode 100644 index 0000000..8d4a659 --- /dev/null +++ b/tests/unit/test_report_feedback.py @@ -0,0 +1,69 @@ +"""报告结论反馈的纯函数与 schema 校验（T-P3-21 最小可用版，不连库）。 + +覆盖：ok_dimension_keys 的可反馈维度集合口径（status=ok 才可选）、 +FeedbackCreateRequest 的 verdict 枚举与 correction 长度上限。 +""" + +from __future__ import annotations + +import pytest +from pydantic import ValidationError + +from packages.report.feedback_service import ok_dimension_keys +from packages.report.schemas import FeedbackCreateRequest + + +def _result(dimensions: list[dict[str, object]]) -> dict[str, object]: + return {"diagnosis": {"dimensions": dimensions, "coverage": {"total": len(dimensions)}}} + + +class TestOkDimensionKeys: + def test_only_status_ok_dimensions_are_feedbackable(self) -> None: + result = _result( + [ + {"key": "body_shape", "status": "ok"}, + {"key": "color_season", "status": "skipped"}, + {"key": "style_type", "status": "ok"}, + ] + ) + assert ok_dimension_keys(result) == {"body_shape", "style_type"} + + def test_empty_or_missing_diagnosis_yields_empty(self) -> None: + assert ok_dimension_keys({}) == set() + assert ok_dimension_keys({"diagnosis": {}}) == set() + assert ok_dimension_keys({"diagnosis": {"dimensions": "not-a-list"}}) == set() + + def test_malformed_items_are_ignored(self) -> None: + result = _result( + [ + "garbage", + {"status": "ok"}, # 缺 key + {"key": 123, "status": "ok"}, # key 非字符串 + {"key": "face_shape", "status": "ok"}, + ] + ) + assert ok_dimension_keys(result) == {"face_shape"} + + +class TestFeedbackCreateRequest: + def test_verdict_enum_rejects_unknown_value(self) -> None: + with pytest.raises(ValidationError): + FeedbackCreateRequest.model_validate( + {"dimension_key": "body_shape", "verdict": "kind-of"} + ) + + def test_correction_max_200_chars(self) -> None: + with pytest.raises(ValidationError): + FeedbackCreateRequest.model_validate( + { + "dimension_key": "body_shape", + "verdict": "inaccurate", + "correction": "长" * 201, + } + ) + + def test_minimal_valid_payload(self) -> None: + req = FeedbackCreateRequest.model_validate( + {"dimension_key": "body_shape", "verdict": "accurate"} + ) + assert req.correction is None diff --git a/tests/unit/test_report_profile.py b/tests/unit/test_report_profile.py new file mode 100644 index 0000000..c6db77d --- /dev/null +++ b/tests/unit/test_report_profile.py @@ -0,0 +1,83 @@ +"""report.profile.latest_profile_summary：repo 打桩，验证摘要选取纪律。""" + +from __future__ import annotations + +import json +from types import SimpleNamespace +from typing import Any + +import pytest + +from packages.report import repo +from packages.report.profile import latest_profile_summary + +_DIAGNOSIS = { + "dimensions": [ + { + "key": "body_shape", + "label": "体型", + "status": "ok", + "value": "梨形", + "confidence": 0.9, + "evidence": ["问卷自报肩窄胯宽"], + }, + {"key": "style_impression", "status": "skipped", "value": None}, + ], + "rule_hits": [ + {"rule_key": "pear_lower_dark", "dimension": "body_shape", "suggestion": "下装选深色"} + ], +} + + +def _report(report_id: int, *, status: str, result: dict[str, Any]) -> Any: + return SimpleNamespace(report_id=report_id, status=status, result=result) + + +@pytest.fixture +def stub_list(monkeypatch: pytest.MonkeyPatch) -> dict[str, Any]: + state: dict[str, Any] = {"reports": []} + + async def fake_list(session: Any, *, user_id: int, limit: int = 20) -> list[Any]: + state["limit"] = limit + return state["reports"] + + monkeypatch.setattr(repo, "list_reports", fake_list) + return state + + +class TestLatestProfileSummary: + async def test_no_reports_returns_none(self, stub_list: dict[str, Any]) -> None: + assert await latest_profile_summary(None, user_id=1, max_chars=800) is None # type: ignore[arg-type] + + async def test_succeeded_report_with_diagnosis(self, stub_list: dict[str, Any]) -> None: + stub_list["reports"] = [_report(1, status="succeeded", result={"diagnosis": _DIAGNOSIS})] + summary = await latest_profile_summary(None, user_id=1, max_chars=800) # type: ignore[arg-type] + assert summary is not None + payload = json.loads(summary) + values = {d["key"]: d["value"] for d in payload["dimensions"]} + assert values["body_shape"] == "梨形" + # skipped 维度不进素材（防幻觉第一道闸） + assert "style_impression" not in values + assert payload["rule_hits"][0]["rule_key"] == "pear_lower_dark" + + async def test_latest_failed_falls_back_to_earlier_succeeded( + self, stub_list: dict[str, Any] + ) -> None: + stub_list["reports"] = [ + _report(2, status="failed", result={}), + _report(1, status="succeeded", result={"diagnosis": _DIAGNOSIS}), + ] + summary = await latest_profile_summary(None, user_id=1, max_chars=800) # type: ignore[arg-type] + assert summary is not None and "梨形" in summary + + async def test_succeeded_without_diagnosis_is_skipped(self, stub_list: dict[str, Any]) -> None: + stub_list["reports"] = [_report(1, status="succeeded", result={"vision": {}})] + assert await latest_profile_summary(None, user_id=1, max_chars=800) is None # type: ignore[arg-type] + + async def test_budget_is_forwarded(self, stub_list: dict[str, Any]) -> None: + """max_chars 收紧时低优先级维度被裁、硬结论保留（复用 copy 的裁剪纪律）。""" + stub_list["reports"] = [_report(1, status="succeeded", result={"diagnosis": _DIAGNOSIS})] + summary = await latest_profile_summary(None, user_id=1, max_chars=60) # type: ignore[arg-type] + assert summary is not None + assert len(summary) > 60 # 硬结论永不裁，载荷可超预算（copy.py 既定语义） + assert "梨形" in summary