Ashmatics Tools

Last updated: 2026-07-12

Version 0.8.2

A Python package providing shared utilities, base classes, and common functionality for Ashmatics Knowledge Base applications.

Overview

ashmatics-tools is a foundational library that centralizes reusable components across Ashmatics healthcare AI applications. It provides:

• Data Import/Export Utilities: Excel data loading, GraphQL integration, batch processing with Hasura
• Document Processors: Abstract base classes for MongoDB document processing
• GraphQL Clients: Generic GraphQL query/mutation builders and client utilities
• Schema Management: GraphQL schema introspection and analysis tools
• Document Parsers: Advanced parsing for PDFs, DOCX, and other formats
• Document Chunkers: Token-aware and semantic chunking strategies
• Embedders: Generate embeddings using Azure OpenAI or OpenAI APIs
• Vector Stores: Integration with CosmosDB, PostgreSQL, and Qdrant for vector search
• Storage Backends: Cloud-agnostic storage abstraction for ADLS Gen2, MinIO, and AWS S3
• LLM Clients: Unified interface for Azure OpenAI, OpenAI, HuggingFace, and custom providers
• Ontology Services: Medical ontology management including SNOMED CT, RADLEX, LOINC, and custom Ashmatics ontologies
• Term Services: Term resolution, hierarchical category management, and external ontology validation
• External APIs: Clients for external data sources (FDA, Census, CMS) with retry, rate limiting, and pagination
• MCP Servers: Model Context Protocol servers exposing APIs to LLMs with tool-based interfaces
• Search/RAG: Retrieval-Augmented Generation strategies with streaming, context window management, and MCP tool definitions
• Document Enrichers: Table classification, consolidation, and metrics extraction for parsed documents
• Document Storage: Figure and table storage managers with content-addressed hashing and manifests
• Population Graph: Bayesian network population generation engine using pgmpy — data-driven graph structure, CPDs, and state names from JSON model definitions with ExpertKnowledge hooks for clinical prior-constrained structure learning

Installation

ashmatics-tools is published on PyPI. The core install is intentionally lightweight and does not pull in torch/CUDA — heavy backends live behind optional extras (see Dependencies below).

# Core (from PyPI)
pip install ashmatics-tools
uv add ashmatics-tools

# With a convenience bundle
pip install "ashmatics-tools[api]"       # Azure storage + MongoDB — what API apps need
pip install "ashmatics-tools[docproc]"   # full parse -> chunk pipeline
pip install "ashmatics-tools[full]"      # everything except dev tools

From Git (unreleased changes)

pip install git+https://github.com/AshMatics/ashmatics-tools.git
uv add git+https://github.com/AshMatics/ashmatics-tools.git

From Local Development

git clone https://github.com/AshMatics/ashmatics-tools.git
cd ashmatics-tools

# Editable install with dev tools + the API bundle
pip install -e ".[dev,api]"

Configuration

Environment Variables

ashmatics-tools requires various environment variables depending on which components you use. This library does not load .env files automatically - your application must handle environment variable loading.

See ENV_VARIABLES.md for:

• Complete list of required environment variables by component
• Example application setups (development with .env, production with Key Vault)
• Environment-specific configurations

Quick example:

from dotenv import load_dotenv

# Load .env BEFORE importing ashmatics_tools
load_dotenv()

# Now use the library
from ashmatics_tools.embedders import create_embedder
embedder = create_embedder(provider="azure")

Usage

Knowledge Base Importer

from ashmatics_tools.utils.import_utils import KBImporter

# Initialize the importer
importer = KBImporter(
    graphql_endpoint="https://kb-api.ashmatics.com/v1/graphql",
    admin_secret="your-admin-secret",
    batch_size=100
)

# Load data from Excel
df = importer.load_excel_data("data.xlsx", sheet_name="Sheet1")

# Import to Knowledge Base via GraphQL
result = importer.import_to_kb(
    df=df,
    table_name="my_table",
    column_mapping={"excel_col": "db_col"}
)

Document Processor (MongoDB)

from ashmatics_tools.processors.base import DocumentProcessor
from pymongo import MongoClient

class MyDocumentProcessor(DocumentProcessor):
    def extract_metadata(self, document: dict) -> dict:
        return {"title": document.get("title"), "author": document.get("author")}

    def clean_text(self, text: str) -> str:
        return text.strip().lower()

    def get_identifier_key(self) -> str:
        return "document_id"

    def get_document_type(self) -> str:
        return "my_document_type"

    def process_document(self, file_path: str) -> dict:
        # Your document processing logic
        return {"document_id": "123", "content": "..."}

# Use the processor
client = MongoClient("mongodb://localhost:27017")
processor = MyDocumentProcessor(client, "my_database", "my_collection")
result = processor.upsert_document({"document_id": "123", "content": "..."})

Document Chunking

from ashmatics_tools.chunkers.factory import create_chunker

# Initialize chunker
chunker = create_chunker(strategy="docling")

# Chunk document
chunks = chunker.chunk_document(
    content="This is a sample document content.",
    title="Sample Document",
    source="document.pdf"
)

Embedding Generation

from ashmatics_tools.embedders.factory import create_embedder

# Initialize embedder
embedder = create_embedder(provider="azure")
embedder.initialize()

# Generate embeddings
embeddings = embedder.embed_chunks(["chunk1", "chunk2"])

Vector Store Integration

from ashmatics_tools.vector_stores.factory import create_vector_store

# Initialize vector store
vector_store = create_vector_store(provider="cosmosdb")

# Store embeddings
success, failed = vector_store.store_embeddings_batch(embeddings)

# Perform similarity search
results = vector_store.similarity_search(query_embedding, top_k=10)

Storage Backend Integration

from ashmatics_tools.storage import create_storage_client, StorageConfig, AuthType

# Initialize ADLS storage with DefaultAzureCredential (production)
config = StorageConfig(
    provider="adls",
    account_url="https://mystorageaccount.dfs.core.windows.net",
    container_name="my-container",
    auth_type=AuthType.DEFAULT_CREDENTIAL
)
storage = create_storage_client("adls", config)

# Or use connection string (development)
config = StorageConfig(
    provider="adls",
    connection_string="DefaultEndpointsProtocol=https;AccountName=...",
    container_name="my-container",
    auth_type=AuthType.CONNECTION_STRING
)
storage = create_storage_client("adls", config)

# Initialize MinIO storage
config = StorageConfig(
    provider="minio",
    endpoint="minio.example.com:9000",
    access_key="minioadmin",
    secret_key="minioadmin",
    container_name="my-bucket",
    auth_type=AuthType.ACCESS_KEY,
    use_ssl=False
)
storage = create_storage_client("minio", config)

# Use async context manager
async with storage:
    # Write object
    await storage.write_object("path/file.txt", b"Hello, World!")

    # Read object
    content = await storage.read_object("path/file.txt")

    # List objects
    objects = await storage.list_objects(prefix="path/", pattern="*.txt")

    # Stream large files
    async for chunk in storage.read_object_stream("large-file.bin"):
        process(chunk)

    # Check if exists
    exists = await storage.exists("path/file.txt")

    # Get metadata
    metadata = await storage.get_metadata("path/file.txt")

    # Copy object
    await storage.copy_object("src/file.txt", "dest/file.txt")

    # Delete object
    await storage.delete_object("path/file.txt")

LLM Module Usage

The LLM module provides a unified interface for working with various language model providers. Supports Azure OpenAI, OpenAI, HuggingFace, and custom providers via plugin registry.

Key Features

• Async-first API: All operations are async for high-performance pipelines
• Unified interface: Same API across all providers
• Cost tracking: Automatic token counting and cost estimation
• Plugin registry: Extensible with custom providers
• Optional dependencies: HuggingFace support via [huggingface] extra

Azure OpenAI (Primary)

from ashmatics_tools.llm import create_llm_client, AzureOpenAIConfig

config = AzureOpenAIConfig(
    endpoint="https://my-resource.openai.azure.com/",
    api_key=os.getenv("AZURE_OPENAI_API_KEY"),
    deployment_name="gpt-4"
)

async with create_llm_client("azure_openai", config) as llm:
    response = await llm.complete(
        prompt="What is asthma?",
        temperature=0.7,
        max_tokens=500
    )
    print(response.text)
    print(f"Cost: ${response.tokens.estimated_cost:.4f}")

Ontology and Term Services

The ontology module provides comprehensive medical ontology management including term resolution, hierarchical category management, and integration with external ontologies.

Key Features

• Term Resolution: MongoDB-based term lookup and management
• Category Management: Hierarchical category structures for document tagging
• External Ontology Integration: BioPortal API for validating terms against SNOMED CT, RADLEX, LOINC, NCIT
• Custom Ontology: ASHMATICS domain-specific ontology for medical imaging AI concepts

Term Resolution

from ashmatics_tools.ontology import TermResolver
from pymongo import MongoClient

# Initialize term resolver
client = MongoClient("mongodb://localhost:27017")
term_resolver = TermResolver(mongodb_client=client)

# Resolve term
result = await term_resolver.resolve_term("breast cancer")
print(f"Resolved: {result.prefLabel} - {result.definition}")

Category Management

from ashmatics_tools.ontology import CategoryManager

# Initialize category manager
category_manager = CategoryManager(
    mongodb_database=client["ashmatics_kb"],
    term_resolver=term_resolver
)

# Create hierarchical category
category = await category_manager.create_category(
    name="Medical Imaging",
    parent_id=None,
    description="Top-level category for medical imaging"
)

# Add subcategory
subcategory = await category_manager.create_category(
    name="Breast Imaging",
    parent_id=category.id,
    description="Breast imaging techniques and AI models"
)

External Ontology Validation

from ashmatics_tools.ontology import BioPortalClient

# Initialize BioPortal client
bioportal = BioPortalClient(api_key="your-bioportal-api-key")

# Check term in external ontologies
exists, ontologies = await bioportal.check_term_in_ontology("Breast Cancer")
print(f"Term exists: {exists}")
print(f"Found in ontologies: {ontologies}")

Custom ASHMATICS Ontology

from ashmatics_tools.ontology import AshmaticsOntology

# Initialize custom ontology manager
ashmatics_ontology = AshmaticsOntology(mongodb_database=client["ashmatics_kb"])

# Create concept
concept = await ashmatics_ontology.create_concept(
    prefLabel="AI Breast Cancer Detector",
    definition="AI model for detecting breast cancer in medical images",
    synonyms=["Breast Cancer AI", "Mammography AI"]
)

# Add relationship
await ashmatics_ontology.add_relationship(
    source_id=concept.id,
    target_id=another_concept.id,
    relationship_type="related_to"
)

ASHCAI Clinical AI Governance Ontology

The ASHCAI (AshMatics Clinical AI Governance) ontology provides governance concepts for the CAI Framework, including policies, processes, controls, and regulatory crosswalks.

from ashmatics_tools.ontology import AshcaiOntology

# Initialize ASHCAI ontology manager
ashcai = AshcaiOntology(mongodb_database=client["ashmatics_kb"])

# Initialize collections and indexes
await ashcai.initialize_ontology()

# Create a policy domain with natural business ID
policy = await ashcai.create_policy_domain(
    domain_id="MMP-001",
    domain_code="MMP",
    label="Model Monitoring Policy",
    description="Policy governing AI model monitoring requirements",
    specifies=["MON"]  # Links to process domains
)

# Create a process domain
process = await ashcai.create_process_domain(
    domain_id="MON",
    label="Model Monitoring",
    primary_function="Continuous monitoring of AI model performance",
    integrates_with=["RM", "SA", "OVR"]
)

# Create base practice
practice = await ashcai.create_base_practice(
    practice_id="MON.BP01",
    label="Rollout and Change Management",
    process_domain="MON",
    sequence_order=1
)

# Create SOP template
sop = await ashcai.create_sop_template(
    sop_id="SOP-MON-01",
    label="Model Deployment SOP",
    purpose="Standard procedure for deploying AI models",
    base_practice="MON.BP01"
)

# Create work product template
wp = await ashcai.create_work_product_template(
    wp_id="WP-MON-02-Dashboard",
    label="Monitoring Dashboard",
    evidence_type="Dashboard",
    produced_by="SOP-MON-02",
    serves_as_evidence_for=["EXC-A7-04"]
)

# Create exemplar control
control = await ashcai.create_exemplar_control(
    control_id="EXC-A7-04",
    label="Performance Monitoring Control",
    iso_control="A.7.5",
    evidenced_by=["WP-MON-02-Dashboard"]
)

# Create regulatory requirement with crosswalk
requirement = await ashcai.create_regulatory_requirement(
    requirement_id="NIST-MAP-4.2",
    label="MAP 4.2",
    framework_id="NIST-AI-RMF",
    function="MAP",
    category="MAP-4",
    description="Internal risk controls for third-party AI resources",
    crosswalk={
        "addressedBy": ["TPP-001", "MMP-001"],
        "implementedThrough": ["EXC-A10-02", "EXC-A6-02"],
        "operationalizedIn": ["MON", "PV"],
        "evidencedBy": ["WP-MON-01", "WP-PV-03"]
    }
)

# Create relationships
await ashcai.link_policy_to_process("MMP-001", "MON")
await ashcai.link_process_to_practice("MON", "MON.BP01")
await ashcai.link_practice_to_sop("MON.BP01", "SOP-MON-01")
await ashcai.link_sop_to_workproduct("SOP-MON-01", "WP-MON-02-Dashboard")
await ashcai.link_workproduct_to_control("WP-MON-02-Dashboard", "EXC-A7-04")

# Traversal helpers
hierarchy = await ashcai.get_policy_hierarchy("MMP-001")
# Returns: policy, processes (with practices, SOPs, work products), controls

evidence_chain = await ashcai.get_evidence_chain("EXC-A7-04")
# Returns: control with all work products that evidence it

crosswalk = await ashcai.get_regulatory_crosswalk("NIST-MAP-4.2")
# Returns: requirement with all policies, controls, processes, evidence

# OWL/RDF export
uri = ashcai.generate_uri("MMP-001")
# Returns: http://asherinformatics.com/ontology/ashcai/MMP-001

Key Features:

• Natural Business IDs: Human-readable identifiers (MMP-001, MON, SOP-MON-01) with regex validation
• Type Discrimination: All documents include ontology: "ashcai" for filtering
• 44 Relationship Types: Comprehensive relationships from TDD-001 specification
• Traversal Helpers: Pre-built queries for policy hierarchies and regulatory crosswalks
• OWL/RDF Export: Generate standard URIs from natural business IDs

External API Integration

The external_apis module provides clients for accessing external data sources with built-in retry logic, rate limiting, and pagination.

OpenFDA API Client

from ashmatics_tools.external_apis import create_api_client, OpenFDAConfig, OpenFDAEndpoint

# Create client with API key (recommended)
config = OpenFDAConfig(api_key="your_api_key")
async with create_api_client("openfda", config) as client:
    # Search device adverse events
    async for event in client.search(
        endpoint=OpenFDAEndpoint.DEVICE_EVENT,
        query="device_name:pacemaker AND date_received:[20230101 TO 20231231]",
        limit=100,
        max_records=1000
    ):
        device = event.get("device", [{}])[0]
        print(f"Device: {device.get('device_name')}")
        print(f"Event Date: {event.get('date_received')}")

    # Search 510(k) clearances
    async for clearance in client.search(
        endpoint=OpenFDAEndpoint.DEVICE_510K,
        query="product_code:OZP",
        limit=100
    ):
        print(f"K Number: {clearance.get('k_number')}")
        print(f"Applicant: {clearance.get('applicant')}")

    # Analytics - count by field
    counts = await client.count(
        endpoint=OpenFDAEndpoint.DEVICE_EVENT,
        query="date_received:[20230101 TO 20231231]",
        count_field="device.device_class.exact"
    )
    for item in counts:
        print(f"Class {item['term']}: {item['count']} events")

AccessGUDID API Client

from ashmatics_tools.external_apis import AccessGUDIDClient, AccessGUDIDConfig

# Create client (no API key required for basic operations)
config = AccessGUDIDConfig()
async with AccessGUDIDClient(config) as client:
    # Lookup device by Device Identifier (DI)
    device = await client.lookup_device(di="08717648200274")
    print(f"Brand: {device['gudid']['device']['brandName']}")
    print(f"Company: {device['gudid']['device']['companyName']}")

    # Parse a UDI string (GS1, HIBCC, or ICCBBA format)
    parsed = await client.parse_udi(
        udi="(01)00844588012919(17)141231(10)A213B1"
    )
    print(f"DI: {parsed['di']}")
    print(f"Issuing Agency: {parsed['issuingAgency']}")
    print(f"Expiration: {parsed['expirationDate']}")
    print(f"Lot Number: {parsed['lotNumber']}")

    # Get device version history
    history = await client.get_device_history(di="08717648200274")
    for version in history.get('deviceHistory', []):
        print(f"Version {version['publicVersionNumber']}: {version['publicVersionDate']}")

    # List implantable devices with date filtering
    async for device in client.list_implantable_devices(
        from_date="2024-01-01",
        max_records=100
    ):
        print(f"{device['brandName']} - {device['companyName']}")

# With UMLS API key for SNOMED lookups
config = AccessGUDIDConfig(umls_api_key="your_umls_key")
async with AccessGUDIDClient(config) as client:
    snomed = await client.get_device_snomed(di="08717648200274")
    for concept in snomed.get('concepts', []):
        print(f"{concept['snomedCTName']}: {concept['snomedIdentifier']}")

MCP Server Integration

The mcp_servers module provides Model Context Protocol servers that expose external APIs as tools for LLM consumption.

OpenFDA MCP Server

from ashmatics_tools.mcp_servers import create_mcp_server, OpenFDAMCPConfig
from ashmatics_tools.external_apis import OpenFDAConfig

# Create MCP server
config = OpenFDAMCPConfig(
    api_config=OpenFDAConfig(api_key="your_key")
)
server = create_mcp_server("openfda", config)

# Get available tools
tools = server.get_tools()
# Returns: search_devices, search_drugs, count_by_field

# Call a tool
result = await server.call_tool("search_devices", {
    "endpoint": "device_event",
    "query": "device_name:pacemaker",
    "limit": 10
})

print(f"Found {result['count']} results")
for item in result['results']:
    print(item)

AccessGUDID MCP Server

from ashmatics_tools.mcp_servers import create_mcp_server, AccessGUDIDMCPConfig
from ashmatics_tools.external_apis import AccessGUDIDConfig

# Create MCP server
config = AccessGUDIDMCPConfig(
    api_config=AccessGUDIDConfig()  # No API key required for basic operations
)
server = create_mcp_server("accessgudid", config)

# Get available tools
tools = server.get_tools()
# Returns: lookup_device, parse_udi, get_device_history, get_device_snomed, list_implantable_devices

# Lookup a device by DI
result = await server.call_tool("lookup_device", {"di": "08717648200274"})
print(f"Device: {result['summary']['brandName']}")

# Parse a UDI string
result = await server.call_tool("parse_udi", {
    "udi": "(01)00844588012919(17)141231(10)A213B1"
})
print(f"Parsed DI: {result['parsed']['di']}")

# List implantable devices
result = await server.call_tool("list_implantable_devices", {
    "from_date": "2024-01-01",
    "max_records": 50
})
print(f"Found {result['count']} implantable devices")

Running MCP Servers via stdio

# Run OpenFDA MCP server (set FDA_API_KEY for higher rate limits)
export FDA_API_KEY=your_key
python -m ashmatics_tools.mcp_servers.openfda

# Run AccessGUDID MCP server (set UMLS_API_KEY for SNOMED lookups)
export UMLS_API_KEY=your_key
python -m ashmatics_tools.mcp_servers.accessgudid

For detailed usage examples, see FDA API Usage Guide.

Search/RAG Module

The search module provides RAG (Retrieval-Augmented Generation) strategies for building AI-powered search applications.

Key Features

• RAG Strategies: SimpleRAG and MultiQueryRAG with streaming support
• Context Window Management: Automatic fitting of sources to model context limits
• MCP Tool Definitions: Generic tool schemas for agent integration
• LLM Streaming: SSE and NDJSON streaming support across all LLM providers

Simple RAG Query

from ashmatics_tools.llm import create_llm_client, AzureOpenAIConfig
from ashmatics_tools.embedders import create_embedder
from ashmatics_tools.vector_stores import create_vector_store
from ashmatics_tools.search import create_search_strategy, RAGConfig

# Setup components
llm = create_llm_client("azure_openai", AzureOpenAIConfig(...))
embedder = create_embedder("azure")
vector_store = create_vector_store("cosmosdb", config)

# Create RAG strategy
rag = create_search_strategy(
    "simple_rag",
    llm=llm,
    vector_store=vector_store,
    embedder=embedder,
    config=RAGConfig(top_k=10, temperature=0.7)
)

# Query with answer generation
async with llm:
    result = await rag.query("What are ISO 42001 requirements?")
    print(result.answer)
    print(f"Sources: {len(result.sources)}")
    print(f"Tokens: {result.metrics.total_tokens}")

Multi-Query RAG (Query Expansion)

from ashmatics_tools.search import create_search_strategy
from ashmatics_tools.search.strategies import MultiQueryConfig

# Multi-query expands to multiple query variants for better coverage
config = MultiQueryConfig(
    top_k=10,
    num_query_variants=3,  # Generate 3 query variants
    rrf_k=60,              # RRF ranking parameter
)

rag = create_search_strategy(
    "multi_query_rag",
    llm=llm,
    vector_store=vector_store,
    embedder=embedder,
    config=config
)

async with llm:
    result = await rag.query("What is risk management in AI governance?")
    print(f"Expanded queries: {result.metadata.get('expanded_queries')}")
    print(result.answer)

Streaming RAG Responses

# Stream answer generation for real-time UI
async with llm:
    async for chunk in rag.stream_query("Explain AI governance controls"):
        if chunk.text:
            print(chunk.text, end="", flush=True)
        if chunk.is_final:
            print(f"\n\nSources: {len(chunk.sources)}")

Context Window Management

from ashmatics_tools.llm import ContextWindowManager, ModelContextLimits

# Create manager for GPT-4 Turbo
manager = ContextWindowManager(
    model_limits=ModelContextLimits.GPT4_TURBO(),
    reserved_output=2000
)

# Fit sources into available context
fitted_sources = manager.fit_sources(
    sources=search_results,
    query="What are ISO 42001 requirements?",
    system_prompt=system_prompt
)
print(f"Fitted {len(fitted_sources)} of {len(search_results)} sources")

MCP Tool Definitions

from ashmatics_tools.search.mcp_tools import (
    get_tool_definitions,
    export_tools_yaml,
    RAG_SEARCH_TOOL,
)

# Get all tool definitions for MCP server registration
tools = get_tool_definitions()
for tool in tools:
    print(f"{tool.name}: {tool.description}")

# Export as YAML for configuration
yaml_config = export_tools_yaml()

Document Enrichers

The enrichers module provides post-parsing content analysis for tables and extracted data.

Table Classification

from ashmatics_tools.enrichers import TableClassifier, TableCategory

# Initialize classifier
classifier = TableClassifier(provider="azure_openai")

# Classify tables from a parsed document
categories, tokens = await classifier.classify_tables(parsed_doc.tables)

for table, category in zip(parsed_doc.tables, categories):
    if category == TableCategory.PERFORMANCE_METRICS:
        # Extract metrics from performance tables
        pass
    elif category == TableCategory.COMPARISON:
        # Process comparison tables
        pass

Table Consolidation (Multi-page Tables)

from ashmatics_tools.enrichers import TableConsolidator

# Handle tables that span multiple PDF pages
consolidator = TableConsolidator(
    column_similarity_threshold=0.85,
    use_llm_validation=True
)

consolidated = await consolidator.consolidate_tables(
    parsed_doc.tables,
    parsed_doc.markdown
)

for table in consolidated:
    if table.merged_from:
        print(f"{table.table_id} merged from pages: {table.merged_from}")

Metrics Extraction

from ashmatics_tools.enrichers import MetricsExtractor, DomainKnowledgeProvider

# With optional domain knowledge injection
extractor = MetricsExtractor(domain_knowledge=my_provider)
result = await extractor.extract_from_tables(
    tables=performance_tables,
    document_text=section_text
)

for metric in result.performance_metrics:
    print(f"{metric.metric_name}: {metric.value} [{metric.ci_lower}, {metric.ci_upper}]")

Document Storage

Storage managers for document processing artifacts with manifest generation.

Figure Storage

from ashmatics_tools.document_storage import FigureStorageManager

# Filter small images (logos, icons) and save significant figures
manager = FigureStorageManager(min_size=200)
processed = manager.process_figures(parsed_doc.figures, parsed_doc.markdown)
saved = manager.save_figures(processed, output_dir / 'figures', doc_id)
# Creates figures_manifest.json with metadata

Table Storage

from ashmatics_tools.document_storage import TableStorageManager

# Save tables in dual format (Markdown + JSON)
manager = TableStorageManager()
stored = manager.save_tables(consolidated_tables, output_dir / 'tables', doc_id)
# Creates tables_manifest.json with metadata

Population Graph Engine

The population_graph module provides Bayesian network-based synthetic population generation for medical AI/ML testing and health technology assessment.

from ashmatics_tools.population_graph import PopulationGraphModel
from ashmatics_tools.population_graph.loaders import list_available_models

# List bundled model definitions
models = list_available_models()
# [{'name': 'LDCT_Baseline_v1', ...}, {'name': 'LDCT_Baseline_v1b', ...}]

# Load a model, build the network, validate, and sample
model = PopulationGraphModel("LDCT_Baseline_v1")
model.build()
assert model.validate()
samples = model.sample(1000)  # Returns pandas DataFrame

# Visualize the Bayesian network graph
img_path = model.visualize("/tmp/my_graph.png")

# Load from a custom JSON definition
from pathlib import Path
from ashmatics_tools.population_graph.loaders import load_from_json_file

definition = load_from_json_file(Path("my_custom_model.json"))
custom_model = PopulationGraphModel("custom", model_definition=definition)
custom_model.build()

# Learn structure from data with clinical priors (pgmpy 1.1 ExpertKnowledge)
model.learn_structure(patient_data, expert_knowledge={
    "required_edges": [["Sex", "Nodule"], ["Ethnicity", "Nodule"]],
    "forbidden_edges": [["kVP", "Ethnicity"]],
    "tier_ordering": [["Ethnicity", "Sex"], ["Nodule", "Manufacturer"]],
})

# Fit CPDs from observed data
model.fit_cpds(patient_data, estimator="mle")

Features

Modern HTTP Client (httpx)

Most HTTP communication uses httpx, providing:

• Native Type Annotations: Full type safety without separate stub packages
• Async/Await Support: Ready for async operations in performance-critical applications
• HTTP/2 Support: Modern protocol support for improved performance
• Familiar API: A synchronous surface close to requests for easy adoption

Secure by Default

• SSL Verification Enabled: All HTTP requests verify SSL certificates by default
• Explicit Opt-Out: SSL verification can only be disabled by explicitly passing verify=False to methods
• Security Warnings: Disabling SSL verification triggers warning logs

Flexible Configuration

• Environment Variables: Supports .env files for configuration
• Configurable Endpoints: All API endpoints configurable via environment or parameters
• Batch Processing: Configurable batch sizes for large dataset operations

MongoDB Integration (Optional)

• Optional Dependency: MongoDB support is optional via the [db-mongo] extra
• Abstract Base Classes: Extensible DocumentProcessor for custom document types
• Upsert Operations: Intelligent upsert with identifier-based conflict resolution

Comprehensive Error Handling

• Detailed Logging: Structured logging throughout all operations
• Graceful Failures: Proper error handling with informative messages
• Validation: Input validation and JSON compliance checking

Architecture

ashmatics-tools/
├── src/ashmatics_tools/
│   ├── __init__.py           # Public API exports
│   ├── chunkers/
│   │   ├── __init__.py
│   │   ├── azure_chunker.py
│   │   ├── base.py
│   │   ├── docling_chunker.py
│   │   └── simple_chunker.py
│   ├── document_storage/
│   │   ├── __init__.py
│   │   ├── figure_storage.py
│   │   └── table_storage.py
│   ├── embedders/
│   │   ├── __init__.py
│   │   ├── azure_embedder.py
│   │   ├── base.py
│   │   ├── factory.py
│   │   └── openai_embedder.py
│   ├── embedding/
│   │   ├── __init__.py
│   │   ├── base.py
│   │   ├── mongodb_pipeline.py
│   │   └── specialized/
│   ├── enrichers/
│   │   ├── __init__.py
│   │   ├── table_classifier.py
│   │   ├── table_consolidator.py
│   │   ├── metrics_extractor.py
│   │   └── training_data_extractor.py
│   ├── graphql/
│   │   ├── __init__.py
│   │   └── client.py
│   ├── llm/
│   │   ├── __init__.py
│   │   ├── azure_openai.py
│   │   ├── azure_ai_foundry.py
│   │   ├── base.py
│   │   ├── context.py          # ContextWindowManager, ModelContextLimits
│   │   ├── factory.py
│   │   ├── huggingface.py
│   │   ├── llamacpp.py
│   │   ├── ollama.py
│   │   ├── openai.py
│   │   └── retry.py            # RetryConfig, call_with_backoff
│   ├── external_apis/
│   │   ├── __init__.py
│   │   ├── base.py
│   │   ├── factory.py
│   │   ├── openfda/
│   │   │   ├── __init__.py
│   │   │   ├── client.py
│   │   │   └── config.py
│   │   └── accessgudid/
│   │       ├── __init__.py
│   │       ├── client.py
│   │       └── config.py
│   ├── mcp_servers/
│   │   ├── __init__.py
│   │   ├── base.py
│   │   ├── factory.py
│   │   ├── openfda/
│   │   │   ├── __init__.py
│   │   │   ├── config.py
│   │   │   └── server.py
│   │   └── accessgudid/
│   │       ├── __init__.py
│   │       ├── config.py
│   │       └── server.py
│   ├── ontology/
│   │   ├── __init__.py
│   │   ├── categories/
│   │   ├── core/
│   │   ├── data/
│   │   └── terms/
│   ├── population_graph/
│   │   ├── __init__.py
│   │   ├── engine.py          # PopulationGraphModel class
│   │   ├── schema.py          # ModelDefinition, CPDDefinition dataclasses
│   │   ├── loaders.py         # JSON file/dict model loading
│   │   ├── evidence.py        # Contingency table generation
│   │   └── model_definitions/ # Bundled JSON model configs
│   │       ├── ldct_baseline_v1.json
│   │       └── ldct_baseline_v1b.json
│   ├── parsers/
│   │   ├── __init__.py
│   │   ├── base.py
│   │   ├── docling_parser.py
│   │   ├── factory.py
│   │   ├── llama_parser.py
│   │   └── simple_parser.py
│   ├── processors/
│   │   ├── __init__.py
│   │   └── base.py
│   ├── search/
│   │   ├── __init__.py
│   │   ├── base.py
│   │   ├── exceptions.py
│   │   ├── factory.py
│   │   ├── mcp_tools.py         # RAG/semantic/multi-query tool definitions
│   │   ├── prompts.py
│   │   └── strategies/          # SimpleRAGStrategy, MultiQueryRAGStrategy
│   ├── storage/
│   │   ├── __init__.py
│   │   ├── base.py
│   │   ├── factory.py
│   │   ├── adls_store.py
│   │   └── minio_store.py
│   ├── utils/
│   │   ├── __init__.py
│   │   ├── export_utils.py
│   │   ├── import_utils.py
│   │   └── schema_utils.py
│   └── vector_stores/
│       ├── __init__.py
│       ├── base.py
│       ├── cosmosdb_store.py
│       ├── factory.py
│       ├── pgvector_store.py
│       └── qdrant_store.py
├── tests/                   # Test suite
├── pyproject.toml          # Package configuration
└── README.md

Dependencies

Core Dependencies

The core install is deliberately lightweight (no torch/CUDA). Key packages:

• httpx>=0.27.0 — primary HTTP client (native type annotations, async support)
• pydantic>=2.0.0 — schema and configuration models
• ashmatics-datamodels>=0.7.0,<0.8.0 — shared Ashmatics Pydantic contracts (on PyPI)
• pandas>=2.1.0, numpy>=1.24.0, openpyxl>=3.1.2 — data handling
• openai>=1.0.0, tiktoken>=0.5.0 — embeddings and tokenization
• graphql-core>=3.2.0 — GraphQL support
• pgmpy>=1.1, pygraphviz>=1.14 — population graph engine
• Pillow>=10.0.0, PyPDF2>=3.0.0 — lightweight image/PDF handling
• python-dotenv, pyyaml, rich, tqdm, markdown, requests, aiohttp

See pyproject.toml for the authoritative, version-pinned list.

Optional Dependencies (extras)

Heavy or backend-specific packages are grouped into extras so you install only what you use:

Example: pip install "ashmatics-tools[api]"

Development

Running Tests

# Run all tests
pytest

# Run with coverage
pytest --cov=ashmatics_tools --cov-report=html

# Run specific test file
pytest tests/test_import_utils.py

Code Quality

# Lint with ruff
ruff check src/

# Format with ruff
ruff format src/

# Type check with mypy
mypy src/

Environment Variables

# GraphQL/Hasura Configuration
HASURA_GRAPHQL_ENDPOINT=https://kb-api.ashmatics.com/v1/graphql
HASURA_ADMIN_SECRET=your-admin-secret-here

# MongoDB Configuration (optional)
MONGODB_CONNECTION_STRING=mongodb://localhost:27017
MONGODB_DATABASE=ashmatics_kb

Python Version Support

• Minimum: Python 3.11
• Tested: Python 3.11, 3.12
• Recommended: Python 3.12+

License

Proprietary — Copyright © 2025 Asher Informatics PBC. All Rights Reserved. See the LICENSE file for details.

Contributing

This is a private package for Ashmatics internal use. For questions or issues, please contact the development team.

Version History

The full, detailed release history lives in CHANGELOG.md — that is the source of truth for what changed in each version. This README documents current functionality only.

Complete Document Processing Pipeline

The package now provides a complete end-to-end pipeline for document processing:

from ashmatics_tools import (
    create_parser,        # Parse documents (PDF, DOCX, etc.)
    create_chunker,       # Chunk into manageable pieces
    create_embedder,      # Generate embeddings
    create_vector_store   # Store and search vectors
)

# 1. Parse document
parser = create_parser("docling")
parsed_doc = await parser.parse_file("document.pdf")

# 2. Chunk document
chunker = create_chunker(strategy="docling")
chunks = await chunker.chunk_document(
    content=parsed_doc.markdown,
    title="Document Title",
    source="document.pdf"
)

# 3. Generate embeddings
embedder = create_embedder(provider="azure")
await embedder.initialize()
embedded_chunks = await embedder.embed_chunks(chunks)

# 4. Store in vector database
vector_store = create_vector_store(provider="cosmosdb")
success, failed = await vector_store.store_embeddings_batch(embedded_chunks)

# 5. Search
query_embedding = await embedder.generate_embedding("search query")
results = await vector_store.similarity_search(query_embedding, top_k=10)

Module Overview

Parsers (ashmatics_tools.parsers)

Document parsing with multiple backends:

• SimpleParser: Basic fallback parser
• DoclingParser: Advanced PDF parsing with tables/figures
• LlamaParser: LlamaParse cloud service integration
• Factory: create_parser(provider)

Chunkers (ashmatics_tools.chunkers)

Document chunking strategies:

• SimpleChunker: Paragraph-based chunking
• AzureChunker: Azure-compatible with tiktoken
• DoclingChunker: Token-aware semantic chunking
• Factory: create_chunker(strategy)

Embedders (ashmatics_tools.embedders)

Embedding generation:

• AzureEmbedder: Azure OpenAI embeddings
• OpenAIEmbedder: OpenAI embeddings
• Factory: create_embedder(provider)

Embedding Pipelines (ashmatics_tools.embedding)

MongoDB-based embedding workflows:

• MongoDBEmbeddingPipeline: Generic pipeline
• Specialized Pipelines: Framework, use cases, cards

Vector Stores (ashmatics_tools.vector_stores)

Vector database integrations:

• CosmosDBVectorStore: Azure CosmosDB with MongoDB vCore API
• PgVectorStore: PostgreSQL with pgvector extension
• QdrantVectorStore: Qdrant vector database
• Factory: create_vector_store(provider)

Storage Backends (ashmatics_tools.storage)

Cloud-agnostic storage abstraction:

• ADLSStorageClient: Azure Data Lake Storage Gen2 with dual auth (connection string or DefaultAzureCredential)
• MinIOStorageClient: MinIO object storage (S3-compatible)
• S3StorageClient: AWS S3 (reserved for future implementation)
• Factory: create_storage_client(provider, config)
• Features: Async API, buffered and streaming reads/writes, glob pattern matching, metadata operations

LLM Clients (ashmatics_tools.llm)

Unified interface for language model providers:

• AzureOpenAIClient: Azure OpenAI Service
• OpenAIClient: OpenAI direct API
• HuggingFaceInferenceClient: HuggingFace Inference API (requires [huggingface] extra)
• HuggingFaceLocalClient: Local HuggingFace models (requires [huggingface] extra)
• AzureAIFoundryClient: Full Azure AI Foundry model catalog (requires [azure-ai] extra)
• OllamaClient: Local/ACA/K8s Ollama inference with SDK (requires [ollama] extra) - embeddings, vision, tools, model management
• Factory: create_llm_client(provider, config) with plugin registry
• Features: Async-first API, unified completion interface, cost tracking, plugin registry, extensible via register_llm_provider()

Ontology Services (ashmatics_tools.ontology)

Medical ontology management and term services:

• TermResolver: MongoDB-based term lookup and resolution
• CategoryManager: Hierarchical category management for document tagging
• BioPortalClient: External ontology validation via NCBO BioPortal API (SNOMED CT, RADLEX, LOINC, NCIT)
• AshmaticsOntology: Custom ASHMATICS domain-specific ontology for medical imaging AI concepts
• Features: Async API, comprehensive schema validation, integration with external ontologies

External APIs (ashmatics_tools.external_apis)

Clients for external data sources with robust error handling:

• OpenFDAClient: US FDA Open Data Portal (open.fda.gov) integration
• AccessGUDIDClient: NIH/FDA Global Unique Device Identification Database (accessgudid.nlm.nih.gov) integration
• BaseAPIClient: Abstract base for creating custom API clients
• OpenFDA Endpoints: Device 510(k), adverse events, recalls, drug labels, FAERS, enforcement actions
• AccessGUDID Endpoints: Device lookup, UDI parsing, device history, SNOMED mappings, implantable device listings
• Factory: create_api_client(provider, config) with plugin registry
• Features: Async API, retry with exponential backoff, client-side rate limiting, automatic pagination
• Query Syntax: Support for field search, date ranges, boolean operators, wildcards (OpenFDA)
• Extensibility: Register custom providers via register_api_provider() for Census, CMS, etc.

MCP Servers (ashmatics_tools.mcp_servers)

Model Context Protocol servers for LLM integration:

• OpenFDAMCPServer: Expose OpenFDA API as LLM tools (search_devices, search_drugs, count_by_field)
• AccessGUDIDMCPServer: Expose AccessGUDID API as LLM tools (lookup_device, parse_udi, get_device_history, get_device_snomed, list_implantable_devices)
• BaseMCPServer: Abstract base for creating MCP tool servers
• Factory: create_mcp_server(name, config) with plugin registry
• Features: JSON Schema validation, response formatting, error handling, streaming support
• Use Case: Thin adapter layer between LLMs and external data sources
• Extensibility: Register custom servers via register_mcp_server()
• Stdio Runner: Run servers via python -m ashmatics_tools.mcp_servers.{openfda,accessgudid}

Search/RAG (ashmatics_tools.search)

RAG (Retrieval-Augmented Generation) strategies for AI-powered search:

• SimpleRAGStrategy: Basic RAG flow with embed → retrieve → generate
• MultiQueryRAGStrategy: Query expansion with parallel retrieval and RRF ranking
• RAGConfig: Configuration for top_k, temperature, max_tokens, system_prompt
• RAGResult: Answer with sources, metrics, and metadata
• RAGStreamChunk: Streaming response chunks with partial sources
• Factory: create_search_strategy(name, llm, vector_store, embedder, config) with plugin registry
• MCP Tools: Generic tool definitions (rag_search, semantic_search, multi_query_search)
• Context Management: ContextWindowManager for automatic source fitting
• Model Presets: ModelContextLimits for GPT-4, Claude, Llama, Mistral
• Features: Async-first API, streaming support, ADR-045 governance metadata

Enrichers (ashmatics_tools.enrichers)

Post-parsing document enrichment for tables and extracted data:

• TableClassifier: LLM-based table categorization by content type
• TableConsolidator: Multi-page table merge with heuristics and LLM validation
• MetricsExtractor: Performance metrics extraction with statistical context
• TrainingDataExtractor: AI/ML training dataset characteristics extraction
• DomainKnowledgeProvider: Abstract base for domain-specific context injection
• Categories: COMPARISON, PERFORMANCE_METRICS, STUDY_DESIGN, TECHNICAL_SPECS, DEMOGRAPHICS, etc.
• Features: Handles PDF parser fragmentation, continuation markers, column similarity matching

Population Graph (ashmatics_tools.population_graph)

Bayesian network population generation engine:

• PopulationGraphModel: Data-driven BN construction, sampling, and visualization
• ModelDefinition: Dataclass schema for JSON model specifications (edges, state names, CPDs)
• Loaders: Load model definitions from JSON files, dicts, or custom directories
• Evidence: Contingency table generation and reference data loading
• ExpertKnowledge: pgmpy 1.1 hooks for clinical prior-constrained structure learning
• Bundled Models: LDCT_Baseline_v1 (8-node, 2-category ethnicity), LDCT_Baseline_v1b (9-node, 5-category ethnicity with age)
• Features: Forward/Gibbs sampling, MLE/Bayesian parameter estimation, graphviz visualization, JSON roundtrip serialization

Document Storage (ashmatics_tools.document_storage)

Artifact storage managers for document processing outputs:

• FigureStorageManager: Figure filtering, PNG conversion, content-addressed storage
• TableStorageManager: Dual-format (Markdown + JSON) table storage
• ProcessedFigure: Dataclass for processed figures with metadata
• StoredTable: Dataclass for stored tables with file paths
• Features: Automatic manifest generation, content hashing, size filtering