API Reference¶

This page documents the complete FairSense-AgentiX API, including the Python API and REST API endpoints.

Python API¶

The Python API provides a clean, Pydantic-based interface for bias detection and risk assessment. All classes are FastAPI-compatible and type-safe.

FairSense Class¶

The main entry point for programmatic bias detection and risk assessment.

FairSense ¶

High-level API for bias detection and AI risk assessment.

This class provides a clean interface for analyzing text, images, and AI deployment scenarios. It manages graph initialization internally and can be reused for multiple analyses (recommended for performance).

Configuration

Configuration is controlled via environment variables or .env file. Set these before initializing FairSense:

- FAIRSENSE_LLM_PROVIDER: "openai", "anthropic", "local", "fake"
- FAIRSENSE_LLM_API_KEY: Your API key (required for real analysis)
- FAIRSENSE_LLM_MODEL_NAME: Model to use (default per provider)
- FAIRSENSE_OCR_TOOL: "tesseract", "paddleocr", "fake" (default "tesseract")
- FAIRSENSE_CAPTION_MODEL: "blip", "blip2", "fake" (default "blip")
- FAIRSENSE_TELEMETRY_ENABLED: "true" or "false" (default "true")

See Settings documentation for all available configuration options.

Per-Request Options

All analysis methods accept optional **options kwargs:

Text/Image Bias Detection: - temperature: LLM temperature (0.0-1.0, default 0.3) - max_tokens: Maximum LLM response tokens (default 2000) - ocr_language: OCR language code (default "eng") - caption_max_length: Max caption length (default 100)

Risk Assessment: - top_k: Number of risks to retrieve (default 5, max 100) - rmf_per_risk: RMF recommendations per risk (default 3, max 20)

Examples:

Basic usage:

>>> fs = FairSense()
>>> result = fs.analyze_text("Job posting text")
>>> print(result.bias_detected, result.risk_level)

With configuration:

>>> result = fs.analyze_text(
...     "Job posting",
...     temperature=0.2,  # Lower for consistency
...     max_tokens=3000,  # More detailed analysis
... )

Batch processing:

>>> fs = FairSense()  # Initialize once
>>> for text in texts:
...     result = fs.analyze_text(text)
...     print(f"{result.metadata.run_id}: {result.status}")

FastAPI integration:

>>> from fastapi import FastAPI
>>> app = FastAPI()
>>> fs = FairSense()
>>> @app.post("/analyze", response_model=BiasResult)
>>> def analyze(text: str):
...     return fs.analyze_text(text)

init ¶

__init__()

Initialize FairSense API.

This creates and compiles the orchestrator graph. Graph compilation is expensive (~1-2s), so reuse this instance for multiple analyses.

Performance Note: Models are preloaded at module import time, so FairSense initialization is fast and all analysis calls are instant.

analyze_text ¶

analyze_text(text, run_id=None, **options)

Detect bias in text.

Analyzes text for various forms of bias including gender, age, racial, disability, and socioeconomic bias. Uses LLM to identify bias instances, generate explanations, and produce color-coded HTML output.

Parameters:

Name	Type	Description	Default
`text`	`str`	Text to analyze for bias	required
`run_id`	`str \| None`	Pre-generated run ID for tracing (if None, auto-generated)	`None`
`**options`	`Any`	Optional configuration overrides: - temperature: LLM temperature (0.0-1.0, default 0.3) - max_tokens: Maximum response tokens (default 2000)	`{}`

Returns:

Type	Description
`BiasResult`	Complete analysis result with bias detection, instances, HTML output, execution metadata, and any errors/warnings

Examples:

>>> fs = FairSense()
>>> result = fs.analyze_text("We need young rockstar developers")
>>> if result.bias_detected:
...     print(f"Risk: {result.risk_level}")
...     for inst in result.bias_instances or []:
...         print(f"  {inst['type']}: {inst['text_span']}")

analyze_image ¶

analyze_image(image, run_id=None, **options)

Detect bias in images.

Analyzes images by extracting text (OCR), generating captions, and performing bias analysis on the combined text. Uses parallel execution for OCR and captioning for optimal performance.

Parameters:

Name	Type	Description	Default
`image`	`bytes \| Path`	Image data as bytes or path to image file	required
`run_id`	`str \| None`	Pre-generated run ID for tracing (if None, auto-generated)	`None`
`**options`	`Any`	Optional configuration overrides: - ocr_language: OCR language code (default "eng") - caption_max_length: Max caption length (default 100) - temperature: LLM temperature (0.0-1.0, default 0.3)	`{}`

Returns:

Type	Description
`BiasResult`	Complete analysis result including OCR text, caption, bias detection, HTML output, execution metadata, and any errors/warnings

Examples:

>>> fs = FairSense()
>>> result = fs.analyze_image(Path("hiring_ad.png"))
>>> print(f"OCR: {result.ocr_text}")
>>> print(f"Caption: {result.caption_text}")
>>> if result.bias_detected:
...     print(f"Bias found: {result.risk_level}")

assess_risk ¶

assess_risk(scenario, run_id=None, **options)

Assess AI risks in deployment scenario.

Performs semantic search against 280+ AI risks from the NIST AI Risk Management Framework. Retrieves relevant risks and corresponding mitigation recommendations.

Parameters:

Name	Type	Description	Default
`scenario`	`str`	AI deployment scenario description	required
`run_id`	`str \| None`	Pre-generated run ID for tracing (if None, auto-generated)	`None`
`**options`	`Any`	Optional configuration overrides: - top_k: Number of risks to retrieve (default 5, max 100) - rmf_per_risk: Recommendations per risk (default 3, max 20)	`{}`

Returns:

Type	Description
`RiskResult`	Complete risk assessment including identified risks, RMF recommendations, HTML table, CSV export path, execution metadata, and any errors/warnings

Examples:

>>> fs = FairSense()
>>> result = fs.assess_risk(
...     "Facial recognition for employee monitoring", top_k=10
... )
>>> print(f"Found {len(result.risks)} risks")
>>> for risk in result.risks:
...     print(f"  {risk['name']}: {risk['score']:.3f}")
>>> print(f"CSV exported to: {result.csv_path}")

Result Objects¶

Result objects are Pydantic models that provide type-safe access to analysis outputs.

BiasResult¶

Returned by analyze_text() and analyze_image() methods.

BiasResult ¶

Bases: BaseModel

Result from text or image bias analysis.

This model is compatible with FastAPI (Pydantic-based) and provides attribute-style access to all results and metadata.

Attributes:

Name	Type	Description
`status`	`str`	Execution status: "success", "failed", or "partial"
`bias_detected`	`bool`	Whether bias was detected (None if using fake LLM)
`risk_level`	`str`	Risk level: "low", "medium", "high" (None if using fake LLM)
`bias_analysis`	`BiasAnalysisOutput \| str \| None`	Structured bias analysis (BiasAnalysisOutput) from real LLM, or string summary from fake LLM, or None if failed
`bias_instances`	`list[dict] \| None`	List of bias instances with type, severity, text_span, explanation. None if using fake LLM or if no structured output
`summary`	`str \| None`	Optional condensed summary of findings
`highlighted_html`	`str \| None`	HTML with bias spans color-coded by type
`ocr_text`	`str \| None`	Text extracted via OCR (image workflow only)
`caption_text`	`str \| None`	Generated image caption (image workflow only)
`merged_text`	`str \| None`	Combined OCR + caption text (image workflow only)
`metadata`	`ResultMetadata`	Execution metadata (timing, routing, quality scores)
`errors`	`list[str]`	Error messages (empty if successful)
`warnings`	`list[str]`	Warning messages (empty if none)

Examples:

>>> result = analyze_text("Job posting")
>>> if result.status == "success":
...     print(f"Bias detected: {result.bias_detected}")
...     print(f"Risk level: {result.risk_level}")
...     if result.bias_instances:
...         for inst in result.bias_instances:
...             print(f"  - {inst['type']}: {inst['text_span']}")
>>> print(f"Execution time: {result.metadata.execution_time_seconds:.2f}s")
>>> print(f"Router confidence: {result.metadata.router_confidence:.2%}")

status `class-attribute` `instance-attribute` ¶

status = Field(description='Execution status')

bias_detected `class-attribute` `instance-attribute` ¶

bias_detected = Field(
    None, description="Whether bias was detected"
)

risk_level `class-attribute` `instance-attribute` ¶

risk_level = Field(None, description='Overall risk level')

bias_analysis `class-attribute` `instance-attribute` ¶

bias_analysis = Field(
    None, description="Structured or string bias analysis"
)

bias_instances `class-attribute` `instance-attribute` ¶

bias_instances = Field(
    None,
    description="Individual bias instances with details",
)

summary `class-attribute` `instance-attribute` ¶

summary = Field(
    None, description="Optional condensed summary"
)

highlighted_html `class-attribute` `instance-attribute` ¶

highlighted_html = Field(
    None,
    description="HTML with color-coded bias highlighting",
)

ocr_text `class-attribute` `instance-attribute` ¶

ocr_text = Field(
    None, description="OCR extracted text (image only)"
)

caption_text `class-attribute` `instance-attribute` ¶

caption_text = Field(
    None, description="Generated caption (image only)"
)

merged_text `class-attribute` `instance-attribute` ¶

merged_text = Field(
    None,
    description="Combined OCR + caption text (image only)",
)

image_base64 `class-attribute` `instance-attribute` ¶

image_base64 = Field(
    None,
    description="Base64-encoded image with data URL for UI display (VLM workflow only)",
)

metadata `class-attribute` `instance-attribute` ¶

metadata = Field(description='Execution metadata')

errors `class-attribute` `instance-attribute` ¶

errors = Field(
    default_factory=list, description="Error messages"
)

warnings `class-attribute` `instance-attribute` ¶

warnings = Field(
    default_factory=list, description="Warning messages"
)

RiskResult¶

Returned by assess_risk() method.

RiskResult ¶

Bases: BaseModel

Result from AI risk assessment.

This model is compatible with FastAPI (Pydantic-based) and provides attribute-style access to all results and metadata.

Attributes:

Name	Type	Description
`status`	`str`	Execution status: "success", "failed", or "partial"
`embedding`	`list[float] \| None`	Vector embedding of scenario (384 dimensions for default model)
`risks`	`list[dict[str, Any]]`	Top-k AI risks found via semantic search, each with: - name: str (risk name) - description: str (risk description) - score: float (similarity score 0.0-1.0) - category: str (risk category)
`rmf_recommendations`	`dict[str, list[dict[str, Any]]]`	AI-RMF recommendations per risk, keyed by risk ID
`html_table`	`str \| None`	Formatted HTML table of risks + recommendations
`csv_path`	`str \| None`	Path to exported CSV file
`metadata`	`ResultMetadata`	Execution metadata (timing, routing, quality scores)
`errors`	`list[str]`	Error messages (empty if successful)
`warnings`	`list[str]`	Warning messages (empty if none)

Examples:

>>> result = assess_risk("AI deployment scenario")
>>> if result.status == "success":
...     print(f"Found {len(result.risks)} risks")
...     for risk in result.risks[:3]:
...         print(f"  - {risk['name']}: {risk['score']:.3f}")
>>> print(f"Execution time: {result.metadata.execution_time_seconds:.2f}s")

status `class-attribute` `instance-attribute` ¶

status = Field(description='Execution status')

embedding `class-attribute` `instance-attribute` ¶

embedding = Field(
    None, description="Vector embedding of scenario"
)

risks `class-attribute` `instance-attribute` ¶

risks = Field(
    default_factory=list,
    description="Top-k AI risks from semantic search",
)

rmf_recommendations `class-attribute` `instance-attribute` ¶

rmf_recommendations = Field(
    default_factory=dict,
    description="AI-RMF recommendations per risk",
)

html_table `class-attribute` `instance-attribute` ¶

html_table = Field(
    None,
    description="Formatted HTML table of risks + recommendations",
)

csv_path `class-attribute` `instance-attribute` ¶

csv_path = Field(
    None, description="Path to exported CSV file"
)

metadata `class-attribute` `instance-attribute` ¶

metadata = Field(description='Execution metadata')

errors `class-attribute` `instance-attribute` ¶

errors = Field(
    default_factory=list, description="Error messages"
)

warnings `class-attribute` `instance-attribute` ¶

warnings = Field(
    default_factory=list, description="Warning messages"
)

ResultMetadata¶

Execution metadata included in all results.

ResultMetadata ¶

Bases: BaseModel

Metadata about analysis execution.

Attributes:

Name	Type	Description
`run_id`	`str`	Unique identifier for tracing this execution
`workflow_id`	`str`	Which workflow was executed ("bias_text", "bias_image", "risk")
`execution_time_seconds`	`float`	Total execution time in seconds
`router_reasoning`	`str`	Human-readable explanation of why this workflow was selected
`router_confidence`	`float`	Router's confidence in workflow selection (0.0-1.0)
`refinement_count`	`int`	Number of refinement iterations performed (0 if none)
`preflight_score`	`float \| None`	Pre-execution quality score (None if not evaluated, Phase 7+)
`posthoc_score`	`float \| None`	Post-execution quality score (None if not evaluated, Phase 7+)

run_id `class-attribute` `instance-attribute` ¶

run_id = Field(
    description="Unique execution identifier for tracing"
)

workflow_id `class-attribute` `instance-attribute` ¶

workflow_id = Field(description='Workflow executed')

execution_time_seconds `class-attribute` `instance-attribute` ¶

execution_time_seconds = Field(
    description="Total execution time"
)

router_reasoning `class-attribute` `instance-attribute` ¶

router_reasoning = Field(
    description="Why this workflow was selected"
)

router_confidence `class-attribute` `instance-attribute` ¶

router_confidence = Field(
    ge=0.0,
    le=1.0,
    description="Router confidence (0.0-1.0)",
)

refinement_count `class-attribute` `instance-attribute` ¶

refinement_count = Field(
    ge=0, description="Number of refinement iterations"
)

preflight_score `class-attribute` `instance-attribute` ¶

preflight_score = Field(
    None,
    ge=0.0,
    le=1.0,
    description="Pre-execution quality score (Phase 7+)",
)

posthoc_score `class-attribute` `instance-attribute` ¶

posthoc_score = Field(
    None,
    ge=0.0,
    le=1.0,
    description="Post-execution quality score (Phase 7+)",
)

REST API¶

The FastAPI backend exposes a production-ready REST API for remote analysis.

Base URL¶

http://localhost:8000

All endpoints are versioned under /v1/.

Endpoints¶

Health Check¶

Check if the server is ready to accept requests.

Endpoint: GET /v1/health

Response:

{
  "status": "ok"
}

cURL Example:

curl http://localhost:8000/v1/health

Analyze Text/Content (JSON)¶

Analyze text or base64-encoded content with automatic input type detection.

Endpoint: POST /v1/analyze

Request Body:

{
  "content": "Text to analyze for bias...",
  "input_type": "bias_text",  // Optional: "bias_text", "bias_image", "risk", null (auto-detect)
  "options": {                 // Optional per-request overrides
    "temperature": 0.3,
    "max_tokens": 2000
  }
}

Response: Returns AnalyzeResponse object

href="#__codelineno-4-1">{ "workflow_id": "bias_text", "run_id": "a1b2c3d4-...", "bias_result": { "status": "success", "bias_detected": true, "risk_level": "medium", "bias_instances": [ { "type": "age", "severity": "high", "text_span": "young, energetic", "explanation": "Age-related descriptors may discourage older applicants", "start_index": 14, "end_index": 30 } ], "summary": "The text contains age-related bias...", "highlighted_html": "<span class='bias-age'>young, energetic</span>...", "metadata": { "run_id": "a1b2c3d4-...", "workflow_id": "bias_text", "execution_time_seconds": 2.34, "router_confidence": 0.95, "refinement_count": 1 }, "errors": [], "warnings": [] }, "risk_result": null, "metadata": { /* ... */ } }

cURL Example:

curl -X POST http://localhost:8000/v1/analyze \
  -H "Content-Type: application/json" \
  -d '{
    "content": "We need young, energetic developers for our startup team.",
    "options": {"temperature": 0.2}
  }'

Analyze File Upload¶

Analyze uploaded files (images, text files, CSV).

Endpoint: POST /v1/analyze/upload

Request: multipart/form-data

file: File to analyze (required)
input_type: Optional workflow hint ("image", "text", "csv")

Response: Same structure as /v1/analyze

cURL Example:

# Analyze an image
curl -X POST http://localhost:8000/v1/analyze/upload \
  -F "file=@team_photo.jpg"

# Analyze a text file with explicit type
curl -X POST http://localhost:8000/v1/analyze/upload \
  -F "file=@job_posting.txt" \
  -F "input_type=text"

Start Analysis (for WebSocket streaming)¶

Start an analysis and return run_id immediately for WebSocket connection. Use this to receive real-time agent telemetry.

Endpoint: POST /v1/analyze/start

Request Body: Same as /v1/analyze

Response:

{
  "run_id": "a1b2c3d4-...",
  "status": "started",
  "message": "Analysis started. Connect to WebSocket to receive events."
}

Usage Pattern:

import requests
import websockets
import asyncio
import json

async def analyze_with_streaming():
    # 1. Start analysis
    response = requests.post(
        "http://localhost:8000/v1/analyze/start",
        json={"content": "Text to analyze..."}
    )
    run_id = response.json()["run_id"]

    # 2. Connect to WebSocket immediately
    uri = f"ws://localhost:8000/v1/stream/{run_id}"
    async with websockets.connect(uri) as ws:
        async for message in ws:
            event = json.loads(message)
            print(f"[{event['event']}] {event['context'].get('message', '')}")

            if event["event"] == "analysis_complete":
                result = event["context"]["result"]
                print(f"\n✅ Complete: {result['bias_result']['summary']}")
                break

asyncio.run(analyze_with_streaming())

Start File Upload Analysis (for WebSocket streaming)¶

Upload a file and return run_id for WebSocket streaming.

Endpoint: POST /v1/analyze/upload/start

Request: multipart/form-data (same as /v1/analyze/upload)

Response: Same as /v1/analyze/start

Batch Analysis¶

Submit multiple items for batch processing.

Endpoint: POST /v1/batch

Request Body:

{
  "items": [
    {
      "content": "First text to analyze...",
      "input_type": "bias_text",
      "options": {"temperature": 0.3}
    },
    {
      "content": "Second text to analyze...",
      "input_type": null,  // Auto-detect
      "options": {}
    }
  ]
}

Response: Returns BatchStatus with job_id

{
  "job_id": "batch-abc123",
  "status": "pending",  // "pending" | "running" | "completed" | "failed"
  "total": 2,
  "completed": 0,
  "errors": [],
  "results": []
}

Status Code: 202 Accepted

Get Batch Status¶

Poll for batch job progress and results.

Endpoint: GET /v1/batch/{job_id}

Response:

{
  "job_id": "batch-abc123",
  "status": "completed",
  "total": 2,
  "completed": 2,
  "errors": [],
  "results": [
    { /* AnalyzeResponse for item 1 */ },
    { /* AnalyzeResponse for item 2 */ }
  ]
}

cURL Example:

# Submit batch job
JOB_ID=$(curl -X POST http://localhost:8000/v1/batch \
  -H "Content-Type: application/json" \
  -d '{"items": [{"content": "Text 1"}, {"content": "Text 2"}]}' \
  | jq -r '.job_id')

# Poll for completion
while true; do
  STATUS=$(curl -s http://localhost:8000/v1/batch/$JOB_ID | jq -r '.status')
  echo "Status: $STATUS"
  [[ "$STATUS" == "completed" || "$STATUS" == "failed" ]] && break
  sleep 2
done

# Get final results
curl http://localhost:8000/v1/batch/$JOB_ID | jq '.results'

Shutdown Server¶

Gracefully shutdown the backend server (used by UI shutdown button).

Endpoint: POST /v1/shutdown

Response:

{
  "status": "shutting_down",
  "message": "Server will shutdown in 1 second"
}

cURL Example:

curl -X POST http://localhost:8000/v1/shutdown

WebSocket Protocol¶

The WebSocket API provides real-time streaming of agent telemetry events during analysis.

Connection URL¶

ws://localhost:8000/v1/stream/{run_id}

Connect to this WebSocket after calling /v1/analyze/start to receive live events.

Event Structure¶

All events follow this JSON structure:

{
  "run_id": "a1b2c3d4-...",
  "timestamp": 1234567890.123,
  "event": "event_name",
  "level": "info",  // "info" | "warning" | "error"
  "context": {
    "message": "Human-readable event description",
    "phase": "planning",  // Current workflow phase
    // ... additional event-specific fields
  }
}

Event Types¶

Workflow Events¶

Event	Description	Context Fields
`workflow_start`	Analysis begins	`input_type`, `workflow_id`
`phase_transition`	Agent moves to new phase	`from_phase`, `to_phase`, `phase_number`
`tool_call_start`	Tool execution begins	`tool_name`, `inputs`
`tool_call_end`	Tool execution completes	`tool_name`, `outputs`, `duration_ms`
`llm_call_start`	LLM request begins	`model`, `temperature`, `max_tokens`
`llm_call_end`	LLM response received	`model`, `tokens_used`, `duration_ms`
`refinement_start`	Refinement iteration begins	`iteration_number`, `reason`
`refinement_end`	Refinement iteration completes	`iteration_number`, `improved`
`analysis_complete`	Analysis finished successfully	`result` (full result object)
`analysis_error`	Analysis failed	`error_type`, `error_message`

Phase Events¶

Phase	Event Name	Description
Planning	`phase_planning`	Agent planning analysis strategy
Tool Selection	`phase_tool_selection`	Selecting appropriate tools
Tool Execution	`phase_tool_execution`	Running OCR, captioning, embeddings
Evidence Synthesis	`phase_synthesis`	Combining tool outputs
Evaluation	`phase_evaluation`	Quality assessment
Refinement	`phase_refinement`	Iterative improvement

Connection Lifecycle¶

Start Analysis:

response = requests.post(
    "http://localhost:8000/v1/analyze/start",
    json={"content": "..."}
)
run_id = response.json()["run_id"]

Connect to WebSocket:

uri = f"ws://localhost:8000/v1/stream/{run_id}"
async with websockets.connect(uri) as ws:
    # Start receiving events

Receive Events:

async for message in ws:
    event = json.loads(message)

    # Handle different event types
    if event["event"] == "tool_call_start":
        print(f"Running {event['context']['tool_name']}...")

    elif event["event"] == "analysis_complete":
        result = event["context"]["result"]
        break  # Analysis done

    elif event["event"] == "analysis_error":
        print(f"Error: {event['context']['error_message']}")
        break

Disconnection:
WebSocket closes automatically after analysis_complete or analysis_error
Client can disconnect early without affecting analysis

Example: Full Event Stream¶

import asyncio
import json
import requests
import websockets


async def stream_analysis():
    # Start analysis
    response = requests.post(
        "http://localhost:8000/v1/analyze/start",
        json={
            "content": "We need young, energetic developers",
            "options": {"temperature": 0.3}
        }
    )
    run_id = response.json()["run_id"]
    print(f"Started analysis: {run_id}")

    # Connect to WebSocket
    uri = f"ws://localhost:8000/v1/stream/{run_id}"
    async with websockets.connect(uri) as ws:
        print("Connected to event stream\n")

        async for message in ws:
            event = json.loads(message)

            # Format event for display
            timestamp = event["timestamp"]
            event_type = event["event"]
            level = event["level"]
            msg = event["context"].get("message", "")

            # Pretty print based on event type
            if event_type == "workflow_start":
                print(f"🚀 [{level}] Workflow started: {msg}")

            elif event_type == "phase_transition":
                phase = event["context"]["to_phase"]
                print(f"📍 [{level}] Phase: {phase}")

            elif event_type == "tool_call_start":
                tool = event["context"]["tool_name"]
                print(f"🔧 [{level}] Running tool: {tool}")

            elif event_type == "tool_call_end":
                tool = event["context"]["tool_name"]
                duration = event["context"]["duration_ms"]
                print(f"✅ [{level}] {tool} completed ({duration}ms)")

            elif event_type == "llm_call_start":
                model = event["context"]["model"]
                print(f"🤖 [{level}] LLM call: {model}")

            elif event_type == "llm_call_end":
                tokens = event["context"]["tokens_used"]
                print(f"✅ [{level}] LLM response ({tokens} tokens)")

            elif event_type == "refinement_start":
                iteration = event["context"]["iteration_number"]
                print(f"🔄 [{level}] Refinement iteration {iteration}")

            elif event_type == "analysis_complete":
                result = event["context"]["result"]
                print(f"\n✅ [{level}] Analysis complete!")
                print(f"Bias detected: {result['bias_result']['bias_detected']}")
                print(f"Risk level: {result['bias_result']['risk_level']}")
                break

            elif event_type == "analysis_error":
                error = event["context"]["error_message"]
                print(f"\n❌ [{level}] Analysis failed: {error}")
                break

            else:
                # Generic event
                print(f"📋 [{level}] {event_type}: {msg}")


# Run the streaming example
asyncio.run(stream_analysis())

Example Output:

Started analysis: a1b2c3d4-5678-90ab-cdef-1234567890ab
Connected to event stream

🚀 [info] Workflow started: Starting bias_text workflow
📍 [info] Phase: planning
🔧 [info] Running tool: embedding
✅ [info] embedding completed (45ms)
📍 [info] Phase: tool_execution
🔧 [info] Running tool: knowledge_retrieval
✅ [info] knowledge_retrieval completed (120ms)
🤖 [info] LLM call: claude-3-5-sonnet-20241022
✅ [info] LLM response (523 tokens)
📍 [info] Phase: evaluation
🔄 [info] Refinement iteration 1
🤖 [info] LLM call: claude-3-5-sonnet-20241022
✅ [info] LLM response (612 tokens)
📍 [info] Phase: synthesis

✅ [info] Analysis complete!
Bias detected: True
Risk level: medium

Error Handling¶

WebSocket Connection Errors:

try:
    async with websockets.connect(uri) as ws:
        async for message in ws:
            # ... handle events
except websockets.exceptions.ConnectionClosed:
    print("WebSocket connection closed")
except Exception as e:
    print(f"WebSocket error: {e}")

Analysis Errors (via Event Stream):

When analysis fails, you'll receive an analysis_error event:

{
  "run_id": "...",
  "timestamp": 1234567890.123,
  "event": "analysis_error",
  "level": "error",
  "context": {
    "message": "Analysis failed: Tool execution error",
    "error_type": "ToolExecutionError",
    "error_message": "OCR tool failed: Tesseract not found"
  }
}

API Usage Examples¶

Python: Batch Processing¶

from fairsense_agentix import FairSense

# Initialize once (expensive)
fs = FairSense()

# Process multiple texts efficiently
texts = [
    "Job posting 1...",
    "Job posting 2...",
    "Job posting 3...",
]

for text in texts:
    result = fs.analyze_text(text)
    print(f"{result.metadata.run_id}: {result.risk_level}")

Python: Custom Configuration¶

import os
from fairsense_agentix import FairSense

# Set environment variables
os.environ["FAIRSENSE_LLM_PROVIDER"] = "anthropic"
os.environ["FAIRSENSE_LLM_MODEL_NAME"] = "claude-3-5-sonnet-20241022"
os.environ["FAIRSENSE_LLM_API_KEY"] = "sk-ant-..."

# Initialize with custom settings
fs = FairSense()

# Per-request options
result = fs.analyze_text(
    "Text to analyze",
    temperature=0.2,  # Lower for consistency
    max_tokens=3000,  # More detailed analysis
)

REST API: Image Analysis with Node.js¶

const fs = require('fs');
const axios = require('axios');
const FormData = require('form-data');

async function analyzeImage(imagePath) {
  const form = new FormData();
  form.append('file', fs.createReadStream(imagePath));

  const response = await axios.post(
    'http://localhost:8000/v1/analyze/upload',
    form,
    { headers: form.getHeaders() }
  );

  const result = response.data.bias_result;
  console.log(`Bias detected: ${result.bias_detected}`);
  console.log(`Risk level: ${result.risk_level}`);
  console.log(`OCR text: ${result.ocr_text}`);
}

analyzeImage('team_photo.jpg');

REST API: Batch Processing with Python¶

import requests
import time

# Submit batch job
response = requests.post(
    "http://localhost:8000/v1/batch",
    json={
        "items": [
            {"content": "Job posting 1..."},
            {"content": "Job posting 2..."},
            {"content": "Job posting 3..."},
        ]
    }
)
job_id = response.json()["job_id"]
print(f"Batch job started: {job_id}")

# Poll for completion
while True:
    response = requests.get(f"http://localhost:8000/v1/batch/{job_id}")
    status = response.json()

    print(f"Progress: {status['completed']}/{status['total']} ({status['status']})")

    if status["status"] in ["completed", "failed"]:
        break

    time.sleep(2)

# Process results
for i, result in enumerate(status["results"]):
    bias_result = result["bias_result"]
    print(f"Item {i+1}: {bias_result['risk_level']}")

Interactive API Documentation¶

The FastAPI backend provides interactive API documentation:

Swagger UI: http://localhost:8000/docs
ReDoc: http://localhost:8000/redoc

These interfaces allow you to:

Explore all available endpoints
View detailed request/response schemas
Test API calls directly in the browser
Download OpenAPI specification (JSON/YAML)

Rate Limits and Performance¶

Python API:

No built-in rate limits
Performance depends on LLM provider API limits
Typical analysis time: 2-5 seconds (with refinement enabled)

REST API:

No built-in rate limits (configure reverse proxy for production)
Concurrent requests supported via FastAPI async
WebSocket connections: Limited only by server resources

Recommendations:

For high-throughput scenarios, use batch endpoints
Reuse FairSense instances in Python for better performance
Monitor LLM provider token usage and costs
Consider disabling refinement (FAIRSENSE_ENABLE_REFINEMENT=false) for faster results

Error Codes¶

HTTP Code	Meaning	Common Causes
200	Success	Request completed successfully
202	Accepted	Batch job queued
400	Bad Request	Invalid input format, malformed JSON
404	Not Found	Batch job ID doesn't exist
500	Internal Server Error	Tool execution failure, LLM API error

Error Response Format:

{
  "detail": "Error message describing what went wrong"
}

Next Steps¶

Getting Started - Install and configure FairSense
User Guide - Detailed usage examples for each workflow
Server Guide - Deploy and configure the FastAPI backend

API Reference¶

Python API¶

FairSense Class¶

FairSense ¶

__init__ ¶

analyze_text ¶

analyze_image ¶

assess_risk ¶

Result Objects¶

BiasResult¶

BiasResult ¶

status class-attribute instance-attribute ¶

bias_detected class-attribute instance-attribute ¶

risk_level class-attribute instance-attribute ¶

bias_analysis class-attribute instance-attribute ¶

bias_instances class-attribute instance-attribute ¶

summary class-attribute instance-attribute ¶

highlighted_html class-attribute instance-attribute ¶

ocr_text class-attribute instance-attribute ¶

caption_text class-attribute instance-attribute ¶

merged_text class-attribute instance-attribute ¶

image_base64 class-attribute instance-attribute ¶

metadata class-attribute instance-attribute ¶

errors class-attribute instance-attribute ¶

warnings class-attribute instance-attribute ¶

RiskResult¶

RiskResult ¶

status class-attribute instance-attribute ¶

embedding class-attribute instance-attribute ¶

risks class-attribute instance-attribute ¶

rmf_recommendations class-attribute instance-attribute ¶

html_table class-attribute instance-attribute ¶

csv_path class-attribute instance-attribute ¶

metadata class-attribute instance-attribute ¶

errors class-attribute instance-attribute ¶

warnings class-attribute instance-attribute ¶

ResultMetadata¶

ResultMetadata ¶

run_id class-attribute instance-attribute ¶

workflow_id class-attribute instance-attribute ¶

execution_time_seconds class-attribute instance-attribute ¶

router_reasoning class-attribute instance-attribute ¶

router_confidence class-attribute instance-attribute ¶

refinement_count class-attribute instance-attribute ¶

preflight_score class-attribute instance-attribute ¶

posthoc_score class-attribute instance-attribute ¶

REST API¶

Base URL¶

Endpoints¶

Health Check¶

Analyze Text/Content (JSON)¶

Analyze File Upload¶

Start Analysis (for WebSocket streaming)¶

Start File Upload Analysis (for WebSocket streaming)¶

Batch Analysis¶

Get Batch Status¶

Shutdown Server¶

WebSocket Protocol¶

Connection URL¶

Event Structure¶

Event Types¶

Workflow Events¶

Phase Events¶

Connection Lifecycle¶

Example: Full Event Stream¶

Error Handling¶

API Usage Examples¶

Python: Batch Processing¶

Python: Custom Configuration¶

REST API: Image Analysis with Node.js¶

REST API: Batch Processing with Python¶

Interactive API Documentation¶

Rate Limits and Performance¶

Error Codes¶

Next Steps¶

init ¶

status `class-attribute` `instance-attribute` ¶

bias_detected `class-attribute` `instance-attribute` ¶

risk_level `class-attribute` `instance-attribute` ¶

bias_analysis `class-attribute` `instance-attribute` ¶

bias_instances `class-attribute` `instance-attribute` ¶

summary `class-attribute` `instance-attribute` ¶

highlighted_html `class-attribute` `instance-attribute` ¶

ocr_text `class-attribute` `instance-attribute` ¶

caption_text `class-attribute` `instance-attribute` ¶

merged_text `class-attribute` `instance-attribute` ¶

image_base64 `class-attribute` `instance-attribute` ¶

metadata `class-attribute` `instance-attribute` ¶

errors `class-attribute` `instance-attribute` ¶

warnings `class-attribute` `instance-attribute` ¶

status `class-attribute` `instance-attribute` ¶

embedding `class-attribute` `instance-attribute` ¶

risks `class-attribute` `instance-attribute` ¶

rmf_recommendations `class-attribute` `instance-attribute` ¶

html_table `class-attribute` `instance-attribute` ¶

csv_path `class-attribute` `instance-attribute` ¶

metadata `class-attribute` `instance-attribute` ¶

errors `class-attribute` `instance-attribute` ¶

warnings `class-attribute` `instance-attribute` ¶

run_id `class-attribute` `instance-attribute` ¶

workflow_id `class-attribute` `instance-attribute` ¶

execution_time_seconds `class-attribute` `instance-attribute` ¶

router_reasoning `class-attribute` `instance-attribute` ¶

router_confidence `class-attribute` `instance-attribute` ¶

refinement_count `class-attribute` `instance-attribute` ¶

preflight_score `class-attribute` `instance-attribute` ¶

posthoc_score `class-attribute` `instance-attribute` ¶