Spaces:
Sleeping
Sleeping
File size: 5,888 Bytes
9e95393 775cbe6 aaa4456 9e95393 aaa4456 9e95393 aaa4456 775cbe6 aaa4456 9e95393 aaa4456 9e95393 775cbe6 9e95393 aaa4456 9e95393 775cbe6 9e95393 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 | from __future__ import annotations
import json
import os
from collections.abc import Iterable, Mapping
from pathlib import Path
from .bootstrap import RuntimeDependencies, build_work
from .hf_client import (
HF_CHAT_API_ADVERSARIAL,
HF_CHAT_API_CONSTRUCTIVE,
HF_CHAT_API_EVIDENCE,
)
from .kb_bucket import (
HF_KB_ACTIVE_POINTER_DEFAULT,
HF_KB_BUCKET_DEFAULT,
HF_KB_MOUNT_DEFAULT,
load_mounted_kb_release,
load_remote_kb_release,
)
from .live_state import EmptyLiveStateProvider, HybridLiveStateProvider
from .service import CustomerAIWork
class RuntimeNotReady(RuntimeError):
def __init__(self, code: str):
super().__init__(code)
self.code = code
def _required_file(value: str, code: str) -> Path:
path = Path(value).expanduser() if value else None
if path is None or not path.is_file():
raise RuntimeNotReady(code)
return path
def _required_value(values: Mapping[str, str], key: str, code: str) -> str:
value = values.get(key, "").strip()
if not value:
raise RuntimeNotReady(code)
return value
def _load_alias_registry(path_value: str) -> Mapping[str, Iterable[str]]:
if not path_value.strip():
return {}
path = _required_file(path_value.strip(), "alias_registry_missing")
try:
raw = json.loads(path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as exc:
raise RuntimeNotReady("alias_registry_invalid") from exc
if not isinstance(raw, dict):
raise RuntimeNotReady("alias_registry_invalid")
normalized: dict[str, list[str]] = {}
for canonical, aliases in raw.items():
if not isinstance(canonical, str) or not isinstance(aliases, list):
raise RuntimeNotReady("alias_registry_invalid")
normalized[canonical] = [str(alias) for alias in aliases if str(alias).strip()]
return normalized
def _production_kb_files(values: Mapping[str, str]) -> tuple[Path, Path | None, Path | None, str]:
build_id = _required_value(values, "CUSTOMER_AI_KB_BUILD_ID", "kb_build_id_missing")
mount_path = values.get("CUSTOMER_AI_KB_MOUNT_PATH", "").strip() or HF_KB_MOUNT_DEFAULT
pointer_name = values.get("CUSTOMER_AI_KB_ACTIVE_POINTER", "").strip() or HF_KB_ACTIVE_POINTER_DEFAULT
bucket_id = values.get("CUSTOMER_AI_KB_BUCKET_ID", "").strip() or HF_KB_BUCKET_DEFAULT
token = (values.get("HF_TOKEN", "") or values.get("HF_KEY", "")).strip()
try:
release = load_mounted_kb_release(
mount_path=mount_path,
expected_build_id=build_id,
pointer_name=pointer_name,
)
except ValueError as exc:
if str(exc) != "kb_bucket_mount_missing":
raise RuntimeNotReady(str(exc)) from exc
if not token:
raise RuntimeNotReady("hf_token_missing") from exc
try:
release = load_remote_kb_release(
bucket_id=bucket_id,
token=token,
expected_build_id=build_id,
pointer_name=pointer_name,
)
except ValueError as remote_exc:
raise RuntimeNotReady(str(remote_exc)) from remote_exc
return (
release.canonical_path,
release.current_facts_path,
release.aliases_path,
release.build_id,
)
def create_work_from_environment(
env: Mapping[str, str] | None = None,
*,
role_pool: object | None = None,
) -> CustomerAIWork:
values = os.environ if env is None else env
if role_pool is None:
kb_path, current_path, alias_path, build_id = _production_kb_files(values)
generation_id = values.get("CUSTOMER_AI_KB_GENERATION_ID", "").strip() or build_id
else:
kb_path = _required_file(
values.get("CUSTOMER_AI_KB_SNAPSHOT_PATH", "").strip(),
"kb_snapshot_missing",
)
generation_id = values.get("CUSTOMER_AI_KB_GENERATION_ID", "").strip() or kb_path.stem
current_value = values.get("CUSTOMER_AI_CURRENT_FACTS_PATH", "").strip()
current_path = _required_file(current_value, "current_facts_missing") if current_value else None
alias_value = values.get("CUSTOMER_AI_ALIAS_REGISTRY_PATH", "").strip()
alias_path = _required_file(alias_value, "alias_registry_missing") if alias_value else None
if current_path is not None:
live_provider = HybridLiveStateProvider.from_jsonl(
current_path,
generation_id=f"{generation_id}:current",
)
else:
live_provider = EmptyLiveStateProvider()
# The HF token is needed only when the private KB bucket must be read remotely.
# Local role inference never uses the token.
token = (values.get("HF_TOKEN", "") or values.get("HF_KEY", "")).strip()
try:
fuzzy_threshold = float(values.get("CUSTOMER_AI_JA_FUZZY_THRESHOLD", "90"))
except ValueError as exc:
raise RuntimeNotReady("japanese_fuzzy_threshold_invalid") from exc
return build_work(
RuntimeDependencies(
live_state_provider=live_provider,
japanese_alias_registry=_load_alias_registry(str(alias_path) if alias_path else ""),
japanese_fuzzy_threshold=fuzzy_threshold,
kb_snapshot_path=str(kb_path),
kb_generation_id=generation_id,
hf_token=token or None,
role_pool=role_pool,
constructive_api_url=values.get("CUSTOMER_AI_CONSTRUCTIVE_API_URL", "").strip()
or HF_CHAT_API_CONSTRUCTIVE,
adversarial_api_url=values.get("CUSTOMER_AI_ADVERSARIAL_API_URL", "").strip()
or HF_CHAT_API_ADVERSARIAL,
evidence_api_url=values.get("CUSTOMER_AI_EVIDENCE_API_URL", "").strip()
or HF_CHAT_API_EVIDENCE,
timeout_seconds=float(values.get("CUSTOMER_AI_ROLE_TIMEOUT_SECONDS", "600")),
)
)
|