from __future__ import annotations from typing import Any, Optional from sqlalchemy import and_, select from sqlalchemy.ext.asyncio import AsyncSession from ...models import ( Experiment, FeedbackCenterRecord, JudgeRun, OptimizerCandidate, PromptRelease, QARecord, QualityAsset, QualityAssetPair, ) from ...utils.timezone import now_cn_naive class ReplyIterationService: def __init__(self, db: AsyncSession): self.db = db @staticmethod def _asset_snapshot(row: QualityAsset) -> dict[str, Any]: return { "id": int(row.id), "tenant_id": int(row.tenant_id), "project_id": int(row.project_id) if row.project_id is not None else None, "asset_type": row.asset_type, "status": row.status, "source_feedback_record_id": int(row.source_feedback_record_id) if row.source_feedback_record_id is not None else None, "qa_record_id": int(row.qa_record_id) if row.qa_record_id is not None else None, "ai_call_log_id": int(row.ai_call_log_id) if row.ai_call_log_id is not None else None, "prompt_key": row.prompt_key, "workflow_type": row.workflow_type, "asked_via": row.asked_via, "planned_tier": row.planned_tier, "executed_tier": row.executed_tier, "route_source": row.route_source, "route_rule_name": row.route_rule_name, "question_text": row.question_text, "answer_text": row.answer_text, "corrected_answer": row.corrected_answer, "source_snapshot": row.source_snapshot or {}, "tags": row.tags or [], "notes": row.notes, "approved_by": int(row.approved_by) if row.approved_by is not None else None, "approved_at": row.approved_at, } @staticmethod def _pair_snapshot(row: QualityAssetPair) -> dict[str, Any]: return { "id": int(row.id), "tenant_id": int(row.tenant_id), "project_id": int(row.project_id) if row.project_id is not None else None, "pair_type": row.pair_type, "status": row.status, "chosen_asset_id": int(row.chosen_asset_id), "rejected_asset_id": int(row.rejected_asset_id), "reason_text": row.reason_text, "approved_by": int(row.approved_by) if row.approved_by is not None else None, "approved_at": row.approved_at, } @staticmethod def _normalize_feedback_rating(value: Any) -> str: return str(value or "").strip().lower() @classmethod def _build_feedback_backfill_judge_payload( cls, *, asset: QualityAsset, rating: str, ) -> dict[str, Any]: findings = cls._build_rule_findings(asset) if rating == "positive": base_rule = 96.0 base_llm = 95.0 else: base_rule = 88.0 base_llm = 87.0 penalty = min(8.0, float(len(findings)) * 2.0) rule_score = round(max(0.0, base_rule - penalty), 2) llm_score = round(max(0.0, base_llm - penalty), 2) overall = round((rule_score + llm_score) / 2.0, 2) return { "findings": findings, "rule_score": rule_score, "llm_score": llm_score, "overall_score": overall, } @staticmethod def _judge_snapshot(row: JudgeRun) -> dict[str, Any]: return { "id": int(row.id), "tenant_id": int(row.tenant_id), "project_id": int(row.project_id) if row.project_id is not None else None, "asset_id": int(row.asset_id) if row.asset_id is not None else None, "pair_id": int(row.pair_id) if row.pair_id is not None else None, "prompt_key": row.prompt_key, "workflow_type": row.workflow_type, "asked_via": row.asked_via, "planned_tier": row.planned_tier, "executed_tier": row.executed_tier, "route_rule_name": row.route_rule_name, "judge_status": row.judge_status, "bucket_key": row.bucket_key, "rule_score": float(row.rule_score or 0.0) if row.rule_score is not None else None, "llm_score": float(row.llm_score or 0.0) if row.llm_score is not None else None, "overall_score": float(row.overall_score or 0.0) if row.overall_score is not None else None, "findings": row.findings or [], "metadata_json": row.metadata_json or {}, } @staticmethod def _candidate_snapshot(row: OptimizerCandidate) -> dict[str, Any]: return { "id": int(row.id), "tenant_id": int(row.tenant_id), "project_id": int(row.project_id) if row.project_id is not None else None, "judge_run_id": int(row.judge_run_id), "candidate_type": row.candidate_type, "target_type": row.target_type, "target_ref": row.target_ref, "status": row.status, "suggestion": row.suggestion or {}, "evidence": row.evidence or {}, "notes": row.notes, } @staticmethod def _experiment_snapshot(row: Experiment) -> dict[str, Any]: return { "id": int(row.id), "tenant_id": int(row.tenant_id), "title": row.title, "hypothesis": row.hypothesis, "status": row.status, "target_metric": row.target_metric, "current_score": float(row.current_score or 0.0) if row.current_score is not None else None, "experiment_type": row.experiment_type, "baseline_release_id": int(row.baseline_release_id) if row.baseline_release_id is not None else None, "candidate_release_id": int(row.candidate_release_id) if row.candidate_release_id is not None else None, "scope_json": row.scope_json or {}, "result_summary": row.result_summary or {}, "decision": row.decision, "project_id": int(row.project_id) if row.project_id is not None else None, } async def promote_feedback_record( self, *, tenant_id: int, record_id: int, actor_id: Optional[int], asset_type: str, prompt_key: Optional[str] = None, workflow_type: Optional[str] = None, notes: Optional[str] = None, pair_with_asset_id: Optional[int] = None, pair_role: Optional[str] = None, pair_reason: Optional[str] = None, ) -> dict[str, Any]: record = ( await self.db.execute( select(FeedbackCenterRecord).where( FeedbackCenterRecord.tenant_id == int(tenant_id), FeedbackCenterRecord.id == int(record_id), ) ) ).scalar_one_or_none() if record is None: raise ValueError("feedback record not found") qa_record = None if str(record.object_type or "").strip() == "qa_record" and record.object_id is not None: qa_record = ( await self.db.execute( select(QARecord).where( and_( QARecord.tenant_id == int(tenant_id), QARecord.id == int(record.object_id), ) ) ) ).scalar_one_or_none() before = record.before_value if isinstance(record.before_value, dict) else {} after = record.after_value if isinstance(record.after_value, dict) else {} ai_snapshot = record.ai_snapshot if isinstance(record.ai_snapshot, dict) else {} routing_snapshot = ( qa_record.routing_snapshot if qa_record is not None and isinstance(qa_record.routing_snapshot, dict) else {} ) asset = QualityAsset( tenant_id=int(tenant_id), project_id=int(record.project_id) if record.project_id is not None else None, asset_type=str(asset_type or "").strip().lower(), status="approved", source_feedback_record_id=int(record.id), qa_record_id=int(qa_record.id) if qa_record is not None else None, prompt_key=str(prompt_key or ai_snapshot.get("prompt_key") or "").strip() or None, workflow_type=str(workflow_type or ai_snapshot.get("workflow_type") or "query.read_only").strip() or None, asked_via=str(getattr(qa_record, "asked_via", "") or "").strip() or None, planned_tier=str(routing_snapshot.get("planned_tier") or "").strip() or None, executed_tier=str(routing_snapshot.get("executed_tier") or "").strip() or None, route_source=str(routing_snapshot.get("route_source") or "").strip() or None, route_rule_name=str(routing_snapshot.get("route_rule_name") or "").strip() or None, question_text=str(before.get("question") or getattr(qa_record, "question", "") or "").strip(), answer_text=str(before.get("answer") or getattr(qa_record, "answer", "") or "").strip(), corrected_answer=str(after.get("correct_answer") or record.reason_text or "").strip() or None, source_snapshot={ "feedback_record_id": int(record.id), "reason_code": record.reason_code, "review_status": record.review_status, "routing_snapshot": routing_snapshot, }, tags=[ code for code in ( str(record.reason_code or "").strip(), str(after.get("rating") or "").strip(), ) if code ], notes=str(notes or "").strip() or None, approved_by=actor_id, approved_at=now_cn_naive().isoformat(), ) if not asset.question_text or not asset.answer_text: raise ValueError("feedback record missing question/answer evidence") self.db.add(asset) await self.db.flush() pair_payload = None if pair_with_asset_id is not None: paired_asset = ( await self.db.execute( select(QualityAsset).where( QualityAsset.tenant_id == int(tenant_id), QualityAsset.id == int(pair_with_asset_id), ) ) ).scalar_one_or_none() if paired_asset is None: raise ValueError("paired quality asset not found") role = str(pair_role or "chosen").strip().lower() chosen_id = int(asset.id) if role == "chosen" else int(paired_asset.id) rejected_id = int(paired_asset.id) if role == "chosen" else int(asset.id) pair = QualityAssetPair( tenant_id=int(tenant_id), project_id=int(record.project_id) if record.project_id is not None else None, pair_type="preference", status="approved", chosen_asset_id=chosen_id, rejected_asset_id=rejected_id, reason_text=str(pair_reason or "").strip() or None, approved_by=actor_id, approved_at=now_cn_naive().isoformat(), ) self.db.add(pair) await self.db.flush() pair_payload = self._pair_snapshot(pair) await self.db.commit() await self.db.refresh(asset) return {"asset": self._asset_snapshot(asset), "pair": pair_payload} async def backfill_mounted_training_assets_from_feedback( self, *, tenant_id: int, actor_id: Optional[int], limit: int = 200, ) -> dict[str, Any]: records = ( await self.db.execute( select(FeedbackCenterRecord, QARecord) .outerjoin( QARecord, and_( QARecord.tenant_id == FeedbackCenterRecord.tenant_id, FeedbackCenterRecord.object_type == "qa_record", QARecord.id == FeedbackCenterRecord.object_id, ), ) .where( FeedbackCenterRecord.tenant_id == int(tenant_id), FeedbackCenterRecord.review_status == "reviewed", FeedbackCenterRecord.object_type == "qa_record", ) .order_by(FeedbackCenterRecord.id.asc()) .limit(max(1, int(limit))), ) ).all() positive_assets: list[QualityAsset] = [] negative_assets: list[QualityAsset] = [] created_assets = 0 created_pairs = 0 created_judges = 0 skipped_records = 0 existing_assets = ( await self.db.execute( select(QualityAsset).where( QualityAsset.tenant_id == int(tenant_id), QualityAsset.source_feedback_record_id.is_not(None), ) ) ).scalars().all() asset_by_feedback: dict[int, QualityAsset] = { int(row.source_feedback_record_id): row for row in existing_assets if row.source_feedback_record_id is not None } for row in existing_assets: if str(row.asset_type or "").strip().lower() == "golden": positive_assets.append(row) elif str(row.asset_type or "").strip().lower() == "failure": negative_assets.append(row) for feedback, qa_record in records: after_value = feedback.after_value if isinstance(feedback.after_value, dict) else {} before_value = feedback.before_value if isinstance(feedback.before_value, dict) else {} ai_snapshot = feedback.ai_snapshot if isinstance(feedback.ai_snapshot, dict) else {} routing_snapshot = ( qa_record.routing_snapshot if qa_record is not None and isinstance(qa_record.routing_snapshot, dict) else {} ) rating = self._normalize_feedback_rating(after_value.get("rating")) if rating not in {"positive", "negative"}: skipped_records += 1 continue asset = asset_by_feedback.get(int(feedback.id)) if asset is None: question_text = str( before_value.get("question") or getattr(qa_record, "question", "") or "" ).strip() answer_text = str( before_value.get("answer") or getattr(qa_record, "answer", "") or "" ).strip() corrected_answer = str( after_value.get("correct_answer") or feedback.reason_text or answer_text ).strip() if not question_text or not answer_text: skipped_records += 1 continue asset = QualityAsset( tenant_id=int(tenant_id), project_id=int(feedback.project_id) if feedback.project_id is not None else None, asset_type="golden" if rating == "positive" else "failure", status="approved", source_feedback_record_id=int(feedback.id), qa_record_id=int(qa_record.id) if qa_record is not None else None, prompt_key=str(ai_snapshot.get("prompt_key") or "mounted.runtime.reply").strip() or None, workflow_type=str(ai_snapshot.get("workflow_type") or "query.read_only").strip() or None, asked_via=str(getattr(qa_record, "asked_via", "") or "").strip() or None, planned_tier=str(routing_snapshot.get("planned_tier") or "").strip() or None, executed_tier=str(routing_snapshot.get("executed_tier") or "").strip() or None, route_source=str(routing_snapshot.get("route_source") or "").strip() or None, route_rule_name=str(routing_snapshot.get("route_rule_name") or "").strip() or None, question_text=question_text, answer_text=answer_text, corrected_answer=corrected_answer or None, source_snapshot={ "backfill_source": "feedback_center_qa", "feedback_record_id": int(feedback.id), "rating": rating, "reason_code": feedback.reason_code, }, tags=["mounted_runtime", "real_feedback_backfill", f"rating_{rating}"], notes="Backfilled from reviewed feedback_center_records + qa_records.", approved_by=actor_id, approved_at=now_cn_naive().isoformat(), ) self.db.add(asset) await self.db.flush() asset_by_feedback[int(feedback.id)] = asset created_assets += 1 judge = ( await self.db.execute( select(JudgeRun).where( JudgeRun.tenant_id == int(tenant_id), JudgeRun.asset_id == int(asset.id), ) ) ).scalar_one_or_none() if judge is None: score_payload = self._build_feedback_backfill_judge_payload(asset=asset, rating=rating) judge = JudgeRun( tenant_id=int(tenant_id), project_id=int(asset.project_id) if asset.project_id is not None else None, asset_id=int(asset.id), prompt_key=asset.prompt_key, workflow_type=asset.workflow_type, asked_via=asset.asked_via, planned_tier=asset.planned_tier, executed_tier=asset.executed_tier, route_rule_name=asset.route_rule_name, judge_status="completed", bucket_key="|".join( [ str(asset.prompt_key or "unknown"), str(asset.workflow_type or "unknown"), str(asset.asked_via or "unknown"), str(asset.planned_tier or "unknown"), str(asset.executed_tier or "unknown"), ] ), rule_score=score_payload["rule_score"], llm_score=score_payload["llm_score"], overall_score=score_payload["overall_score"], findings=score_payload["findings"], metadata_json={ "backfill_source": "feedback_center_qa", "feedback_record_id": int(feedback.id), "rating": rating, }, ) self.db.add(judge) created_judges += 1 if rating == "positive": if all(int(item.id) != int(asset.id) for item in positive_assets): positive_assets.append(asset) else: if all(int(item.id) != int(asset.id) for item in negative_assets): negative_assets.append(asset) await self.db.flush() for negative_asset in negative_assets: if not positive_assets: break pair_exists = ( await self.db.execute( select(QualityAssetPair).where( QualityAssetPair.tenant_id == int(tenant_id), QualityAssetPair.rejected_asset_id == int(negative_asset.id), ) ) ).scalar_one_or_none() if pair_exists is not None: continue chosen_asset = next( ( item for item in positive_assets if str(item.asked_via or "").strip() == str(negative_asset.asked_via or "").strip() ), positive_assets[0] if positive_assets else None, ) if chosen_asset is None: continue pair = QualityAssetPair( tenant_id=int(tenant_id), project_id=int(chosen_asset.project_id) if chosen_asset.project_id is not None else None, pair_type="preference", status="approved", chosen_asset_id=int(chosen_asset.id), rejected_asset_id=int(negative_asset.id), reason_text="Backfilled pair from reviewed positive/negative feedback samples.", approved_by=actor_id, approved_at=now_cn_naive().isoformat(), ) self.db.add(pair) created_pairs += 1 await self.db.commit() return { "created_assets": int(created_assets), "created_pairs": int(created_pairs), "created_judge_runs": int(created_judges), "skipped_records": int(skipped_records), "positive_asset_count": len(positive_assets), "negative_asset_count": len(negative_assets), } async def list_assets( self, *, tenant_id: int, asset_type: Optional[str] = None, prompt_key: Optional[str] = None, limit: int = 20, offset: int = 0, ) -> dict[str, Any]: stmt = select(QualityAsset).where(QualityAsset.tenant_id == int(tenant_id)) if asset_type: stmt = stmt.where(QualityAsset.asset_type == str(asset_type).strip().lower()) if prompt_key: stmt = stmt.where(QualityAsset.prompt_key == str(prompt_key).strip()) rows = ( await self.db.execute( stmt.order_by(QualityAsset.id.desc()).offset(int(offset)).limit(int(limit)) ) ).scalars().all() return { "items": [self._asset_snapshot(row) for row in rows], "total": len(rows), "limit": int(limit), "offset": int(offset), } @staticmethod def _build_rule_findings(asset: QualityAsset) -> list[dict[str, Any]]: findings: list[dict[str, Any]] = [] if str(asset.planned_tier or "").strip() == "low" and str(asset.executed_tier or "").strip() == "high": findings.append({"code": "tier_escalation", "severity": "info", "message": "存在 low -> high 升级"}) return findings async def create_judge_run( self, *, tenant_id: int, asset_id: int, actor_id: Optional[int], ) -> dict[str, Any]: _ = actor_id asset = ( await self.db.execute( select(QualityAsset).where( QualityAsset.tenant_id == int(tenant_id), QualityAsset.id == int(asset_id), ) ) ).scalar_one_or_none() if asset is None: raise ValueError("quality asset not found") findings = self._build_rule_findings(asset) penalty = min(0.8, 0.25 * len(findings)) score = round(max(0.0, 1.0 - penalty), 4) bucket_key = "|".join( [ str(asset.prompt_key or "unknown"), str(asset.workflow_type or "unknown"), str(asset.asked_via or "unknown"), str(asset.planned_tier or "unknown"), str(asset.executed_tier or "unknown"), ] ) judge = JudgeRun( tenant_id=int(tenant_id), project_id=int(asset.project_id) if asset.project_id is not None else None, asset_id=int(asset.id), prompt_key=asset.prompt_key, workflow_type=asset.workflow_type, asked_via=asset.asked_via, planned_tier=asset.planned_tier, executed_tier=asset.executed_tier, route_rule_name=asset.route_rule_name, judge_status="completed", bucket_key=bucket_key, rule_score=score, llm_score=None, overall_score=score, findings=findings, metadata_json={"llm_status": "pending", "asset_type": asset.asset_type}, ) self.db.add(judge) await self.db.commit() await self.db.refresh(judge) return self._judge_snapshot(judge) async def list_judge_runs(self, *, tenant_id: int, limit: int = 20, offset: int = 0) -> dict[str, Any]: rows = ( await self.db.execute( select(JudgeRun) .where(JudgeRun.tenant_id == int(tenant_id)) .order_by(JudgeRun.id.desc()) .offset(int(offset)) .limit(int(limit)) ) ).scalars().all() return {"items": [self._judge_snapshot(row) for row in rows], "total": len(rows), "limit": int(limit), "offset": int(offset)} async def create_optimizer_candidate( self, *, tenant_id: int, judge_run_id: int, actor_id: Optional[int], ) -> dict[str, Any]: _ = actor_id judge = ( await self.db.execute( select(JudgeRun).where( JudgeRun.tenant_id == int(tenant_id), JudgeRun.id == int(judge_run_id), ) ) ).scalar_one_or_none() if judge is None: raise ValueError("judge run not found") findings = judge.findings if isinstance(judge.findings, list) else [] finding_codes = {str(item.get("code") or "").strip() for item in findings if isinstance(item, dict)} if {"process_talk_prefix", "execution_claim_prefix"}.intersection(finding_codes): candidate_type = "renderer_policy_change" target_type = "renderer_policy" target_ref = "reply_renderer" summary = "收紧过程话术和执行型前缀" elif "tier_escalation" in finding_codes: candidate_type = "routing_policy_change" target_type = "routing_policy" target_ref = "model_routing" summary = "复核 low/high 路由升级门槛" else: candidate_type = "prompt_change" target_type = "prompt_key" target_ref = str(judge.prompt_key or "unknown") summary = "优化回答提示词结构" row = OptimizerCandidate( tenant_id=int(tenant_id), project_id=int(judge.project_id) if judge.project_id is not None else None, judge_run_id=int(judge.id), candidate_type=candidate_type, target_type=target_type, target_ref=target_ref, status="draft", suggestion={"summary": summary, "bucket_key": judge.bucket_key}, evidence={"findings": findings, "overall_score": judge.overall_score}, ) self.db.add(row) await self.db.commit() await self.db.refresh(row) return self._candidate_snapshot(row) async def list_candidates(self, *, tenant_id: int, limit: int = 20, offset: int = 0) -> dict[str, Any]: rows = ( await self.db.execute( select(OptimizerCandidate) .where(OptimizerCandidate.tenant_id == int(tenant_id)) .order_by(OptimizerCandidate.id.desc()) .offset(int(offset)) .limit(int(limit)) ) ).scalars().all() return {"items": [self._candidate_snapshot(row) for row in rows], "total": len(rows), "limit": int(limit), "offset": int(offset)} async def list_trials(self, *, tenant_id: int, limit: int = 20, offset: int = 0) -> dict[str, Any]: rows = ( await self.db.execute( select(Experiment) .where( Experiment.tenant_id == int(tenant_id), Experiment.experiment_type == "prompt_release_trial", ) .order_by(Experiment.id.desc()) .offset(int(offset)) .limit(int(limit)) ) ).scalars().all() return {"items": [self._experiment_snapshot(row) for row in rows], "total": len(rows), "limit": int(limit), "offset": int(offset)} async def create_trial( self, *, tenant_id: int, actor_id: Optional[int], title: str, hypothesis: str, target_metric: str, prompt_key: str, baseline_release_id: int, candidate_release_id: int, scope_json: Optional[dict[str, Any]], ) -> dict[str, Any]: for release_id in (baseline_release_id, candidate_release_id): release = ( await self.db.execute( select(PromptRelease).where( PromptRelease.tenant_id == int(tenant_id), PromptRelease.id == int(release_id), PromptRelease.prompt_key == str(prompt_key or "").strip(), ) ) ).scalar_one_or_none() if release is None: raise ValueError("prompt release not found for trial") row = Experiment( tenant_id=int(tenant_id), title=str(title or "").strip(), hypothesis=str(hypothesis or "").strip() or None, status="running", target_metric=str(target_metric or "").strip() or None, experiment_type="prompt_release_trial", baseline_release_id=int(baseline_release_id), candidate_release_id=int(candidate_release_id), scope_json=scope_json or {}, decision=None, ) row.linked_context_nodes = {"prompt_key": str(prompt_key or "").strip(), "actor_id": actor_id} self.db.add(row) await self.db.commit() await self.db.refresh(row) return self._experiment_snapshot(row) async def decide_trial( self, *, tenant_id: int, trial_id: int, actor_id: Optional[int], decision: str, result_summary: Optional[dict[str, Any]], ) -> dict[str, Any]: row = ( await self.db.execute( select(Experiment).where( Experiment.tenant_id == int(tenant_id), Experiment.id == int(trial_id), Experiment.experiment_type == "prompt_release_trial", ) ) ).scalar_one_or_none() if row is None: raise ValueError("prompt release trial not found") normalized = str(decision or "").strip().lower() if normalized not in {"accepted", "rejected", "rolled_back"}: raise ValueError("unsupported decision") row.decision = normalized row.status = "success" if normalized == "accepted" else "failed" row.result_summary = result_summary or {} row.current_score = float((result_summary or {}).get("accepted_rate") or 0.0) if result_summary else row.current_score row.linked_context_nodes = { **(row.linked_context_nodes if isinstance(row.linked_context_nodes, dict) else {}), "decision_actor_id": actor_id, "decided_at": now_cn_naive().isoformat(), } await self.db.commit() await self.db.refresh(row) return self._experiment_snapshot(row)