from __future__ import annotations import json import os import urllib.request from collections.abc import Mapping from typing import Any, Callable, Protocol class CardGenerationClient(Protocol): def generate_card(self, source_bundle: dict[str, object]) -> dict[str, object]: """Return structured card sections for a normalized source bundle.""" class DeterministicCardGenerationClient: """Fallback card generator used when no remote model client is configured.""" def generate_card(self, source_bundle: dict[str, object]) -> dict[str, object]: abstract = str(source_bundle.get("abstract", "")).strip() language = str(source_bundle.get("card_language", "en")) notes = [str(value).strip() for value in source_bundle.get("notes", []) if str(value).strip()] attachment_texts = [ str(value).strip() for value in source_bundle.get("attachment_texts", []) if str(value).strip() ] summary = abstract or (notes[0] if notes else "") or (attachment_texts[0][:240] if attachment_texts else "") claims = (notes[:3] or [summary]) if summary else [] return { "summary": summary if language == "en" else f"中文：{summary}", "core_claims": claims if language == "en" else [f"中文：{claim}" for claim in claims], "methods": [], "evidence": attachment_texts[:3], "citations": [ { "claim": claims[0] if claims else summary, "quote": attachment_texts[0][:240], "paraphrase": summary if language == "en" else f"中文：{summary}", "quote_source": "attachment_texts", "use_case": "direct_quote", } ] if attachment_texts else [], "quotable_passages": attachment_texts[:2], "writing_hints": [ "Use this card when the writing intent overlaps with its summary or notes." if language == "en" else "当写作意图与摘要或笔记相关时使用这张卡片。", ], "keywords": [str(value) for value in source_bundle.get("tags", [])], } RequestFunction = Callable[[str, dict[str, object], dict[str, str]], dict[str, object]] class DeepSeekCardGenerationClient: def __init__( self, api_key: str, *, model: str = "deepseek-chat", base_url: str = "https://api.deepseek.com/v1", request_fn: RequestFunction | None = None, ) -> None: self.api_key = api_key self.model = model self.base_url = base_url.rstrip("/") self.request_fn = request_fn or self._default_request def generate_card(self, source_bundle: dict[str, object]) -> dict[str, object]: language = str(source_bundle.get("card_language", "en")) language_instruction = ( "Write summary, core_claims, methods, evidence, paraphrase fields, and writing_hints in Chinese." if language == "zh" else "Write summary, core_claims, methods, evidence, paraphrase fields, and writing_hints in English." ) payload = { "model": self.model, "response_format": {"type": "json_object"}, "messages": [ { "role": "system", "content": ( "You convert research source bundles into structured evidence cards for academic writing. " "Never hallucinate. Use only information explicitly supported by the source bundle. " "Preserve uncertainty and hedging. Return valid JSON only with keys: " "summary, core_claims, methods, evidence, citations, quotable_passages, writing_hints, keywords." ), }, { "role": "user", "content": ( "Given this source bundle, return exactly one JSON object with keys: " "summary, core_claims, methods, evidence, citations, quotable_passages, writing_hints, keywords.\n\n" "Rules:\n" "1. summary: 2-4 factual sentences, no hype.\n" "2. core_claims: concrete findings only; preserve numbers and hedging.\n" "3. methods: brief method or setup details actually stated in the source.\n" "4. evidence: empirical or textual support from the source, numbers preferred.\n" "5. citations: an array of objects with keys claim, quote, paraphrase, quote_source, use_case.\n" "6. citations.quote must be verbatim text from notes or attachment_texts only.\n" "7. Return 1-3 citations whenever the source bundle contains usable supporting text in notes or attachment_texts.\n" "8. quote_source must identify where the quote came from, such as attachment_texts or notes.\n" "9. use_case must be one of direct_quote, paraphrase, or background.\n" "10. quotable_passages: short verbatim excerpts from notes or attachment_texts only.\n" "11. Use empty arrays only when the source bundle truly contains no usable supporting quote; do not fabricate.\n" f"12. {language_instruction}\n\n" f"Source bundle:\n{json.dumps(source_bundle, ensure_ascii=False)}" ), }, ], } headers = { "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json", } response = self.request_fn(f"{self.base_url}/chat/completions", payload, headers) content = str(response["choices"][0]["message"]["content"]) parsed = json.loads(_strip_json_fence(content)) return { "summary": str(parsed.get("summary", "")), "core_claims": [str(value) for value in parsed.get("core_claims", [])], "methods": [str(value) for value in parsed.get("methods", [])], "evidence": [str(value) for value in parsed.get("evidence", [])], "citations": [ { "claim": str(value.get("claim", "")), "quote": str(value.get("quote", "")), "paraphrase": str(value.get("paraphrase", "")), "quote_source": str(value.get("quote_source", "")), "use_case": str(value.get("use_case", "")), } for value in parsed.get("citations", []) if isinstance(value, dict) ], "quotable_passages": [str(value) for value in parsed.get("quotable_passages", [])], "writing_hints": [str(value) for value in parsed.get("writing_hints", [])], "keywords": [str(value) for value in parsed.get("keywords", [])], } @staticmethod def _default_request(url: str, payload: dict[str, object], headers: dict[str, str]) -> dict[str, object]: request = urllib.request.Request( url=url, data=json.dumps(payload, ensure_ascii=False).encode("utf-8"), headers=headers, method="POST", ) with urllib.request.urlopen(request, timeout=120) as response: return json.loads(response.read().decode("utf-8")) def create_card_generation_client( provider: str, model: str, *, env: Mapping[str, str] | None = None, request_fn: RequestFunction | None = None, ) -> CardGenerationClient: current_env = os.environ if env is None else env normalized = provider.lower() if normalized == "deepseek": api_key = current_env.get("DEEPSEEK_API_KEY") if not api_key: raise ValueError("DEEPSEEK_API_KEY is required when llm provider is deepseek") return DeepSeekCardGenerationClient(api_key=api_key, model=model, request_fn=request_fn) return DeterministicCardGenerationClient() def _strip_json_fence(value: str) -> str: stripped = value.strip() if stripped.startswith("```"): lines = stripped.splitlines() if len(lines) >= 3 and lines[0].startswith("```") and lines[-1].strip() == "```": return "\n".join(lines[1:-1]).strip() return stripped