Spaces:
Sleeping
Sleeping
File size: 8,424 Bytes
b76f199 732b14f b76f199 732b14f b76f199 732b14f b76f199 732b14f b76f199 | 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 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 | """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)
|