import json import re from typing import Any, Dict, Iterable, List, Mapping, Optional from fastapi import APIRouter, Depends from fastapi.responses import StreamingResponse from sqlalchemy.orm import Session from app.api.reviews import require_current_user from app.clients.shared_ai_platform import SharedAIPlatformClient, get_shared_ai_platform_client from app.db import get_db_session from app.schemas.chat import ChatStreamRequest from app.services.legal_article_service import LegalArticleService from app.services.legal_domain_registry import BUSINESS_AUXILIARY_PROMPT_KEYS from app.services.prompt_runtime_service import PromptRuntimeService from app.services.review_service import ReviewService from app.services.tenant_service import TenantService router = APIRouter(prefix="/chat", tags=["chat"]) ROUTE_PERMISSION_SCOPE = "chat" ROUTE_PERMISSION_ACTIONS = { "stream": "chat:stream", } def get_chat_review_service(db: Session = Depends(get_db_session)) -> ReviewService: return ReviewService(db) def get_chat_legal_article_service(db: Session = Depends(get_db_session)) -> LegalArticleService: return LegalArticleService(db) def get_chat_prompt_runtime_service(db: Session = Depends(get_db_session)) -> PromptRuntimeService: return PromptRuntimeService(db) def get_tenant_service(db: Session = Depends(get_db_session)) -> TenantService: return TenantService(db) @router.post("/stream") def stream_chat( payload: ChatStreamRequest, shared_ai_platform_client: SharedAIPlatformClient = Depends(get_shared_ai_platform_client), tenant_service: TenantService = Depends(get_tenant_service), current_user: dict = Depends(require_current_user), review_service: Optional[ReviewService] = Depends(get_chat_review_service), legal_article_service: Optional[LegalArticleService] = Depends(get_chat_legal_article_service), prompt_runtime_service: Optional[PromptRuntimeService] = Depends(get_chat_prompt_runtime_service), ): tenant_service.assert_billing_allows(current_user, "chat") review_context = ( payload.review_context.model_dump(mode="json", by_alias=True, exclude_none=True) if payload.review_context else None ) review_context = _enrich_review_context_with_legal_articles(current_user, review_context, review_service) prompt_runtime = _resolve_chat_prompt_runtime(current_user, review_context, prompt_runtime_service, payload.message) runtime_legal_sources = _search_runtime_legal_article_sources( current_user, payload.message, legal_article_service, prompt_runtime=prompt_runtime, ) review_context = _enrich_review_context_with_runtime_legal_sources(review_context, runtime_legal_sources) review_context = _enrich_review_context_with_prompt_runtime(review_context, prompt_runtime, prompt_runtime_service) template_key = prompt_runtime.get("template_key") if prompt_runtime and prompt_runtime.get("available") else None if prompt_runtime_service is not None: prompt_runtime_service.record_run_link( current_user, prompt_runtime, run_type=str((prompt_runtime or {}).get("prompt_type") or "legal_chat"), run_id=payload.conversation_id or "new_conversation", resource_type="conversation", resource_id=payload.conversation_id, trace={ "message_length": len(payload.message or ""), "has_review_context": bool(review_context), "review_id": (review_context or {}).get("reviewId") or (review_context or {}).get("review_id"), "template_key": template_key, }, ) local_legal_sources = _dedupe_sources([*_build_legal_article_sources(review_context), *runtime_legal_sources]) return StreamingResponse( _merge_local_sources_into_stream( shared_ai_platform_client.stream_answer( current_user, payload.conversation_id, payload.message, review_context, template_key=template_key, ), local_legal_sources, payload.message, ), media_type="text/event-stream", ) def _resolve_chat_prompt_runtime( current_user: Dict[str, Any], review_context: Optional[Dict[str, Any]], prompt_runtime_service: Optional[PromptRuntimeService], message: str = "", ) -> Dict[str, Any]: if prompt_runtime_service is None: return {"available": False, "prompt_type": "legal_chat", "prompt_version": "legal_chat.shared_ai_stream.v1", "source": "local_default"} has_review_context = bool(review_context and ((review_context.get("reviewId") or review_context.get("review_id")))) business_auxiliary_key = _detect_business_auxiliary_prompt_key(message, review_context) if business_auxiliary_key: return prompt_runtime_service.resolve_published_prompt( current_user, "business_auxiliary", preferred_keys=[ business_auxiliary_key, ], fallback_prompt_version="business_auxiliary.shared_ai_stream.v1", ) return prompt_runtime_service.resolve_published_prompt( current_user, "legal_chat", preferred_keys=[ "legal_chat_review" if has_review_context else None, "legal_chat_default", ], fallback_prompt_version="legal_chat.shared_ai_stream.v1", ) def _detect_business_auxiliary_prompt_key( message: str, review_context: Optional[Dict[str, Any]], ) -> Optional[str]: context_text = _build_review_context_action_text(review_context) if context_text: routed_key = _detect_business_auxiliary_prompt_key_from_text(context_text) if routed_key: return routed_key return _detect_business_auxiliary_prompt_key_from_text(_build_prompt_intent_text(message, review_context)) def _detect_business_auxiliary_prompt_key_from_text(text: str) -> Optional[str]: if _contains_any(text, ["法律意见书强边界", "法律意见书兜底", "法律意见书边界", "使用限制", "前提假设", "不能出具", "正式法律意见", "legal opinion boundary", "legal opinion fallback"]): return BUSINESS_AUXILIARY_PROMPT_KEYS["legal_opinion_boundary_fallback"] if _contains_any(text, ["履约义务", "履行义务", "义务提取", "履约清单", "义务清单", "交付义务", "配合义务", "双方要做什么", "对方需要做什么", "obligation extraction", "performance obligation", "performance checklist"]): return BUSINESS_AUXILIARY_PROMPT_KEYS["obligation_extraction"] if _contains_any(text, ["工商", "备案材料", "变更登记", "登记材料", "材料清单", "提交材料", "申报材料", "登记机关", "filing materials", "registration materials", "registration checklist", "filing checklist"]): return BUSINESS_AUXILIARY_PROMPT_KEYS["filing_materials_checklist"] if _contains_any(text, ["谈判策略", "谈判建议", "怎么谈", "如何谈", "议价", "让步方案", "让步空间", "底线", "沟通话术", "对方不接受", "替代方案", "negotiate", "negotiation strategy", "negotiation plan", "fallback position"]): return BUSINESS_AUXILIARY_PROMPT_KEYS["negotiation_strategy"] if _contains_any(text, ["法律意见书", "legal opinion"]) and _contains_any( text, ["边界", "兜底", "免责声明", "不能替代", "人工复核", "使用限制", "前提假设", "boundary", "fallback", "disclaimer", "human review", "assumption", "limitation"], ): return BUSINESS_AUXILIARY_PROMPT_KEYS["legal_opinion_boundary_fallback"] return None def _build_review_context_action_text(review_context: Optional[Dict[str, Any]]) -> str: if not isinstance(review_context, dict): return "" metadata = review_context.get("metadata") if isinstance(review_context.get("metadata"), Mapping) else {} parts = [ metadata.get("actionLabel"), metadata.get("focusLabel"), metadata.get("reviewOpinion"), review_context.get("actionLabel"), review_context.get("focusLabel"), review_context.get("prompt"), ] return "\n".join(str(item) for item in parts if item).lower() def _build_prompt_intent_text(message: str, review_context: Optional[Dict[str, Any]]) -> str: parts = [message or ""] if isinstance(review_context, dict): parts.append(json.dumps(review_context, ensure_ascii=False, default=str)) return "\n".join(parts).lower() def _contains_any(value: str, candidates: Iterable[str]) -> bool: return any(candidate.lower() in value for candidate in candidates) def _enrich_review_context_with_prompt_runtime( review_context: Optional[Dict[str, Any]], prompt_runtime: Optional[Dict[str, Any]], prompt_runtime_service: Optional[PromptRuntimeService], ) -> Optional[Dict[str, Any]]: if prompt_runtime_service is None or not prompt_runtime: return review_context enriched = dict(review_context or {}) prompt_trace = prompt_runtime_service.to_trace(prompt_runtime) enriched["promptRuntime"] = prompt_trace enriched["prompt_runtime"] = prompt_trace metadata = dict(enriched.get("metadata") or {}) metadata["promptRuntime"] = prompt_trace metadata["promptVersion"] = prompt_trace.get("prompt_version") metadata["promptKey"] = prompt_trace.get("prompt_key") enriched["metadata"] = metadata return enriched def _enrich_review_context_with_legal_articles( current_user: Dict[str, Any], review_context: Optional[Dict[str, Any]], review_service: Optional[ReviewService], ) -> Optional[Dict[str, Any]]: if not review_context or review_service is None: return review_context review_id = review_context.get("reviewId") or review_context.get("review_id") if not review_id: return review_context try: result = review_service.get_review_result(current_user, str(review_id)) except Exception: return review_context if not result: return review_context refs = _collect_legal_article_refs(result.get("risk_items") or []) if not refs: return review_context enriched = dict(review_context) enriched["legalArticleRefs"] = refs enriched["legal_article_refs"] = refs metadata = dict(enriched.get("metadata") or {}) metadata["legalArticleRefCount"] = len(refs) metadata["legalArticleSources"] = [ { "id": ref.get("id"), "title": _build_legal_article_source_title(ref), "detail_path": ref.get("detail_path"), } for ref in refs ] enriched["metadata"] = metadata return enriched def _search_runtime_legal_article_sources( current_user: Dict[str, Any], message: str, legal_article_service: Optional[LegalArticleService], prompt_runtime: Optional[Dict[str, Any]] = None, ) -> List[Dict[str, Any]]: if legal_article_service is None: return [] normalized_message = str(message or "").strip() if not normalized_message: return [] if str((prompt_runtime or {}).get("prompt_type") or "").strip().lower() == "business_auxiliary": return [] if not _should_search_runtime_legal_sources(normalized_message): return [] try: return legal_article_service.search_runtime_sources(current_user, normalized_message, limit=5) except Exception: return [] def _should_search_runtime_legal_sources(message: str) -> bool: if re.search(r"[\u4e00-\u9fff]", message): return True lowered = message.lower() english_legal_markers = ( "article", "law", "regulation", "civil code", "company law", "labor contract law", "employment law", "privacy law", "pipl", "data security law", "cybersecurity law", "e-signature law", ) return any(marker in lowered for marker in english_legal_markers) def _enrich_review_context_with_runtime_legal_sources( review_context: Optional[Dict[str, Any]], runtime_sources: List[Dict[str, Any]], ) -> Optional[Dict[str, Any]]: if not runtime_sources: return review_context enriched = dict(review_context or {}) enriched["legalArticleSearchSources"] = runtime_sources enriched["legal_article_search_sources"] = runtime_sources metadata = dict(enriched.get("metadata") or {}) metadata["legalArticleSearchSourceCount"] = len(runtime_sources) metadata["legalArticleSearchSources"] = [ { "id": source.get("source_id"), "title": source.get("title"), "detail_path": source.get("detail_path"), } for source in runtime_sources ] enriched["metadata"] = metadata return enriched def _collect_legal_article_refs(risk_items: Iterable[Mapping[str, Any]]) -> List[Dict[str, Any]]: refs: List[Dict[str, Any]] = [] seen: set[str] = set() for item in risk_items: for ref in item.get("legal_article_refs") or []: if not isinstance(ref, Mapping): continue ref_id = str(ref.get("id") or "") key = ref_id or f"{ref.get('source_title')}::{ref.get('article_no')}::{ref.get('detail_path')}" if not key or key in seen: continue seen.add(key) refs.append(dict(ref)) return refs def _build_legal_article_sources(review_context: Optional[Mapping[str, Any]]) -> List[Dict[str, Any]]: if not review_context: return [] refs = review_context.get("legalArticleRefs") or review_context.get("legal_article_refs") or [] if not isinstance(refs, list): return [] sources: List[Dict[str, Any]] = [] for ref in refs: if not isinstance(ref, Mapping): continue source_id = ref.get("id") or ref.get("source_id") title = _build_legal_article_source_title(ref) content = ref.get("article_excerpt") or ref.get("match_reason") or ref.get("detail_path") or "" sources.append( { "title": title, "content": content, "score": 1.0, "source_type": "legal_article", "source_id": source_id, "detail_path": ref.get("detail_path"), "article_no": ref.get("article_no"), "authority": ref.get("authority"), "knowledge_base_id": ref.get("knowledge_base_id"), "knowledge_document_id": ref.get("knowledge_document_id"), "metadata": { "source_id": ref.get("source_id"), "source_title": ref.get("source_title"), "match_reason": ref.get("match_reason"), "tags": ref.get("tags") or [], }, } ) return _dedupe_sources(sources) def _build_legal_article_source_title(ref: Mapping[str, Any]) -> str: source_title = str(ref.get("source_title") or "法条全文") article_no = str(ref.get("article_no") or "") return f"{source_title} {article_no}".strip() def _merge_local_sources_into_stream(upstream: Iterable[str], local_sources: List[Dict[str, Any]], message: str = "") -> Iterable[str]: if not local_sources: yield from upstream return emitted_sources = False for chunk in upstream: event = _parse_sse_event(chunk) if event and event.get("type") == "text": content = str(event.get("content") or "") yield _format_sse_event({"type": "text", "content": _polish_chat_answer(content, message, local_sources)}) continue if event and event.get("type") == "sources": merged = _dedupe_sources([*(event.get("sources") or []), *local_sources]) emitted_sources = True yield _format_sse_event({"type": "sources", "sources": merged}) continue if event and event.get("type") == "end" and not emitted_sources: emitted_sources = True yield _format_sse_event({"type": "sources", "sources": local_sources}) yield chunk def _polish_chat_answer(content: str, message: str, sources: List[Dict[str, Any]]) -> str: if not _looks_like_raw_retrieval_answer(content): return content source_titles = [str(source.get("title") or "依据来源").strip() for source in sources[:3] if source.get("title")] source_line = "、".join(source_titles) if source_titles else "下方引用来源" return ( "我已检索到相关依据,但当前平台返回的是依据片段而不是完整法律分析。\n\n" f"针对你的问题“{message.strip()}”,建议先按以下方向人工核查:\n" "1. 明确付款节点、付款前置条件、发票要求和验收标准是否写清。\n" "2. 检查逾期付款、拒付、扣款和违约金规则是否具体、对等、可执行。\n" "3. 核对付款条款是否和交付、验收、解除、争议解决条款互相冲突。\n\n" f"可优先复核这些依据:{source_line}。" ) def _looks_like_raw_retrieval_answer(content: str) -> bool: normalized = content.strip().lower() if not normalized: return False return ( normalized.startswith("i found relevant knowledge-base content") or "please treat this as ai-assisted guidance" in normalized or "chapter_title:" in normalized or "official_url:" in normalized ) def _parse_sse_event(chunk: str) -> Optional[Dict[str, Any]]: for line in chunk.splitlines(): if not line.startswith("data: "): continue try: payload = json.loads(line[6:]) except json.JSONDecodeError: return None return payload if isinstance(payload, dict) else None return None def _format_sse_event(payload: Dict[str, Any]) -> str: return f"data: {json.dumps(payload, ensure_ascii=False)}\n\n" def _dedupe_sources(sources: Iterable[Mapping[str, Any]]) -> List[Dict[str, Any]]: deduped: List[Dict[str, Any]] = [] seen: set[str] = set() for source in sources: if not isinstance(source, Mapping): continue key = str(source.get("source_id") or source.get("detail_path") or f"{source.get('title')}::{source.get('content')}") if key in seen: continue seen.add(key) deduped.append(dict(source)) return deduped