From Static Graphs to Thinking Systems: Agentic GraphRAG for COBOL Codebases

Ingestion Agent     → "What files do we have?"
Validation Agent    → "Is this valid COBOL?"
Parsing Agent       → "What entities and relations exist?"
Enrichment Agent    → "What does this code mean?"
Graph Builder Agent → "How do we model this in Neo4j?"
Cypher Agent        → "How do we query this?"
Retrieval Agent     → "What's the answer?"

cobol_agentic_kg/
├── agents/                 # Specialized AI agents
├── workflows/              # LangGraph orchestration
├── utils/                  # State, Neo4j, logging
├── ui/                     # Streamlit dashboard
└── tests/                  # Production-grade tests

"""Ingestion Agent - Handles file uploads and repository scanning"""
import os
from typing import Dict, Any
from utils.state import CobolProcessingState
from utils.logger import logger
import chardet


class IngestionAgent:
    """Agent responsible for ingesting COBOL files"""

    def __init__(self):
        self.supported_extensions = ['.cob', '.cbl', '.cobol', '.cpy']

    def process(self, state: CobolProcessingState) -> CobolProcessingState:
        """
        Ingest COBOL file and extract metadata

        Args:
            state: Current processing state

        Returns:
            Updated state with file content and metadata
        """
        logger.info(f"🔽 INGESTION AGENT: Processing {state['file_path']}")

        try:
            # Read file with encoding detection
            file_content = self._read_file(state['file_path'])

            # Create metadata
            metadata = self._create_metadata(state['file_path'], file_content)

            # Update state
            return {
                **state,
                "file_content": file_content,
                "file_metadata": metadata,
                "stage": "ingestion",
                "status": "completed"
            }
        except Exception as e:
            logger.error(f"Ingestion failed: {e}")
            return {
                **state,
                "stage": "ingestion",
                "status": "failed",
                "errors": state.get('errors', []) + [f"Ingestion error: {str(e)}"]
            }

    def _read_file(self, file_path: str) -> str:
        """
        Read file with automatic encoding detection

        Args:
            file_path: Path to file

        Returns:
            File content as string
        """
        # Try UTF-8 first
        try:
            with open(file_path, 'r', encoding='utf-8') as f:
                return f.read()
        except UnicodeDecodeError:
            # Detect encoding
            with open(file_path, 'rb') as f:
                raw_data = f.read()
                result = chardet.detect(raw_data)
                encoding = result['encoding'] or 'utf-8'
            # Read with detected encoding
            with open(file_path, 'r', encoding=encoding, errors='ignore') as f:
                return f.read()

    def _create_metadata(self, file_path: str, content: str) -> Dict[str, Any]:
        """
        Create file metadata

        Args:
            file_path: Path to file
            content: File content

        Returns:
            Metadata dictionary
        """
        return {
            "filename": os.path.basename(file_path),
            "directory": os.path.dirname(file_path),
            "size_bytes": len(content.encode('utf-8')),
            "line_count": content.count('\n'),
            "char_count": len(content),
            "extension": os.path.splitext(file_path)[1]
        }

    def is_cobol_file(self, file_path: str) -> bool:
        """
        Check if file is a COBOL file by extension

        Args:
            file_path: Path to file

        Returns:
            True if COBOL file, False otherwise
        """
        ext = os.path.splitext(file_path)[1].lower()
        return ext in self.supported_extensions


# Create singleton instance
ingestion_agent = IngestionAgent()


# Wrapper function for LangGraph
def ingestion_agent_node(state: CobolProcessingState) -> CobolProcessingState:
    """LangGraph node wrapper"""
    return ingestion_agent.process(state)

(:CobolProgram)-[:READS]->(:DataFile)
(:Program)-[:CALLS]->(:Program)
(:Program)-[:USES]->(:Copybook)

cypher_chain.invoke({
  "query": "Which programs read from a file?"
})

MATCH (p:CobolProgram)-[:READS]->(:DataFile)
RETURN DISTINCT p.id AS program_id

List all programs in the Finance domain

MATCH (p:CobolProgram) WHERE toLower(p.domain) = toLower('finance') RETURN p.name, p.summary, p.loc ORDER BY p.name

In the Finance domain, we have identified 14 key programs that handle various
financial operations. Notable programs include CBACT01C, which processes account
records, and CBSTM03A, which generates account statements. Programs like COBIL00C
facilitate bill payments, while COPAUA0C and its related programs manage card
authorization processes.

These programs are crucial for maintaining financial transactions and customer
account management, meaning any changes or maintenance will directly impact
financial reporting and customer service. It's important to ensure that updates
are carefully managed to avoid disruptions in these critical functions.