| from __future__ import annotations
|
|
|
| from typing import List, Tuple
|
| import os
|
| import json
|
| from urllib.parse import urlparse
|
|
|
| from ai_agent.retriever.software_doc import SoftwareDoc
|
| from ai_agent.core.pipeline_registry import get_pipeline as get_shared_pipeline
|
| from ai_agent.api.pipeline import RAGImagingPipeline
|
|
|
| _DOCS: List[SoftwareDoc] = []
|
| _KNOWN_NAMES: List[str] = []
|
| MAX_CHARS = 20000
|
|
|
|
|
| def get_catalog_docs() -> List[SoftwareDoc]:
|
| """
|
| Load and return catalog docs without initializing the full pipeline.
|
|
|
| This is a lightweight alternative to get_pipeline() for catalog-only operations
|
| that don't need the embedder, reranker, or index.
|
| """
|
| global _DOCS, _KNOWN_NAMES
|
| if not _DOCS:
|
|
|
| from pathlib import Path
|
|
|
| catalog = os.getenv("SOFTWARE_CATALOG", "data/sample.jsonl")
|
| p = Path(catalog)
|
| docs: List[SoftwareDoc] = []
|
| if p.exists():
|
| text = p.read_text(encoding="utf-8").strip()
|
| try:
|
| obj = json.loads(text)
|
| if isinstance(obj, dict):
|
| obj = [obj]
|
| for o in obj:
|
| docs.append(SoftwareDoc.model_validate(o))
|
| except Exception:
|
| for line in text.splitlines():
|
| line = line.strip()
|
| if not line:
|
| continue
|
| try:
|
| docs.append(SoftwareDoc.model_validate(json.loads(line)))
|
| except Exception:
|
| continue
|
| _DOCS = docs
|
| _KNOWN_NAMES = [d.name for d in _DOCS if getattr(d, "name", None)]
|
| return _DOCS
|
|
|
|
|
| def get_known_names() -> List[str]:
|
| """Return a cached list of all tool names from the catalog."""
|
| if not _DOCS:
|
| get_catalog_docs()
|
| return _KNOWN_NAMES
|
|
|
|
|
| def get_pipeline() -> "RAGImagingPipeline":
|
|
|
| get_catalog_docs()
|
|
|
| return get_shared_pipeline()
|
|
|
|
|
| def _clip(s: str) -> Tuple[str, bool]:
|
| if not s:
|
| return s, False
|
| if len(s) <= MAX_CHARS:
|
| return s, False
|
| return s[:MAX_CHARS] + "\n\n...[truncated for token budget]...", True
|
|
|
|
|
| def _is_github_url(url: str) -> bool:
|
| """Return True only for URLs that actually point to github.com."""
|
| s = (url or "").strip()
|
| if not s:
|
| return False
|
|
|
| parsed = urlparse(s)
|
|
|
|
|
| if not parsed.scheme and not parsed.netloc:
|
| parsed = urlparse("https://" + s)
|
|
|
| return parsed.netloc.lower() == "github.com"
|
|
|
