> ## Documentation Index
> Fetch the complete documentation index at: https://mintlify.com/AllianceBioversityCIAT/alliance-risk-analysis-tool/llms.txt
> Use this file to discover all available pages before exploring further.

# AI Analysis Pipeline

> Multi-agent AI system powered by AWS Bedrock for document parsing, gap detection, and comprehensive risk analysis

## Overview

The CGIAR Risk Intelligence Tool uses a **multi-agent AI pipeline** built on **AWS Bedrock** to automate risk assessment. The system orchestrates specialized AI agents, each handling a specific phase of the analysis workflow.

## Architecture

<Frame>
  ```mermaid theme={null}
  graph LR
      A[Document Upload] --> B[Textract OCR]
      B --> C[Parser Agent]
      C --> D[Gap Detector Agent]
      D --> E[Risk Analysis Agent]
      E --> F[Report Generator]
      F --> G[PDF Report]
      
      style C fill:#4F46E5
      style D fill:#4F46E5
      style E fill:#4F46E5
      style F fill:#4F46E5
  ```
</Frame>

### Processing Flow

1. **Document Upload** → PDF files uploaded to S3
2. **AWS Textract** → Extracts text and tables from documents
3. **Parser Agent** → Structures extracted data into risk categories
4. **Gap Detector** → Identifies missing or incomplete fields
5. **Risk Analysis Agent** → Scores all 7 risk categories with subcategories
6. **Report Generator** → Creates comprehensive PDF with traffic-light indicators

<Info>
  All AI operations run asynchronously as background jobs. The frontend polls job status for completion.
</Info>

## AWS Bedrock Integration

### Foundation Models

All agents use **Claude 3.5 Sonnet v2** from Anthropic:

<CodeGroup>
  ```typescript bedrock.config.ts theme={null}
  import { AgentSection } from '../enums/agent-section.enum';

  export const BEDROCK_MODELS: Record<
    AgentSection,
    { modelId: string; knowledgeBaseId?: string }
  > = {
    [AgentSection.PARSER]: {
      modelId: 'anthropic.claude-3-5-sonnet-20241022-v2:0',
    },
    [AgentSection.GAP_DETECTOR]: {
      modelId: 'anthropic.claude-3-5-sonnet-20241022-v2:0',
    },
    [AgentSection.RISK_ANALYSIS]: {
      modelId: 'anthropic.claude-3-5-sonnet-20241022-v2:0',
    },
    [AgentSection.REPORT_GENERATION]: {
      modelId: 'anthropic.claude-3-5-sonnet-20241022-v2:0',
    },
  };
  ```

  ```typescript BedrockService.invokeModel theme={null}
  async invokeModel(params: InvokeModelParams): Promise<{
    output: string;
    tokensUsed: number;
    processingTime: number;
  }> {
    const body = JSON.stringify({
      anthropic_version: 'bedrock-2023-05-31',
      max_tokens: 4096,
      system: params.systemPrompt,
      messages: [
        { role: 'user', content: params.userPrompt },
      ],
    });

    const response = await this.client.send(
      new InvokeModelCommand({
        modelId: params.modelId,
        contentType: 'application/json',
        accept: 'application/json',
        body: Buffer.from(body),
      })
    );

    // Parse response and extract token usage
    const responseBody = JSON.parse(Buffer.from(response.body).toString('utf-8'));
    const output = responseBody.content
      ?.filter(c => c.type === 'text')
      .map(c => c.text)
      .join('') ?? '';

    const tokensUsed =
      (responseBody.usage?.input_tokens ?? 0) +
      (responseBody.usage?.output_tokens ?? 0);

    return { output, tokensUsed, processingTime: Date.now() - startTime };
  }
  ```
</CodeGroup>

### Resilience Features

<AccordionGroup>
  <Accordion title="Circuit Breaker" icon="plug-circle-xmark">
    Prevents cascading failures by opening after 3 consecutive failures:

    ```typescript theme={null}
    this.circuitBreaker = new CircuitBreaker({
      failureThreshold: 3,
      resetTimeoutMs: 60_000,
      isFailure: (err) => !((err as Error)?.name === 'ValidationException'),
    });
    ```

    * **Open:** Rejects requests immediately for 60 seconds
    * **Half-Open:** Allows one test request after timeout
    * **Closed:** Normal operation
  </Accordion>

  <Accordion title="Exponential Retry" icon="rotate">
    Automatically retries throttled requests:

    ```typescript theme={null}
    await withRetry(
      () => this.client.send(new InvokeModelCommand(...)),
      {
        maxAttempts: 3,
        isRetryable: (err) =>
          (err as Error)?.name === 'ThrottlingException' ||
          (err as Error)?.name === 'ServiceUnavailableException',
      }
    );
    ```

    * Retries up to 3 times for throttling/service errors
    * Exponential backoff between attempts
    * Fails fast for validation errors
  </Accordion>
</AccordionGroup>

## Agent Pipeline

### 1. Parser Agent

**Purpose:** Structures raw Textract output into organized risk category data

**Input:**

* Extracted text content from AWS Textract
* Extracted tables with headers and rows
* Assessment metadata (company name, type, country)

**Processing:**

```typescript theme={null}
interface ParseDocumentInput {
  assessmentId: string;
  documentId: string;
  s3Key: string;
}

// Job Type: PARSE_DOCUMENT
const jobId = await jobsService.create(
  JobType.PARSE_DOCUMENT,
  { assessmentId, documentId, s3Key },
  userId
);
```

**AWS Textract Analysis:**

<CodeGroup>
  ```typescript TextractService.analyzeDocument theme={null}
  async analyzeDocument(s3Bucket: string, s3Key: string): Promise<ExtractionResult> {
    // Step 1: Start async analysis job
    const textractJobId = await this.startAnalysis(s3Bucket, s3Key);
    
    // Step 2: Poll with exponential backoff (2s → 4s → 8s → 16s → 30s)
    const pages = await this.pollUntilComplete(textractJobId);
    
    // Step 3: Extract LINE blocks for text content
    const textContent = lineBlocks
      .map(b => b.Text ?? '')
      .filter(Boolean)
      .join('\n');
    
    // Step 4: Extract TABLE blocks with cells
    const tables: ExtractedTable[] = tableBlocks.map((table) => {
      const cells = getCellBlocks(table);
      const grid = buildTableGrid(cells);
      return {
        page: table.Page ?? 1,
        headers: grid[0],
        rows: grid,
        rowCount: grid.length,
        columnCount: grid[0].length
      };
    });
    
    return { pages: pageCount, textContent, tables, metadata };
  }
  ```

  ```typescript ExtractionResult theme={null}
  {
    "pages": 12,
    "textContent": "BUSINESS PLAN\nGreen Valley Agritech\n...",
    "tables": [
      {
        "page": 3,
        "tableIndex": 0,
        "rowCount": 5,
        "columnCount": 4,
        "headers": ["Year", "Revenue", "Expenses", "Profit"],
        "rows": [
          ["Year", "Revenue", "Expenses", "Profit"],
          ["2023", "$1.2M", "$950K", "$250K"],
          ["2024", "$1.5M", "$1.1M", "$400K"],
          ["2025", "$1.8M", "$1.3M", "$500K"]
        ]
      }
    ],
    "metadata": {
      "textractJobId": "job-abc123",
      "s3Key": "assessments/550e8400/documents/doc-123/business-plan.pdf",
      "processingTimeMs": 12450,
      "processedAt": "2026-03-04T10:35:12Z",
      "textractModel": "AnalyzeDocument/TABLES"
    }
  }
  ```
</CodeGroup>

**Output:** Structured data ready for gap detection

**Document Status Transitions:**

```
PENDING_UPLOAD → UPLOADED → PARSING → PARSED
                                    ↓
                                 FAILED (on error)
```

<Warning>
  Textract jobs can take several minutes for large PDFs. The frontend should poll document status every 3-5 seconds with a maximum timeout of 10 minutes.
</Warning>

### 2. Gap Detector Agent

**Purpose:** Identifies missing or incomplete data fields that need user verification

**Input:**

* Parsed document data
* Assessment metadata
* List of required fields per risk category

**Processing:**

```typescript theme={null}
interface GapDetectionInput {
  assessmentId: string;
}

// Job Type: GAP_DETECTION (auto-chained after PARSE_DOCUMENT)
```

**Gap Field States:**

<Tabs>
  <Tab title="MISSING">
    Field has no extracted value. User must provide data manually.

    ```typescript theme={null}
    {
      "field": "revenue_2025",
      "label": "FY2025 Revenue",
      "extractedValue": null,
      "status": "MISSING",
      "isMandatory": true
    }
    ```
  </Tab>

  <Tab title="PARTIAL">
    Field has an extracted value but needs verification or is incomplete.

    ```typescript theme={null}
    {
      "field": "operating_margin",
      "label": "Operating Margin %",
      "extractedValue": "~12% (estimated)",
      "status": "PARTIAL",
      "isMandatory": true
    }
    ```
  </Tab>

  <Tab title="VERIFIED">
    User has reviewed and confirmed the extracted value.

    ```typescript theme={null}
    {
      "field": "total_employees",
      "label": "Total Employees",
      "extractedValue": "45",
      "correctedValue": "47",
      "status": "VERIFIED",
      "isMandatory": false
    }
    ```
  </Tab>
</Tabs>

**Output:** Gap fields created across all 7 risk categories (5 fields each = 35 total)

**Assessment Status Transition:**

```typescript theme={null}
DRAFT → ANALYZING → ACTION_REQUIRED (if gaps detected)
                  → COMPLETE (if no gaps)
```

<Note>
  Gap detection is triggered for `GUIDED_INTERVIEW` and `MANUAL_ENTRY` intake modes after data submission, and for `UPLOAD` mode after document parsing.
</Note>

### 3. Risk Analysis Agent

**Purpose:** Generates risk scores, narratives, and recommendations for all categories

**Input:**

* Verified gap field data
* Assessment metadata
* Historical risk benchmarks (future enhancement)

**Processing:**

```typescript theme={null}
interface RiskAnalysisInput {
  assessmentId: string;
}

// Job Type: RISK_ANALYSIS
const jobId = await jobsService.create(
  JobType.RISK_ANALYSIS,
  { assessmentId },
  userId
);
```

**For Each Risk Category:**

<Steps>
  <Step title="Score Subcategories">
    AI analyzes data and assigns scores (0-100) to 5 subcategories:

    ```typescript theme={null}
    const subcategories = [
      {
        name: 'Revenue Stability',
        indicator: 'Year-over-year revenue variance',
        score: 42,
        level: 'MODERATE',
        evidence: 'Revenue declined 8% in FY2025 but stabilized in Q4',
        mitigation: 'Diversify customer base and develop new revenue streams'
      },
      // ... 4 more subcategories
    ];
    ```
  </Step>

  <Step title="Calculate Category Score">
    Aggregate subcategory scores (default: equal weights):

    ```typescript theme={null}
    const categoryScore = Math.round(
      subcategories.reduce((sum, sub) => sum + sub.score, 0) / subcategories.length
    );
    ```
  </Step>

  <Step title="Assign Risk Level">
    Map score to traffic-light level:

    ```typescript theme={null}
    const level = 
      categoryScore < 25 ? 'LOW' :
      categoryScore < 50 ? 'MODERATE' :
      categoryScore < 75 ? 'HIGH' : 'CRITICAL';
    ```
  </Step>

  <Step title="Generate Narrative">
    AI creates contextual risk narrative:

    ```typescript theme={null}
    const narrative = `The ${category} risk level is ${level} based on analysis of ${subcategories.map(s => s.name).join(', ')}.`;
    ```
  </Step>

  <Step title="Create Recommendations">
    Generate 2-3 prioritized recommendations:

    ```typescript theme={null}
    const recommendations = [
      {
        text: 'Develop a 3-year financial sustainability plan...',
        priority: 'HIGH',
        order: 0
      },
      {
        text: 'Implement quarterly financial health monitoring...',
        priority: 'MEDIUM',
        order: 1
      }
    ];
    ```
  </Step>
</Steps>

**Output:** Complete risk scores for all 7 categories + overall assessment score

**Assessment Status Transition:**

```typescript theme={null}
ACTION_REQUIRED → COMPLETE (progress: 90)
```

### 4. Report Generation Agent

**Purpose:** Creates PDF report with visualizations and traffic-light indicators

**Input:**

* Complete risk score data
* Assessment metadata
* Recommendations

**Processing:**

```typescript theme={null}
interface ReportGenerationInput {
  assessmentId: string;
}

// Job Type: REPORT_GENERATION
const jobId = await jobsService.create(
  JobType.REPORT_GENERATION,
  { assessmentId },
  userId
);
```

**See:** [Report Generation](/features/report-generation) for detailed documentation

## Asynchronous Job Processing

All AI operations run as background jobs:

<CodeGroup>
  ```typescript Creating a Job theme={null}
  // API Lambda creates job and invokes Worker Lambda
  const jobId = await jobsService.create(
    JobType.PARSE_DOCUMENT,
    { assessmentId, documentId, s3Key },
    userId
  );

  // Returns immediately with job ID
  return { jobId };
  ```

  ```typescript Job Status Lifecycle theme={null}
  enum JobStatus {
    PENDING = 'PENDING',        // Created, waiting for worker
    PROCESSING = 'PROCESSING',  // Worker is executing
    COMPLETED = 'COMPLETED',    // Success
    FAILED = 'FAILED'           // Error after max retries
  }
  ```

  ```typescript Polling Job Status theme={null}
  GET /api/jobs/:id

  {
    "id": "job-550e8400",
    "type": "PARSE_DOCUMENT",
    "status": "PROCESSING",
    "input": { "assessmentId": "...", "documentId": "..." },
    "result": null,
    "error": null,
    "attempts": 1,
    "maxAttempts": 3,
    "createdAt": "2026-03-04T10:30:00Z",
    "startedAt": "2026-03-04T10:30:05Z",
    "completedAt": null
  }
  ```
</CodeGroup>

### Retry Logic

<AccordionGroup>
  <Accordion title="Automatic Retries" icon="rotate">
    Jobs retry up to 3 times on failure:

    ```typescript theme={null}
    if (attempts >= maxAttempts) {
      await updateStatus(jobId, JobStatus.FAILED, undefined, errorMsg);
      // Notify handler of permanent failure
      await handler.onFailure(documentId, error);
    } else {
      // Reset to PENDING for retry
      await updateStatus(jobId, JobStatus.PENDING);
    }
    ```
  </Accordion>

  <Accordion title="Job Chaining" icon="link">
    Jobs can automatically trigger dependent jobs:

    ```typescript theme={null}
    // After PARSE_DOCUMENT completes, auto-chain GAP_DETECTION
    if (job.type === JobType.PARSE_DOCUMENT && job.status === 'COMPLETED') {
      const gapJobId = await this.create(
        JobType.GAP_DETECTION,
        { assessmentId: input.assessmentId },
        job.createdById
      );
      await this.processJob(gapJobId);
    }
    ```
  </Accordion>
</AccordionGroup>

## Prompt Management

AI agents use versioned prompts managed through the Prompt CMS:

<CodeGroup>
  ```typescript GET /api/prompts/section/:section theme={null}
  // Public endpoint for runtime prompt retrieval
  GET /api/prompts/section/parser

  {
    "id": "prompt-123",
    "section": "PARSER",
    "systemPrompt": "You are an expert document analyst...",
    "userPromptTemplate": "Analyze the following document and extract...",
    "tone": "Professional and analytical",
    "outputFormat": "JSON with structured fields",
    "version": 5
  }
  ```

  ```typescript Variable Injection theme={null}
  // Prompts support variable injection
  const userPrompt = userPromptTemplate
    .replace(/{{company_name}}/g, assessment.companyName)
    .replace(/{{country}}/g, assessment.country)
    .replace(/{{categories}}/g, RISK_CATEGORIES.map(c => c.label).join(', '));
  ```
</CodeGroup>

**See:** Prompt Management documentation for versioning, comments, and change tracking

## Code Example: Complete AI Pipeline

```typescript theme={null}
import { useJob } from '@/hooks/use-job';

function AssessmentAnalysis({ assessmentId }: { assessmentId: string }) {
  // Step 1: Trigger risk analysis
  const triggerAnalysis = async () => {
    const { jobId } = await fetch(
      `/api/assessments/${assessmentId}/analyze`,
      { method: 'POST' }
    ).then(r => r.json());
    
    return jobId;
  };
  
  // Step 2: Poll job status with custom hook
  const { job, isLoading, error } = useJob(jobId, {
    pollInterval: 3000,
    maxAttempts: 200, // ~10 minutes
  });
  
  // Step 3: Handle completion
  useEffect(() => {
    if (job?.status === 'COMPLETED') {
      // Refresh assessment data to get new scores
      refreshAssessment();
    } else if (job?.status === 'FAILED') {
      toast.error(`Analysis failed: ${job.error}`);
    }
  }, [job?.status]);
  
  return (
    <div>
      {isLoading && <ProgressBar progress={job?.progress ?? 0} />}
      {job?.status === 'COMPLETED' && <RiskScoresDashboard />}
    </div>
  );
}
```

## Best Practices

<CardGroup cols={2}>
  <Card title="Monitor Token Usage" icon="coins">
    Track Bedrock token consumption per job type to optimize costs and identify inefficient prompts
  </Card>

  <Card title="Implement Timeouts" icon="clock">
    Set reasonable polling timeouts (10 min for Textract, 5 min for AI jobs) to prevent infinite loops
  </Card>

  <Card title="Handle Partial Results" icon="warning">
    Design UIs to show progress and partial results rather than blocking on job completion
  </Card>

  <Card title="Version Prompts" icon="code-branch">
    Always version prompts before deploying to production. Use the Prompt CMS change tracking.
  </Card>
</CardGroup>

## Related Resources

<CardGroup cols={2}>
  <Card title="Risk Scoring" href="/features/risk-scoring" icon="chart-line">
    Understand how AI-generated scores map to risk levels
  </Card>

  <Card title="Assessment Workflow" href="/features/assessments" icon="list-check">
    See how AI agents fit into the assessment lifecycle
  </Card>

  <Card title="Report Generation" href="/features/report-generation" icon="file-pdf">
    Learn about PDF report creation with AI-generated content
  </Card>
</CardGroup>