File size: 2,090 Bytes
9b2d3bb |
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 |
from langchain.tools import BaseTool
from typing import Type, Optional
from pydantic import BaseModel, Field
from clinical_nlp.umls_bioportal import search_bioportal_term
from services.logger import app_logger
from services.metrics import log_tool_usage
class BioPortalInput(BaseModel):
term: str = Field(description="The medical term to search for.")
ontology: Optional[str] = Field(default="SNOMEDCT", description="The specific ontology to search within BioPortal (e.g., SNOMEDCT, ICD10, RxNorm). Defaults to SNOMEDCT.")
class BioPortalLookupTool(BaseTool):
name: str = "bioportal_lookup"
description: str = (
"Useful for searching medical terms, codes, and definitions across various ontologies "
"via BioPortal. Specify the term and optionally the ontology (e.g., SNOMEDCT, ICD10CM, RxNorm)."
)
args_schema: Type[BaseModel] = BioPortalInput
def _run(self, term: str, ontology: Optional[str] = "SNOMEDCT") -> str:
app_logger.info(f"BioPortal Tool called with term: {term}, ontology: {ontology}")
log_tool_usage(self.name)
results = search_bioportal_term(term, ontology=ontology or "SNOMEDCT")
if "error" in results:
return f"Error from BioPortal lookup: {results['error']}"
# Format results for LLM consumption
collection = results.get("collection", [])
if collection:
formatted_results = []
for item in collection[:3]: # Limit to 3 results
defs = item.get('definition', ['N/A'])
definition_str = "; ".join(defs) if defs else "N/A"
formatted_results.append(
f"- Term: {item.get('prefLabel', 'N/A')}, Definition: {definition_str}, CUIs: {item.get('cui', [])}"
)
return f"BioPortal Results (Ontology: {ontology}):\n" + "\n".join(formatted_results)
return f"No results found in BioPortal for ontology {ontology}."
async def _arun(self, term: str, ontology: Optional[str] = "SNOMEDCT") -> str:
return self._run(term, ontology) |