"""Agent tool surface: wrap product features as named, documented capabilities. The REST API exposes many flows (upload, generate, similar content, canonical scan, …). The agentic HeadAgent should depend on this layer rather than reaching straight into retrieval helpers, so each capability has a stable name and schema for future LLM ``tool_calls`` / ``bind_tools`` integration. When ``inspector_tool_agent`` is enabled and an API key is set, the inspector (:mod:`app.agentic.inspector_loop`) invokes these implementations from an OpenAI ``tool_calls`` loop so the model chooses order and queries. The legacy HeadAgent path still calls the same functions in a fixed order when the key is missing or the setting is false (e.g. CI / mock adapter). """ from __future__ import annotations from dataclasses import dataclass from typing import Any from sqlalchemy.ext.asyncio import AsyncSession from app.config import settings from app.models.schemas import SearchResult, SimilarContentRequest, SimilarContentResponse from app.retrieval.retriever import retrieve_for_report, retrieve_for_report_async from app.retrieval.reranker import rerank as jaccard_rerank from app.retrieval.vector_search import async_vs_search from app.services.content_similarity import scan_similar_content from app.vectorstore.factory import get_vectorstore @dataclass(frozen=True) class ToolSpec: """OpenAI-style tool metadata (subset of JSON Schema).""" name: str description: str parameters: dict[str, Any] def list_tool_specs() -> list[ToolSpec]: """Return tool definitions for documentation or future ``bind_tools``.""" return [ ToolSpec( name="retrieve_tenant_evidence", description=( "Semantic search over the tenant's indexed survey and reference uploads, " "prioritising the report's primary document. Returns reranked text chunks with scores." ), parameters={ "type": "object", "properties": { "query": {"type": "string", "description": "Search query (e.g. joined inspection bullets)."}, "tenant_id": {"type": "string"}, "primary_document_id": {"type": "string"}, "secondary_document_ids": { "type": "array", "items": {"type": "string"}, "description": "Optional additional upload UUIDs to include in routing.", }, "k": {"type": "integer", "minimum": 1, "maximum": 80}, "rerank_top_n": {"type": "integer", "minimum": 1, "maximum": 20}, }, "required": ["query", "tenant_id", "primary_document_id", "k", "rerank_top_n"], }, ), ToolSpec( name="retrieve_kb_guidance", description=( "Search the local RICS standards / exemplar knowledge base (reserved tenant). " "Use for professional wording and compliance context; chunks may carry kb_path metadata." ), parameters={ "type": "object", "properties": { "query": {"type": "string"}, "k": {"type": "integer"}, "rerank_top_n": {"type": "integer", "description": "How many KB chunks to keep after reranking."}, "hierarchy_level": { "type": "string", "enum": ["document", "section", "paragraph"], "description": "Optional strict hierarchy filter for chunk granularity.", }, }, "required": ["query", "k"], }, ), ToolSpec( name="find_similar_library_and_peers", description=( "Find semantically similar indexed library chunks and lexical overlaps with other " "sections' draft notes (corpus hygiene / deduplication)." ), parameters={ "type": "object", "properties": { "tenant_id": {"type": "string"}, "text": {"type": "string"}, "section_code": {"type": "string"}, "peer_sections": {"type": "object", "additionalProperties": {"type": "string"}}, "exclude_document_ids": {"type": "array", "items": {"type": "string"}}, "limit": {"type": "integer"}, "min_relevance_percent": {"type": "number"}, }, "required": ["tenant_id", "text"], }, ), ] def retrieve_tenant_evidence( *, query: str, tenant_id: str, primary_document_id: str, secondary_document_ids: list[str] | None, k: int, rerank_top_n: int, ) -> list[SearchResult]: """Tool: tenant-scoped retrieval + rerank (same stack as interactive generate).""" candidates = retrieve_for_report( query=query, tenant_id=tenant_id, primary_document_id=primary_document_id, secondary_document_ids=list(secondary_document_ids or []), k=k, ) return list(jaccard_rerank(query=query, results=candidates, top_n=rerank_top_n)) async def retrieve_tenant_evidence_async( *, query: str, tenant_id: str, primary_document_id: str, secondary_document_ids: list[str] | None, k: int, rerank_top_n: int, ) -> list[SearchResult]: """Async tenant retrieval + rerank (semantic cache when Qdrant enabled).""" from app.retrieval.retriever import retrieve_for_report_unified candidates = await retrieve_for_report_unified( query=query, tenant_id=tenant_id, primary_document_id=primary_document_id, k=k, secondary_document_ids=list(secondary_document_ids or []), ) return list(jaccard_rerank(query=query, results=candidates, top_n=rerank_top_n)) def retrieve_kb_guidance( *, query: str, k: int, hierarchy_level: str | None, rerank_top_n: int = 5, ) -> list[SearchResult]: """Tool: knowledge-base retrieval + rerank (no-op when KB disabled).""" if not settings.knowledge_base_enabled: return [] vs = get_vectorstore() hl = hierarchy_level if hierarchy_level in ("document", "section", "paragraph") else None kb_hits = vs.search( query=query, tenant_id=settings.knowledge_base_tenant_id, k=k, hierarchy_level=hl, doc_id_in=None, ) return list(jaccard_rerank(query=query, results=kb_hits, top_n=rerank_top_n)) async def retrieve_kb_guidance_async( *, query: str, k: int, hierarchy_level: str | None, rerank_top_n: int = 5, ) -> list[SearchResult]: """Async KB retrieval + rerank.""" if not settings.knowledge_base_enabled: return [] vs = get_vectorstore() hl = hierarchy_level if hierarchy_level in ("document", "section", "paragraph") else None kb_hits = await async_vs_search( vs, query, settings.knowledge_base_tenant_id, k=k, hierarchy_level=hl, ) return list(jaccard_rerank(query=query, results=kb_hits, top_n=rerank_top_n)) def dedupe_search_results(results: list[SearchResult]) -> list[SearchResult]: """Tool helper: dedupe by chunk_id preserving first-seen order.""" seen: set[str] = set() out: list[SearchResult] = [] for r in results: if r.chunk_id in seen: continue seen.add(r.chunk_id) out.append(r) return out async def find_similar_library_and_peers( db: AsyncSession, tenant_id: str, *, text: str, section_code: str | None = None, peer_sections: dict[str, str] | None = None, exclude_document_ids: list[str] | None = None, limit: int = 8, min_relevance_percent: float = 28.0, ) -> SimilarContentResponse: """Tool: wraps ``POST /content/similar`` logic without an HTTP round-trip.""" body = SimilarContentRequest( text=text, section_code=section_code, peer_sections=dict(peer_sections or {}), exclude_document_ids=list(exclude_document_ids or []), limit=limit, min_relevance_percent=min_relevance_percent, ) return await scan_similar_content(db, tenant_id, body)