hello-agents/Co-creation-projects/Shawnxyxy-HealthRecordAgent/backend/service/diet_pipeline.py

"""
阶段 2：Nutritionist → Coach → Habit 三 Agent 串行流水线；
每阶段 LLM 输出经 Pydantic 校验，失败自动重试；统一错误码与降级。
"""

from __future__ import annotations

import asyncio
import json
import logging
import re
import uuid
from typing import Any, Dict, List, Optional, Tuple, Type

from pydantic import BaseModel, ValidationError

from core.llm_adapter import get_llm_adapter
from memory.store import format_reflect_memory_for_prompt, save_diet_run
from rag.indexers import index_diet_run
from rag.retriever import retrieve
from service.diet_errors import DietErrorCode, diet_error_record
from service.diet_schemas import (
    SCHEMA_VERSION,
    CoachOutput,
    FoodParseOutput,
    HabitOutput,
    MealPlan,
    MealPlanItem,
    NutritionistOutput,
    NutritionSummary,
)
from tools.diet_tools import dispatch_tool

logger = logging.getLogger(__name__)

DIET_STAGE_TIMEOUT_SEC = 95.0
MAX_STAGE_ATTEMPTS = 2


def _extract_json_object(text: str) -> Optional[Dict[str, Any]]:
    if not text:
        return None
    t = text.strip()
    m = re.search(r"```(?:json)?\s*([\s\S]*?)```", t)
    if m:
        t = m.group(1).strip()
    try:
        return json.loads(t)
    except json.JSONDecodeError:
        i = t.find("{")
        j = t.rfind("}")
        if i >= 0 and j > i:
            try:
                return json.loads(t[i : j + 1])
            except json.JSONDecodeError:
                return None
    return None


def _goal_target_protein(context: Dict[str, Any]) -> float:
    goal = str(context.get("goal") or "maintain")
    if goal == "muscle_gain":
        return 130.0
    if goal == "fat_loss":
        return 95.0
    return 105.0


def _fallback_food_parse(context: Dict[str, Any]) -> FoodParseOutput:
    raw = str(context.get("today_food_log_text") or "")
    pieces = [p.strip(" ，。;；\n\t") for p in re.split(r"[，,;；。]\s*", raw) if p.strip()]
    items = []
    for p in pieces[:10]:
        items.append(
            {
                "meal_time": "",
                "food_name": p[:40],
                "portion_text": "未明确",
                "confidence": 0.45,
            }
        )
    return FoodParseOutput(
        items=items,
        nutrition_summary=NutritionSummary(),
        parse_notes="（降级）食物解析阶段失败，已按日志片段做粗略拆分；营养值未估算。",
    )


def _fallback_nutritionist(context: Dict[str, Any], nutrition_summary: NutritionSummary) -> NutritionistOutput:
    tgt = _goal_target_protein(context)
    cur = float(nutrition_summary.protein_g or 0)
    gap = max(0.0, tgt - cur)
    return NutritionistOutput(
        protein_gap_g=gap,
        rationale="（降级）根据目标与日志解析结果估算蛋白缺口；LLM 阶段未通过校验或超时。",
        suggested_lookup_queries=["鸡蛋,希腊酸奶,牛奶,豆浆,即食鸡胸肉"],
        candidate_focus=["便利店高蛋白", "训练后补充"],
    )


def _fallback_coach(context: Dict[str, Any]) -> CoachOutput:
    activity_text = str(context.get("activity_context") or "")
    train = any(k in activity_text for k in ["训练", "力量", "健身", "workout", "training"])
    return CoachOutput(
        training_recovery_note="（降级）晚间安排力量训练时需优先补充蛋白与适量碳水；具体强度以当日体感为准。"
        if train
        else "（降级）非训练日仍以均衡蛋白为主，避免睡前过饱。",
        timing_constraints="训练后 1～2 小时内尽量安排一餐；便利店即食优先选成分表蛋白较高的品类。"
        if train
        else "晚餐时间尽量规律，避免过晚大量进食。",
        energy_note="",
        coach_constraints_for_menu=["少油炸", "避免单次过量乳糖不耐受品类"],
    )


def _fallback_habit(
    context: Dict[str, Any], reflect_mem: str, nutrition_summary: NutritionSummary
) -> HabitOutput:
    tgt = _goal_target_protein(context)
    cur = float(nutrition_summary.protein_g or 0)
    gap = max(25.0, min(80.0, max(0.0, tgt - cur)))
    return HabitOutput(
        reflect_alignment="（降级）未能生成完整习惯层输出；已忽略部分 Reflect 细节，仅做安全兜底推荐。"
        + (" 近期有用户反馈记录，建议下次缩短决策链或检查模型输出格式。" if "暂无" not in reflect_mem else ""),
        execution_hints=["优先买得到、可立即食用的组合", "若仍失败请改选外卖蛋白套餐"],
        meal_plan=MealPlan(
            items=[
                MealPlanItem(
                    name="希腊酸奶",
                    portion="约 150g×1 杯",
                    est_protein_g=min(18.0, gap * 0.35),
                    why="便利店常见，蛋白密度较高",
                ),
                MealPlanItem(
                    name="水煮蛋",
                    portion="2 个",
                    est_protein_g=12.0,
                    why="易购买、蛋白稳定",
                ),
                MealPlanItem(
                    name="豆浆",
                    portion="300ml",
                    est_protein_g=min(12.0, gap * 0.2),
                    why="补充液体蛋白与水分",
                ),
            ],
            total_est_protein_g=round(min(gap, 45.0), 1),
            tips=["此为 schema/LLM 失败时的安全兜底菜单，建议重试或检查 API。"],
        ),
    )


async def _run_validated_stage(
    llm: Any,
    stage: str,
    prompt: str,
    model_cls: Type[BaseModel],
    errors: List[Dict[str, Any]],
    timeout_sec: float = DIET_STAGE_TIMEOUT_SEC,
) -> Tuple[Optional[BaseModel], List[Dict[str, Any]]]:
    attempts: List[Dict[str, Any]] = []
    repair_hint = ""
    for attempt in range(MAX_STAGE_ATTEMPTS):
        full_prompt = prompt
        if repair_hint:
            full_prompt += (
                "\n\n【修正要求】上一输出未通过 schema 校验或无法解析：\n"
                f"{repair_hint}\n请只输出 **一个** JSON 对象，字段齐全、类型正确，不要 Markdown。"
            )
        try:
            raw = await asyncio.wait_for(llm.ainvoke(full_prompt), timeout=timeout_sec)
        except asyncio.TimeoutError:
            errors.append(
                diet_error_record(
                    stage,
                    DietErrorCode.LLM_TIMEOUT,
                    "LLM 调用超时",
                    attempt=attempt,
                )
            )
            attempts.append(
                {"attempt": attempt, "ok": False, "error_code": DietErrorCode.LLM_TIMEOUT.value}
            )
            repair_hint = "上次超时；请输出更紧凑的 JSON，保留所有必填字段。"
            continue
        except Exception as e:
            # 上游模型网关 5xx / SDK 异常都归一为阶段中止错误，避免接口直接 500。
            errors.append(
                diet_error_record(
                    stage,
                    DietErrorCode.STAGE_ABORTED,
                    f"LLM 调用异常: {type(e).__name__}",
                    attempt=attempt,
                    detail=str(e)[:1200],
                )
            )
            attempts.append(
                {
                    "attempt": attempt,
                    "ok": False,
                    "error_code": DietErrorCode.STAGE_ABORTED.value,
                    "exception": type(e).__name__,
                }
            )
            repair_hint = "上轮调用失败，请仅输出合法 JSON。"
            continue

        obj = _extract_json_object(raw)
        if obj is None:
            errors.append(
                diet_error_record(
                    stage,
                    DietErrorCode.LLM_PARSE_ERROR,
                    "无法从模型输出解析 JSON",
                    attempt=attempt,
                    detail=(raw[:1200] if raw else ""),
                )
            )
            attempts.append(
                {
                    "attempt": attempt,
                    "ok": False,
                    "error_code": DietErrorCode.LLM_PARSE_ERROR.value,
                    "llm_preview": (raw[:1500] if raw else ""),
                }
            )
            repair_hint = "模型输出不是合法 JSON；请严格输出 JSON only。"
            continue

        try:
            validated = model_cls.model_validate(obj)
            attempts.append(
                {
                    "attempt": attempt,
                    "ok": True,
                    "error_code": None,
                    "llm_preview": raw[:2500] if raw else "",
                }
            )
            return validated, attempts
        except ValidationError as ve:
            err_text = ve.json()[:2000]
            errors.append(
                diet_error_record(
                    stage,
                    DietErrorCode.VALIDATION_FAILED,
                    "Pydantic 校验失败",
                    attempt=attempt,
                    detail=err_text,
                )
            )
            attempts.append(
                {
                    "attempt": attempt,
                    "ok": False,
                    "error_code": DietErrorCode.VALIDATION_FAILED.value,
                    "validation_detail": err_text,
                    "parsed_shape": {k: type(v).__name__ for k, v in obj.items()}
                    if isinstance(obj, dict)
                    else None,
                }
            )
            repair_hint = err_text

    return None, attempts


def _prefetch_tools(user_id: str, context: Dict[str, Any]) -> Tuple[Dict[str, Any], List[Dict[str, Any]]]:
    trace_tools: List[Dict[str, Any]] = []
    activity: Dict[str, Any] = {}
    nutrition: Dict[str, Any] = {}
    try:
        activity = dispatch_tool(
            "activity_sleep_summary", {"user_id": user_id}, user_id
        )
    except Exception as e:
        trace_tools.append(
            {
                "tool": "activity_sleep_summary",
                "ok": False,
                "error": str(e),
            }
        )
    else:
        trace_tools.append({"tool": "activity_sleep_summary", "ok": True, "result": activity})

    default_q = "鸡蛋,希腊酸奶,牛奶,豆浆,即食鸡胸肉"
    try:
        nutrition = dispatch_tool(
            "nutrition_lookup",
            {"query": context.get("nutrition_prefetch_query") or default_q},
            user_id,
        )
    except Exception as e:
        trace_tools.append({"tool": "nutrition_lookup", "ok": False, "error": str(e)})
    else:
        trace_tools.append({"tool": "nutrition_lookup", "ok": True, "result": nutrition})

    return {"activity": activity, "nutrition": nutrition}, trace_tools


class DietMultiAgentPipeline:
    def __init__(self) -> None:
        self.llm = get_llm_adapter()

    async def run(
        self,
        user_id: str,
        context: Dict[str, Any],
        *,
        replayed_from_run_id: Optional[str] = None,
    ) -> Dict[str, Any]:
        run_id = str(uuid.uuid4())
        reflect_mem = format_reflect_memory_for_prompt(user_id, limit=8)
        errors: List[Dict[str, Any]] = []
        pipeline_trace: List[Dict[str, Any]] = []
        rag_result = await asyncio.to_thread(
            retrieve,
            user_id,
            {
                "scenario": "diet_recommendation",
                "goal": context.get("goal"),
                "free_notes": context.get("free_notes", ""),
                "today_food_log_text": str(context.get("today_food_log_text") or "")[:600],
                "query": "训练后蛋白补齐与执行阻碍规避",
            },
        )
        rag_summary = rag_result.get("summary", "（暂无召回记忆）")
        pipeline_trace.append({"phase": "rag_retrieve", "debug": rag_result.get("debug", {})})

        tool_bundle, tool_trace = _prefetch_tools(user_id, context)
        pipeline_trace.append({"phase": "tool_prefetch", "tools": tool_trace})

        degraded = False

        # ----- Food Parse (LLM) -----
        fp_prompt = f"""你是食物日志解析 Agent。请把用户自然语言饮食记录解析为 JSON。只输出一个 JSON 对象，不要 Markdown。

结构：
{{
  "items": [
    {{
      "meal_time": string,      // breakfast/lunch/dinner/snack 或空字符串
      "food_name": string,
      "portion_text": string,
      "confidence": number      // 0~1
    }}
  ],
  "nutrition_summary": {{
    "protein_g": number,
    "carb_g": number,
    "fat_g": number,
    "fiber_g": number,
    "sodium_mg": number,
    "calories_kcal": number
  }},
  "parse_notes": string
}}

要求：
- 从 today_food_log_text 中尽可能提取食物与份量；没有明确份量可写“未明确”。
- nutrition_summary 给出粗略估计值；无法判断可填 0。
- 字段齐全且类型正确。

用户场景：
{json.dumps(context, ensure_ascii=False, indent=2)}
"""
        fp, fp_attempts = await _run_validated_stage(
            self.llm, "food_parse", fp_prompt, FoodParseOutput, errors
        )
        fp_fb = False
        if fp is None:
            fp = _fallback_food_parse(context)
            fp_fb = True
            degraded = True
            errors.append(
                diet_error_record(
                    "food_parse",
                    DietErrorCode.DEGRADED_FALLBACK,
                    "食物解析阶段失败，已使用规则降级输出",
                )
            )
        pipeline_trace.append(
            {
                "phase": "food_parse",
                "fallback_used": fp_fb,
                "attempts": fp_attempts,
                "output": fp.model_dump(),
            }
        )

        # ----- Nutritionist -----
        n_prompt = f"""你是 **Nutritionist（营养师）Agent**。只输出 **一个 JSON**，不要其它文字。

字段与类型必须完全一致：
{{
  "protein_gap_g": number,
  "rationale": string,
  "suggested_lookup_queries": string[],
  "candidate_focus": string[]
}}

用户场景：
{json.dumps(context, ensure_ascii=False, indent=2)}

食物解析结果（LLM）：
{json.dumps(fp.model_dump(), ensure_ascii=False, indent=2)}

Reflect 记忆（调整推荐策略）：
{reflect_mem}

历史记忆召回（RAG）：
{rag_summary}

Mock 营养表检索结果（供参考）：
{json.dumps(tool_bundle.get("nutrition", {}), ensure_ascii=False, indent=2)}
"""
        nu, nu_attempts = await _run_validated_stage(
            self.llm, "nutritionist", n_prompt, NutritionistOutput, errors
        )
        nu_fb = False
        if nu is None:
            nu = _fallback_nutritionist(context, fp.nutrition_summary)
            nu_fb = True
            degraded = True
            errors.append(
                diet_error_record(
                    "nutritionist",
                    DietErrorCode.DEGRADED_FALLBACK,
                    "营养师阶段失败，已使用规则降级输出",
                )
            )
        pipeline_trace.append(
            {
                "phase": "nutritionist",
                "fallback_used": nu_fb,
                "attempts": nu_attempts,
                "output": nu.model_dump(),
            }
        )

        # 按营养师建议追加一次营养查询（可选）
        extra_nutrition: Dict[str, Any] = {}
        if nu.suggested_lookup_queries:
            q = ",".join(nu.suggested_lookup_queries[:3])
            try:
                extra_nutrition = dispatch_tool(
                    "nutrition_lookup", {"query": q[:200]}, user_id
                )
            except Exception as e:
                errors.append(
                    diet_error_record(
                        "tool",
                        DietErrorCode.TOOL_ERROR,
                        f"nutrition_lookup 追加查询失败: {e}",
                    )
                )
                extra_nutrition = {"error": str(e)}
        tool_bundle["nutrition_extra"] = extra_nutrition

        # ----- Coach -----
        c_prompt = f"""你是 **Coach（运动恢复）Agent**。只输出 **一个 JSON**。

结构：
{{
  "training_recovery_note": string,
  "timing_constraints": string,
  "energy_note": string,
  "coach_constraints_for_menu": string[]
}}

用户场景：
{json.dumps(context, ensure_ascii=False, indent=2)}

食物解析（营养汇总）：
{json.dumps(fp.nutrition_summary.model_dump(), ensure_ascii=False, indent=2)}

营养师结论：
{json.dumps(nu.model_dump(), ensure_ascii=False, indent=2)}

活动/睡眠摘要：
{json.dumps(tool_bundle.get("activity", {}), ensure_ascii=False, indent=2)}

历史记忆召回（RAG）：
{rag_summary}
"""
        co, co_attempts = await _run_validated_stage(
            self.llm, "coach", c_prompt, CoachOutput, errors
        )
        co_fb = False
        if co is None:
            co = _fallback_coach(context)
            co_fb = True
            degraded = True
            errors.append(
                diet_error_record(
                    "coach",
                    DietErrorCode.DEGRADED_FALLBACK,
                    "Coach 阶段失败，已使用模板降级输出",
                )
            )
        pipeline_trace.append(
            {
                "phase": "coach",
                "fallback_used": co_fb,
                "attempts": co_attempts,
                "output": co.model_dump(),
            }
        )

        # ----- Habit -----
        h_prompt = f"""你是 **Habit（习惯养成）Agent**。只输出 **一个 JSON**。

结构：
{{
  "reflect_alignment": string,
  "execution_hints": string[],
  "meal_plan": {{
    "items": [{{ "name": string, "portion": string, "est_protein_g": number, "why": string }}],
    "total_est_protein_g": number,
    "tips": string[]
  }}
}}

要求：
- meal_plan.items 至少 1 条；份量具体、可执行；适合便利店/外卖。
- 结合 Reflect 记忆，说明本次如何规避上次失败原因。
- est_protein_g 为粗略估计。

用户场景：
{json.dumps(context, ensure_ascii=False, indent=2)}

食物解析结果：
{json.dumps(fp.model_dump(), ensure_ascii=False, indent=2)}

Reflect 记忆：
{reflect_mem}

历史记忆召回（RAG）：
{rag_summary}

营养师：
{json.dumps(nu.model_dump(), ensure_ascii=False, indent=2)}

Coach：
{json.dumps(co.model_dump(), ensure_ascii=False, indent=2)}

营养数据（含追加查询）：
{json.dumps(tool_bundle, ensure_ascii=False, indent=2)[:12000]}
"""
        ha, ha_attempts = await _run_validated_stage(
            self.llm, "habit", h_prompt, HabitOutput, errors
        )
        ha_fb = False
        if ha is None:
            ha = _fallback_habit(context, reflect_mem, fp.nutrition_summary)
            ha_fb = True
            degraded = True
            errors.append(
                diet_error_record(
                    "habit",
                    DietErrorCode.DEGRADED_FALLBACK,
                    "Habit 阶段失败，已使用安全兜底菜单",
                )
            )
        pipeline_trace.append(
            {
                "phase": "habit",
                "fallback_used": ha_fb,
                "attempts": ha_attempts,
                "output": ha.model_dump(),
            }
        )

        meal_plan = ha.meal_plan.model_dump()

        planning = {
            "reasoning": nu.rationale,
            "plan_steps": [
                "FoodParse：从饮食日志抽取食物与份量并估算营养",
                f"Nutritionist：缺口约 {nu.protein_gap_g:.1f}g 蛋白",
                "Coach：训练/进食窗口与恢复约束",
                "Habit：对齐 Reflect 的可执行菜单",
            ],
            "agent_pipeline": [
                "FoodParseAgent",
                "NutritionistAgent",
                "CoachAgent",
                "HabitAgent",
            ],
        }

        output: Dict[str, Any] = {
            "run_id": run_id,
            "user_id": user_id,
            "schema_version": SCHEMA_VERSION,
            "pipeline_mode": "multi_agent",
            "replayed_from": replayed_from_run_id,
            "degraded": degraded,
            "errors": errors,
            "planning": planning,
            "stages": {
                "nutritionist": {
                    "ok": not nu_fb,
                    "fallback_used": nu_fb,
                    "output": nu.model_dump(),
                },
                "coach": {
                    "ok": not co_fb,
                    "fallback_used": co_fb,
                    "output": co.model_dump(),
                },
                "habit": {
                    "ok": not ha_fb,
                    "fallback_used": ha_fb,
                    "output": ha.model_dump(),
                },
            },
            "meal_plan": meal_plan,
            "food_parse": fp.model_dump(),
            "nutrition_summary": fp.nutrition_summary.model_dump(),
            "habit_extras": {
                "reflect_alignment": ha.reflect_alignment,
                "execution_hints": ha.execution_hints,
            },
            "react_trace": pipeline_trace,
            "reflect_memory_used": reflect_mem,
            "retrieved_memory": rag_summary,
            "rag_debug": rag_result.get("debug", {}),
        }

        try:
            save_diet_run(
                user_id=user_id,
                run_id=run_id,
                input_payload=context,
                steps_trace=pipeline_trace,
                output_payload=output,
                replayed_from_run_id=replayed_from_run_id,
            )
        except Exception as e:
            logger.exception("diet_runs 落库失败: %s", e)
        try:
            # 最佳努力索引，不影响主流程
            await asyncio.to_thread(index_diet_run, run_id)
        except Exception as e:
            logger.warning("diet run 向量索引失败（不影响返回）: %s", e)

        return output