forked from zhouruizhe/gmTouringMiniApp
92 lines
3.1 KiB
Python
92 lines
3.1 KiB
Python
from __future__ import annotations
|
|
|
|
import uuid
|
|
|
|
from .llm import LLMError, LLMService
|
|
from .schemas import (
|
|
ModelRecommendationSet,
|
|
PlanRecommendationRequest,
|
|
PlanningMode,
|
|
RecommendationItem,
|
|
RecommendationResponse,
|
|
)
|
|
|
|
|
|
class RecommendationService:
|
|
def __init__(self, llm: LLMService) -> None:
|
|
self.llm = llm
|
|
|
|
async def create(
|
|
self, request: PlanRecommendationRequest
|
|
) -> RecommendationResponse:
|
|
generated = await self.llm.recommend(request)
|
|
recommendations = self._canonicalize(request, generated)
|
|
message = (
|
|
"已保留全部自选点位并给出建议顺序；最终路线与时长由客户端核算。"
|
|
if request.mode == PlanningMode.CUSTOM
|
|
else "已从当前 POI 白名单给出推荐；最终路线与时长由客户端核算。"
|
|
)
|
|
return RecommendationResponse(
|
|
request_id=str(uuid.uuid4()),
|
|
mode=request.mode,
|
|
assistant_message=message,
|
|
recommendations=recommendations,
|
|
generation_mode=self.llm.mode,
|
|
)
|
|
|
|
@staticmethod
|
|
def _canonicalize(
|
|
request: PlanRecommendationRequest,
|
|
generated: ModelRecommendationSet,
|
|
) -> list[RecommendationItem]:
|
|
allowed_ids = {candidate.id for candidate in request.candidates}
|
|
generated_by_id = {}
|
|
for item in generated.recommendations:
|
|
if item.poi_id not in allowed_ids:
|
|
raise LLMError("模型返回了 POI 白名单外的 ID")
|
|
if item.poi_id in generated_by_id:
|
|
raise LLMError("模型返回了重复 POI ID")
|
|
generated_by_id[item.poi_id] = item
|
|
|
|
if request.mode == PlanningMode.CUSTOM:
|
|
selected_ids = set(request.selected_poi_ids)
|
|
ordered_ids = [
|
|
item.poi_id
|
|
for item in sorted(
|
|
generated.recommendations,
|
|
key=lambda item: item.order or 9,
|
|
)
|
|
if item.poi_id in selected_ids
|
|
]
|
|
# Selection belongs to the user, not the model. Keep all selected IDs
|
|
# even if a real model omits one, and discard unselected suggestions.
|
|
ordered_ids.extend(
|
|
poi_id
|
|
for poi_id in request.selected_poi_ids
|
|
if poi_id not in ordered_ids
|
|
)
|
|
else:
|
|
ordered_ids = [
|
|
item.poi_id
|
|
for item in sorted(
|
|
generated.recommendations,
|
|
key=lambda item: item.order or 9,
|
|
)
|
|
][:8]
|
|
|
|
if not ordered_ids:
|
|
raise LLMError("模型未返回可用的 POI 推荐")
|
|
return [
|
|
RecommendationItem(
|
|
poi_id=poi_id,
|
|
reason=(
|
|
generated_by_id[poi_id].reason
|
|
if poi_id in generated_by_id
|
|
else "这是你选择的点位，已由系统保留。"
|
|
),
|
|
order=index,
|
|
)
|
|
for index, poi_id in enumerate(ordered_ids, start=1)
|
|
]
|
|
|