hello-agents/Co-creation-projects/aatanxiao12-beep-YingQian/backend/app/agents/movie_recommender_agent.py

"""多智能体电影推荐编排（串行流水线）。

范式：Pipeline + Tool-use
  ① 画像 Agent（无工具）→ TasteProfile
  ② 检索 Agent（挂 MovieTool）→ 真片候选
  ③ 推荐 Agent（无工具）→ 仅在候选 id 内产出 RecommendResult

本模块只提供编排器；HTTP 路由后续再接。
"""

from __future__ import annotations

import json
import re
import time
from datetime import datetime
from typing import Any, List, Optional, Tuple

from hello_agents import Config, SimpleAgent

from ..config import get_settings
from ..models.schemas import (
    CandidateMovie,
    MovieCard,
    RecommendRequest,
    RecommendResult,
    TasteProfile,
)
from ..services.llm_service import get_llm
from ..services.movie_service import get_movie_service, normalize_tmdb_language
from ..tools.movie_tool import get_movie_tool
from ..utils.logger import get_logger

logger = get_logger("app.agents")

# ============ Prompts ============

PROFILE_AGENT_PROMPT = """你是观影口味画像专家。根据用户偏好输出结构化 JSON，不要推荐具体片名。

只返回如下 JSON（不要 Markdown 代码块外的解释）:
{
  "summary": "一句话口味摘要",
  "genre_hints": ["类型1", "类型2"],
  "language_hints": ["仅填 ISO 码：zh / en / ja / ko，可空；禁止写好莱坞、英语等中文"],
  "avoid": ["需规避的内容"],
  "discover_notes": "给 TMDB discover 用的简短检索说明"
}
"""

SEARCH_AGENT_PROMPT = """你是电影检索专家。必须调用工具从 TMDB 取真实影片，禁止编造片名。

可用工具:
- movies_discover: 主路径，按类型/年份/时长/语言发现（本轮只允许调用 1 次）
- movies_search: 仅当需要解析「已看片名」时再用（可选，最多 1~2 次）

硬规则:
1. movies_discover 只调用一次：用建议参数一次取够候选，禁止换参反复 discover
2. with_original_language 只能是 zh/en/ja/ko；不要传「好莱坞」「英语」等中文
3. 拿到工具结果后立即输出最终 JSON，不要再调工具「精炼」
4. 最终 movies 必须从工具结果原样抄写关键字段（含 poster_url）

取数后，最终回复必须是 JSON（不要多余解释）:
{
  "movies": [
    {
      "id": 123,
      "title": "...",
      "year": 2020,
      "genres": [],
      "rating": 7.5,
      "poster_url": "https://...",
      "overview": "..."
    }
  ]
}

要求:
1. 尽量返回 15~25 部
2. id / title / poster_url 等必须来自工具结果，禁止省略 poster_url
3. 排除用户已给出的 exclude_ids
"""

RECOMMEND_AGENT_PROMPT = """你是电影推荐专家。你没有外部工具，只能从「候选列表」中挑选 3~5 部。

硬约束:
1. 每部电影的 id 必须出现在候选列表中
2. 禁止编造候选之外的片名或 id
3. 遵守 spoilers_ok：若为 false，overview_safe 不要写结局剧透
4. why 要贴合用户心情与人群
5. title / year / genres / rating / poster_url 尽量原样沿用候选列表（勿改写为空）

只返回 JSON:
{
  "playlist_name": "片单主题名",
  "profile_summary": "对用户口味的一句话总结",
  "movies": [
    {
      "id": 123,
      "title": "...",
      "year": 2020,
      "genres": ["..."],
      "runtime": null,
      "rating": 7.5,
      "poster_url": "https://image.tmdb.org/t/p/w500/...",
      "why": "推荐理由",
      "vibe_tags": ["标签"],
      "caution": null,
      "overview_safe": "安全简介"
    }
  ],
  "is_fallback": false
}
"""


REGION_LANGUAGE = {
    "华语": "zh",
    "好莱坞": "en",
    "日韩": "ja",  # 简化：先按日语；韩语可由画像 language_hints 覆盖
    "欧洲": "",
    "不限": "",
}


class MultiAgentMovieRecommender:
    """串行三 Agent 推荐编排器（画像 → 检索 → 推荐 + 白名单校验）。"""

    def __init__(self) -> None:
        """初始化共享 LLM / MovieTool，并创建三个 SimpleAgent。"""
        self.llm = get_llm()
        self.movie_tool = get_movie_tool()
        settings = get_settings()
        # Trace 开关来自 .env：TRACE_ENABLED / TRACE_DIR
        agent_config = Config(
            trace_enabled=settings.trace_enabled,
            trace_dir=settings.trace_dir,
        )

        # 画像：只做偏好结构化，禁止挂工具（避免这步就去搜片/编片名）
        self.profile_agent = SimpleAgent(
            name="画像专家",
            llm=self.llm,
            system_prompt=PROFILE_AGENT_PROMPT,
            config=agent_config,
            enable_tool_calling=False,
        )
        # 检索：唯一允许碰 TMDB 的 Agent；工具展开为 discover / search
        # max_tool_iterations=2：1 轮工具 + 1 轮收尾文本；再高容易反复换参 discover
        self.search_agent = SimpleAgent(
            name="检索专家",
            llm=self.llm,
            system_prompt=SEARCH_AGENT_PROMPT,
            config=agent_config,
            max_tool_iterations=2,
        )
        self.search_agent.add_tool(self.movie_tool)

        # 推荐：无工具，只能在上游候选里选择与说理（防幻觉核心）
        self.recommend_agent = SimpleAgent(
            name="推荐专家",
            llm=self.llm,
            system_prompt=RECOMMEND_AGENT_PROMPT,
            config=agent_config,
            enable_tool_calling=False,
        )
        logger.info(
            "MultiAgentMovieRecommender 就绪: tools=%s trace=%s",
            self.search_agent.list_tools(),
            settings.trace_enabled,
        )

    def recommend(self, request: RecommendRequest) -> Tuple[RecommendResult, str]:
        """跑完整推荐流水线。

        Returns:
            (RecommendResult, message)：业务结果 + 给人看的状态说明（含降级提示）。
        """
        pipeline_t0 = time.perf_counter()
        try:
            logger.info("推荐开始 mood=%s party=%s", request.mood, request.party_type)

            # ① 偏好 → TasteProfile；换一批可携带 taste_profile 跳过画像 LLM
            t0 = time.perf_counter()
            profile, profile_reused = self._resolve_profile(request)
            logger.info(
                "阶段完成 stage=profile elapsed=%.2fs reused=%s summary=%s",
                time.perf_counter() - t0,
                profile_reused,
                profile.summary,
            )

            # ② 真片候选；失败则降级，避免在空列表上瞎荐
            t0 = time.perf_counter()
            candidates = self._run_search(request, profile)
            logger.info(
                "阶段完成 stage=search elapsed=%.2fs candidates=%d",
                time.perf_counter() - t0,
                len(candidates),
            )
            if not candidates:
                result = self._fallback_result(request, profile, [], "未取得候选片")
                return result, "检索无结果，已返回降级片单"

            # ③ 候选内推荐 + 代码层 id 白名单（不信任模型自觉）
            t0 = time.perf_counter()
            result = self._run_recommend(request, profile, candidates)
            logger.info(
                "阶段完成 stage=recommend_llm elapsed=%.2fs",
                time.perf_counter() - t0,
            )
            t0 = time.perf_counter()
            result = self._enforce_candidate_ids(result, candidates, profile)
            result = self._attach_taste_profile(result, profile)
            logger.info(
                "阶段完成 stage=enforce elapsed=%.2fs movies=%d fallback=%s total=%.2fs",
                time.perf_counter() - t0,
                len(result.movies),
                result.is_fallback,
                time.perf_counter() - pipeline_t0,
            )
            msg = "推荐生成成功" if not result.is_fallback else "推荐已做 id 校正/降级"
            if profile_reused:
                msg = f"{msg}（已跳过画像）"
            return result, msg

        except Exception as e:
            # 未捕获异常也返回完整结构，前端不白屏
            logger.exception("推荐流水线异常")
            result = self._fallback_result(request, None, [], str(e))
            return result, f"推荐异常，已降级: {e}"

    # ----- stages -----

    def _resolve_profile(self, request: RecommendRequest) -> Tuple[TasteProfile, bool]:
        """解析画像：请求携带可用 taste_profile 则复用，否则跑画像 Agent。"""
        reused = request.taste_profile
        if reused is not None and (
            (reused.summary or "").strip()
            or reused.genre_hints
            or (reused.discover_notes or "").strip()
        ):
            logger.info("跳过画像 Agent，复用请求中的 taste_profile")
            return self._sanitize_profile(reused, request), True
        return self._sanitize_profile(self._run_profile(request), request), False

    def _sanitize_profile(
        self,
        profile: TasteProfile,
        request: RecommendRequest,
    ) -> TasteProfile:
        """规范化 language_hints 为 ISO 码；非法项丢弃。"""
        cleaned: List[str] = []
        for hint in profile.language_hints or []:
            code = normalize_tmdb_language(hint)
            if code and code not in cleaned:
                cleaned.append(code)
        if not cleaned:
            fallback = normalize_tmdb_language(
                REGION_LANGUAGE.get(request.region_preference, "")
            )
            if fallback:
                cleaned = [fallback]
        if cleaned != list(profile.language_hints or []):
            logger.info(
                "画像 language_hints 已归一化: %s -> %s",
                profile.language_hints,
                cleaned,
            )
        profile.language_hints = cleaned
        return profile

    def _resolve_language(
        self,
        request: RecommendRequest,
        profile: TasteProfile,
    ) -> Optional[str]:
        """解析最终用于 discover 的语言码。"""
        if profile.language_hints:
            code = normalize_tmdb_language(profile.language_hints[0])
            if code:
                return code
        return normalize_tmdb_language(
            REGION_LANGUAGE.get(request.region_preference, "")
        )

    def _attach_taste_profile(
        self,
        result: RecommendResult,
        profile: TasteProfile,
    ) -> RecommendResult:
        """把本次画像挂到结果上，供换一批回传。"""
        result.taste_profile = profile
        if not result.profile_summary:
            result.profile_summary = profile.summary
        return result

    def _run_profile(self, request: RecommendRequest) -> TasteProfile:
        """阶段①：调用画像 Agent，解析为 TasteProfile；失败则用表单字段兜底。"""
        self.profile_agent.clear_history()
        raw = self.profile_agent.run(self._build_profile_query(request))
        data = self._extract_json(raw) or {}
        try:
            return TasteProfile(**data)
        except Exception:
            # 画像 JSON 坏了：用表单字段拼可用 profile，保证后续检索能继续
            return TasteProfile(
                summary=f"{request.mood}/{request.party_type} 观影",
                genre_hints=list(request.genres),
                language_hints=[REGION_LANGUAGE.get(request.region_preference, "")],
                avoid=[],
                discover_notes=request.free_text or "",
            )

    def _run_search(
        self,
        request: RecommendRequest,
        profile: TasteProfile,
    ) -> List[CandidateMovie]:
        """阶段②：检索 Agent 调工具取真片；解析失败则 MovieService 规则 discover 兜底。"""
        self.search_agent.clear_history()
        self.movie_tool.begin_search_run(discover_limit=1)
        t0 = time.perf_counter()
        try:
            raw = self.search_agent.run(self._build_search_query(request, profile))
        finally:
            self.movie_tool.end_search_run()
        logger.info(
            "检索 Agent run 结束 elapsed=%.2fs raw_len=%d",
            time.perf_counter() - t0,
            len(raw or ""),
        )
        movies = self._parse_candidates(raw, request.exclude_ids)
        if movies:
            missing_poster = sum(1 for m in movies if not m.poster_url)
            logger.info(
                "检索 Agent 解析成功 count=%d missing_poster=%d",
                len(movies),
                missing_poster,
            )
            return movies

        # Agent 未给出可用 JSON 时，用 profile 规则直连 MovieService（仍是真数据）
        logger.warning("检索 Agent 未解析出候选，改用 MovieService 规则兜底")
        t0 = time.perf_counter()
        fallback = self._discover_by_profile(request, profile)
        logger.info(
            "阶段完成 stage=search_fallback_discover elapsed=%.2fs count=%d",
            time.perf_counter() - t0,
            len(fallback),
        )
        return fallback

    def _run_recommend(
        self,
        request: RecommendRequest,
        profile: TasteProfile,
        candidates: List[CandidateMovie],
    ) -> RecommendResult:
        """阶段③：推荐 Agent 仅在候选内产出 RecommendResult；JSON 坏则降级。"""
        self.recommend_agent.clear_history()
        raw = self.recommend_agent.run(
            self._build_recommend_query(request, profile, candidates)
        )
        data = self._extract_json(raw)
        if not data:
            return self._fallback_result(request, profile, candidates, "推荐 JSON 解析失败")
        try:
            data.setdefault("is_fallback", False)
            # 画像由编排器挂载，不采信模型自带的 taste_profile 字段
            data.pop("taste_profile", None)
            return RecommendResult(**data)
        except Exception:
            return self._fallback_result(request, profile, candidates, "推荐结构校验失败")

    # ----- queries -----

    def _build_profile_query(self, request: RecommendRequest) -> str:
        """把 RecommendRequest 拼成画像 Agent 的用户输入文本。"""
        return (
            f"心情: {request.mood}\n"
            f"人群: {request.party_type}\n"
            f"类型偏好: {', '.join(request.genres) or '无'}\n"
            f"时长上限(分钟): {request.max_runtime_minutes}\n"
            f"地区: {request.region_preference}\n"
            f"年代: {request.year_preference}\n"
            f"已看过: {', '.join(request.exclude_titles) or '无'}\n"
            f"允许剧透: {request.spoilers_ok}\n"
            f"额外要求: {request.free_text or '无'}\n"
            "请输出 TasteProfile JSON。"
        )

    def _build_search_query(self, request: RecommendRequest, profile: TasteProfile) -> str:
        """把画像 + 表单约束拼成检索 Agent 输入（含建议的 discover 参数）。"""
        # 预先算好 discover 参数提示，降低模型乱填工具参数的概率
        year_gte, year_lte = self._year_bounds(request.year_preference)
        lang = self._resolve_language(request, profile) or ""

        genres = ",".join(profile.genre_hints or request.genres)
        parts = [
            "请只调用一次 movies_discover（用下列建议参数），取到结果后立刻输出 JSON；不要反复换参 discover。",
            f"画像摘要: {profile.summary}",
            f"建议 with_genres: {genres or '不限'}",
            f"建议 year_gte: {year_gte or 0}, year_lte: {year_lte or 0}",
            f"建议 max_runtime: {request.max_runtime_minutes or 0}",
            f"建议 with_original_language: {lang or '不限'}（仅 zh/en/ja/ko）",
            f"discover_notes: {profile.discover_notes}",
            f"exclude_ids: {request.exclude_ids}",
            f"已看片名(仅必要时用 movies_search 辅助排除): {request.exclude_titles}",
            "最终只输出含 movies 数组的 JSON，且每部必须带工具返回的 poster_url。",
        ]
        return "\n".join(parts)

    def _build_recommend_query(
        self,
        request: RecommendRequest,
        profile: TasteProfile,
        candidates: List[CandidateMovie],
    ) -> str:
        """把用户偏好 + 精简候选列表拼成推荐 Agent 输入。"""
        # 只塞精简字段进 prompt；片名/海报等最终以候选元数据为准
        slim = [
            {
                "id": c.id,
                "title": c.title,
                "year": c.year,
                "genres": c.genres,
                "rating": c.rating,
                "poster_url": c.poster_url,
                "overview": (c.overview or "")[:180],
            }
            for c in candidates
        ]
        return (
            f"用户心情: {request.mood}; 人群: {request.party_type}; "
            f"剧透允许: {request.spoilers_ok}\n"
            f"画像: {profile.summary}\n"
            f"额外要求: {request.free_text or '无'}\n"
            f"候选列表(只能从中选):\n{json.dumps(slim, ensure_ascii=False)}\n"
            "请输出 RecommendResult JSON（3~5 部）。"
        )

    # ----- helpers -----

    @staticmethod
    def _year_bounds(year_preference: str) -> Tuple[Optional[int], Optional[int]]:
        """表单年代偏好 → TMDB discover 的 (year_gte, year_lte)。"""
        year = datetime.now().year
        if year_preference == "近5年":
            return year - 5, None
        if year_preference == "近10年":
            return year - 10, None
        if year_preference == "经典":
            return None, 2000
        return None, None

    def _discover_by_profile(
        self,
        request: RecommendRequest,
        profile: TasteProfile,
    ) -> List[CandidateMovie]:
        """不经 LLM，按画像字段确定性 discover；空结果自动放宽条件。"""
        year_gte, year_lte = self._year_bounds(request.year_preference)
        lang = self._resolve_language(request, profile)
        genres = ",".join(profile.genre_hints or request.genres) or None
        return get_movie_service().discover_with_relax(
            with_genres=genres,
            year_gte=year_gte,
            year_lte=year_lte,
            max_runtime=request.max_runtime_minutes,
            with_original_language=lang,
            page=1,
            exclude_ids=request.exclude_ids,
        )

    def _parse_candidates(self, raw: str, exclude_ids: List[int]) -> List[CandidateMovie]:
        """从检索 Agent 文本抽出 movies，过滤 exclude_ids 与空标题。"""
        data = self._extract_json(raw)
        if not data:
            return []
        items = data.get("movies") if isinstance(data, dict) else None
        if not isinstance(items, list):
            return []
        exclude = set(exclude_ids)
        out: List[CandidateMovie] = []
        for item in items:
            if not isinstance(item, dict) or "id" not in item:
                continue
            try:
                movie = CandidateMovie(
                    id=int(item["id"]),
                    title=str(item.get("title") or ""),
                    year=item.get("year"),
                    genres=item.get("genres") or [],
                    runtime=item.get("runtime"),
                    rating=item.get("rating"),
                    poster_url=item.get("poster_url"),
                    overview=item.get("overview") or "",
                )
            except Exception:
                continue
            if movie.id in exclude or not movie.title:
                continue
            out.append(movie)
        return out

    def _card_from_candidate(
        self,
        src: CandidateMovie,
        *,
        why: str = "",
        vibe_tags: Optional[List[str]] = None,
        caution: Optional[str] = None,
        overview_safe: str = "",
        runtime: Optional[int] = None,
        poster_url: Optional[str] = None,
    ) -> MovieCard:
        """候选 → MovieCard；缺海报时按 id 拉 detail 回填（仅最终 3~5 部）。"""
        poster = src.poster_url or poster_url
        title = src.title
        year = src.year
        genres = list(src.genres or [])
        rating = src.rating
        overview = overview_safe or (src.overview or "")[:200]
        rt = runtime if runtime is not None else src.runtime

        if not poster:
            try:
                detail = get_movie_service().get_detail(src.id)
                poster = detail.poster_url
                title = title or detail.title
                year = year if year is not None else detail.year
                genres = genres or list(detail.genres or [])
                rating = rating if rating is not None else detail.rating
                if not overview_safe and detail.overview:
                    overview = detail.overview[:200]
                if rt is None:
                    rt = detail.runtime
            except Exception:
                logger.warning("MovieCard 海报回填失败 id=%s", src.id)

        return MovieCard(
            id=src.id,
            title=title,
            year=year,
            genres=genres,
            runtime=rt,
            rating=rating,
            poster_url=poster,
            why=why,
            vibe_tags=vibe_tags or [],
            caution=caution,
            overview_safe=overview,
        )

    def _enforce_candidate_ids(
        self,
        result: RecommendResult,
        candidates: List[CandidateMovie],
        profile: TasteProfile,
    ) -> RecommendResult:
        """白名单闸：丢弃候选外 id；元数据以 TMDB 候选为准；不足 3 部则补齐并降级。"""
        allowed = {c.id: c for c in candidates}
        kept: List[MovieCard] = []
        for card in result.movies:
            if card.id not in allowed:
                continue
            src = allowed[card.id]
            kept.append(
                self._card_from_candidate(
                    src,
                    why=card.why,
                    vibe_tags=card.vibe_tags,
                    caution=card.caution,
                    overview_safe=card.overview_safe or (src.overview or "")[:200],
                    runtime=card.runtime if card.runtime is not None else src.runtime,
                    poster_url=card.poster_url,
                )
            )
        if 3 <= len(kept) <= 5:
            result.movies = kept
            return result

        # 合法片不足 3 部：按评分从候选补齐，并标记降级
        result.is_fallback = True
        have = {m.id for m in kept}
        ranked = sorted(
            candidates,
            key=lambda m: (m.rating is not None, m.rating or 0),
            reverse=True,
        )
        for c in ranked:
            if c.id in have:
                continue
            kept.append(
                self._card_from_candidate(
                    c,
                    why="系统按候选热度补齐",
                    overview_safe=(c.overview or "")[:200],
                )
            )
            if len(kept) >= 3:
                break
        result.movies = kept[:5]
        if not result.profile_summary and profile:
            result.profile_summary = profile.summary
        if not result.playlist_name:
            result.playlist_name = "今日候选速选"
        return result

    def _fallback_result(
        self,
        request: RecommendRequest,
        profile: Optional[TasteProfile],
        candidates: List[CandidateMovie],
        reason: str,
    ) -> RecommendResult:
        """诚实降级：尽量用真片凑片单，强制 is_fallback=True。"""
        if not candidates:
            try:
                candidates = self._discover_by_profile(
                    request,
                    profile
                    or TasteProfile(
                        summary=reason,
                        genre_hints=list(request.genres),
                    ),
                )
            except Exception:
                candidates = []

        movies: List[MovieCard] = []
        for c in candidates[:5]:
            movies.append(
                self._card_from_candidate(
                    c,
                    why=f"降级推荐（{reason}）",
                    overview_safe=(c.overview or "")[:200],
                )
            )
        return RecommendResult(
            playlist_name="降级片单",
            profile_summary=(profile.summary if profile else reason),
            movies=movies,
            is_fallback=True,
            taste_profile=profile,
        )

    @staticmethod
    def _extract_json(text: str) -> Optional[dict]:
        """从模型文本提取 JSON 对象（纯 JSON / 代码块 / 夹杂说明均可）。"""
        if not text:
            return None
        text = text.strip()
        try:
            data = json.loads(text)
            return data if isinstance(data, dict) else None
        except json.JSONDecodeError:
            pass
        fence = re.search(r"```(?:json)?\s*(\{.*?\})\s*```", text, re.DOTALL)
        if fence:
            try:
                data = json.loads(fence.group(1))
                return data if isinstance(data, dict) else None
            except json.JSONDecodeError:
                pass
        start, end = text.find("{"), text.rfind("}")
        if start >= 0 and end > start:
            try:
                data = json.loads(text[start : end + 1])
                return data if isinstance(data, dict) else None
            except json.JSONDecodeError:
                return None
        return None

    def health_snapshot(self) -> dict[str, Any]:
        """返回各 Agent 名称与工具数量（供 /api/recommend/health）。"""
        return {
            "agents": [
                {"name": self.profile_agent.name, "tools_count": 0},
                {
                    "name": self.search_agent.name,
                    "tools_count": len(self.search_agent.list_tools()),
                },
                {"name": self.recommend_agent.name, "tools_count": 0},
            ]
        }


_recommender: Optional[MultiAgentMovieRecommender] = None


def get_movie_recommender() -> MultiAgentMovieRecommender:
    """获取进程内编排器单例（懒加载，避免重复初始化 LLM/Agent）。"""
    global _recommender
    if _recommender is None:
        _recommender = MultiAgentMovieRecommender()
    return _recommender