Part 1: Translating "Reviews + Descriptions" to Accounting & GL

In e-commerce, review text captures user-experienced reality while product descriptions capture vendor-stated intent. In General Ledger (GL) and financial management, the exact same dual-signal pattern exists—but with different source materials and far higher stakes:

E-Commerce ConceptGL/Accounting EquivalentWhy It Matters for Intent Alignment
Product DescriptionChart of Accounts (COA) definitions, policy docs, mapping rulesStates intended classification, debit/credit rules, and reporting hierarchy
Review TextTransaction line narratives, journal entry memos, auditor notes, AP/AR commentsCaptures actual business context that often deviates from formal definitions
Star Rating / SentimentReconciliation flags, exception codes, audit adjustment frequencySignals confidence/mismatch between stated intent and recorded reality
Q&A SectionController desk inquiries, period-close ticket threads, ERP helpdesk logsReveals recurring intent ambiguities where COA definitions fail practitioners

Why Single-Source Retrieval Fails in GL

A controller asks: "How should we classify the $45K payment to Acme Corp labeled 'Q3 platform services' when Acme is both a software vendor and a subcontractor?"

The Three-Layer Intent Alignment Model for GL

Layer 1: STRUCTURAL INTENT (COA + Policies)
   → What SHOULD this be per formal definitions?
   
Layer 2: BEHAVIORAL INTENT (Transaction Narratives + History)  
   → What HAS this been in practice for similar transactions?
   
Layer 3: RESOLUTION INTENT (Audit Notes + Controller Rulings)
   → How were past ambiguities RESOLVED when Layers 1 & 2 conflicted?

Key insight: In GL, Layer 3 is the most valuable for intent alignment because it encodes institutional memory of judgment calls. This is the accounting equivalent of reading reviews to discover that "runs small" overrides the size chart.

Critical Differences from E-Commerce Intent Alignment

Part 2: Real-Time Use Case — GL Classification Agent with Dual-Signal Intent Alignment

The Business Problem

During month-end close, a staff accountant encounters ambiguous transactions daily. They ask: "Classify invoice #INV-2024-8847 from Meridian Consulting ($28,500, described as 'strategic advisory and implementation support'). Is this consulting expense, professional services, or capitalizable implementation cost? Show me how similar Meridian transactions were handled in prior periods."

This requires aligning three signal sources simultaneously while maintaining full audit trail state.

Architecture: LangGraph Multi-Agent GL System with Memory

"""
General Ledger Intent Alignment RAG System
Dependencies: langgraph, chromadb>=0.5, langchain-openai, pydantic, numpy
"""

import operator
import time
from typing import Annotated, TypedDict, Literal, List, Dict, Any, Optional, Tuplefrom datetime import datetime
from enum import Enum
from dataclasses import dataclass, field
from langgraph.graph import StateGraph, START
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.messages import HumanMessage, AIMessage, BaseMessage
from pydantic import BaseModel
import chromadb


# =============================================================================# 1. DUAL-SIGNAL RETRIEVAL ENGINE FOR GL# =============================================================================

class IntentSignalSource(str, Enum):
    """Three-layer intent model for GL classification."""
    STRUCTURAL = "structural"       # COA definitions, policies, mapping rules
    BEHAVIORAL = "behavioral"       # Transaction narratives, historical entries
    RESOLUTION = "resolution"       # Audit notes, controller rulings, exception resolutionsclass AuthorityLevel(int, Enum):
    """Weighted authority hierarchy for GL intent sources."""
    AUDITOR_RULING = 5
    CONTROLLER_MEMO = 4
    SENIOR_ACCOUNTANT_NOTE = 3
    AP_AR_COMMENT = 2
    SYSTEM_GENERATED = 1@dataclassclass IntentSignal:
    """A single aligned intent signal with provenance and confidence."""
    source_type: IntentSignalSource
    authority: AuthorityLevel
    content: str
    metadata: Dict[str, Any]
    relevance_score: float      # Vector similarity
    temporal_validity: str      # Fiscal period applicability
    confidence: float           # Combined score after authority weighting
    

# Collection configs per intent layer
GL_COLLECTION_CONFIGS: Dict[IntentSignalSource, Dict] = {
    IntentSignalSource.STRUCTURAL: {
        "collection_name": "gl_coa_and_policies",
        "distance_threshold": 0.30,     # Tightest: definitions must be precise
        "base_n_results": 5,
        "required_metadata": ["account_code", "effective_fy"],
        "authority_default": AuthorityLevel.SYSTEM_GENERATED,
    },
    IntentSignalSource.BEHAVIORAL: {
        "collection_name": "gl_transaction_narratives",
        "distance_threshold": 0.38,     # Moderate: narratives vary in quality
        "base_n_results": 10,           # Over-retrieve; filter by authority + recency
        "required_metadata": ["vendor_name", "fiscal_period"],
        "authority_default": AuthorityLevel.AP_AR_COMMENT,
    },
    IntentSignalSource.RESOLUTION: {
        "collection_name": "gl_audit_and_rulings",
        "distance_threshold": 0.35,
        "base_n_results": 7,
        "required_metadata": ["ruling_type", "fiscal_year"],
        "authority_default": AuthorityLevel.CONTROLLER_MEMO,
    },
}


class GLIntentRetriever:
    """Dual-signal retriever that aligns structural, behavioral, and resolution intent."""
    
    def __init__(self, chroma_client: chromadb.ClientAPI, embedder: OpenAIEmbeddings):
        self.client = chroma_client
        self.embedder = embedder
        self._collections: Dict[str, chromadb.Collection] = {}
        self._init_collections()
    
    def _init_collections(self):
        for source_type, config in GL_COLLECTION_CONFIGS.items():
            self._collections[config["collection_name"]] = \
                self.client.get_or_create_collection(
                    name=config["collection_name"],
                    metadata={"intent_layer": source_type.value}
                )
    
    def retrieve_aligned_intent(
        self,
        query: str,
        vendor_name: Optional[str] = None,
        fiscal_period: Optional[str] = None,
        account_codes_hint: Optional[List[str]] = None,
    ) -> Dict[str, Any]:
        """Retrieve and align intent across all three signal layers."""
        
        query_embedding = self.embedder.embed_query(query)
        all_signals: List[IntentSignal] = []
        layer_telemetry: Dict[str, Any] = {}
        
        for source_type, config in GL_COLLECTION_CONFIGS.items():
            collection = self._collections[config["collection_name"]]
            
            # Build metadata filter per layer
            meta_filter: Dict[str, Any] = {}
            if vendor_name and source_type == IntentSignalSource.BEHAVIORAL:
                meta_filter["vendor_name"] = vendor_name
            if fiscal_period:
                meta_filter["fiscal_period"] = fiscal_period
            if account_codes_hint and source_type == IntentSignalSource.STRUCTURAL:
                meta_filter["account_code"] = {"$in": account_codes_hint}
            
            # Over-retrieve then threshold-gate
            raw = collection.query(
                query_embeddings=[query_embedding],
                n_results=config["base_n_results"] * 2,
                where=meta_filter if meta_filter else None,
                include=["documents", "metadatas", "distances"]
            )
            
            accepted_count = 0
            for i, dist in enumerate(raw["distances"][0]):
                if dist <= config["distance_threshold"]:
                    meta = raw["metadatas"][0][i]
                    
                    # Determine authority level from metadata
                    auth_str = meta.get("authority_level", "").upper()
                    try:
                        authority = AuthorityLevel[auth_str]
                    except KeyError:
                        authority = config["authority_default"]
                    
                    # Confidence = relevance × authority_weight
                    auth_weight = authority.value / 5.0
                    confidence = (1.0 - dist) * auth_weight
                    
                    signal = IntentSignal(
                        source_type=source_type,
                        authority=authority,
                        content=raw["documents"][0][i],
                        metadata=meta,
                        relevance_score=round(1.0 - dist, 4),
                        temporal_validity=meta.get("fiscal_period", "unknown"),
                        confidence=round(confidence, 4),
                    )
                    all_signals.append(signal)
                    accepted_count += 1
            
            layer_telemetry[source_type.value] = {
                "raw": len(raw["distances"][0]),
                "accepted": accepted_count,
                "threshold": config["distance_threshold"],
            }
        
        # Sort by confidence (authority-weighted relevance)
        all_signals.sort(key=lambda s: s.confidence, reverse=True)
        
        return {
            "signals": all_signals,
            "telemetry": layer_telemetry,
            "total_signals": len(all_signals),
            "layers_with_results": [
                st.value for st in IntentSignalSource
                if any(s.source_type == st for s in all_signals)
            ],
        }


# =============================================================================# 2. LANGGRAPH STATE & MULTI-AGENT GL SYSTEM# =============================================================================

class GLClassificationState(TypedDict):
    messages: Annotated[List[BaseMessage], operator.add]
    query: str
    vendor_name: Optional[str]
    transaction_amount: Optional[float]
    transaction_description: Optional[str]
    fiscal_period: str
    
    # Intent alignment results
    structural_intent: List[Dict]       # COA/policy matches
    behavioral_intent: List[Dict]       # Historical transaction patterns
    resolution_intent: List[Dict]       # Prior rulings on similar ambiguity
    aligned_recommendation: Optional[str]
    confidence_level: Optional[str]     # HIGH / MEDIUM / LOW / ESCALATE
    conflicting_signals: List[Dict]     # Where layers disagree
    
    # Memory & audit
    conversation_memory: Dict[str, Any] # Persisted across turns
    audit_trail: List[str]
    error: Optional[str]


llm = ChatOpenAI(model="gpt-4o", temperature=0)
embedder = OpenAIEmbeddings(model="text-embedding-3-small")
chroma_client = chromadb.PersistentClient(path="./gl_intent_chroma")
intent_retriever = GLIntentRetriever(chroma_client, embedder)


def intent_retrieval_agent(state: GLClassificationState) -> Dict:
    """Retrieve aligned intent from all three signal layers."""
    
    result = intent_retriever.retrieve_aligned_intent(
        query=state["query"],
        vendor_name=state.get("vendor_name"),
        fiscal_period=state.get("fiscal_period"),
    )
    
    # Separate signals by layer
    structural = [s.__dict__ for s in result["signals"] 
                  if s.source_type == IntentSignalSource.STRUCTURAL]
    behavioral = [s.__dict__ for s in result["signals"] 
                  if s.source_type == IntentSignalSource.BEHAVIORAL]
    resolution = [s.__dict__ for s in result["signals"] 
                  if s.source_type == IntentSignalSource.RESOLUTION]
    
    audit = [
        f"[INTENT RETRIEVAL] {result['total_signals']} total signals across "
        f"{len(result['layers_with_results'])} layers",
        *[f"  {k}: {v['accepted']}/{v['raw']} accepted (threshold={v['threshold']})"
          for k, v in result["telemetry"].items()]
    ]
    
    return {
        "structural_intent": structural,
        "behavioral_intent": behavioral,
        "resolution_intent": resolution,
        "audit_trail": audit,
        "messages": [AIMessage(content=(
            f"Retrieved {len(structural)} structural, {len(behavioral)} behavioral, "
            f"{len(resolution)} resolution signals"
        ))]
    }


def conflict_detection_agent(state: GLClassificationState) -> Dict:
    """Detect disagreements between intent layers.
    This is where dual-signal alignment adds value beyond simple retrieval."""
    
    conflicts = []
    
    # Extract suggested account codes from each layer
    structural_accounts = set()
    for s in state["structural_intent"]:
        code = s.get("metadata", {}).get("account_code")
        if code:
            structural_accounts.add(code)
    
    behavioral_accounts = {}
    for b in state["behavioral_intent"]:
        code = b.get("metadata", {}).get("account_code")
        if code:
            behavioral_accounts[code] = behavioral_accounts.get(code, 0) + 1
    
    resolution_accounts = set()
    for r in state["resolution_intent"]:
        code = r.get("metadata", {}).get("resolved_account_code")
        if code:
            resolution_accounts.add(code)
    
    # Detect structural vs. behavioral disagreement
    if structural_accounts and behavioral_accounts:
        top_behavioral = max(behavioral_accounts, key=behavioral_accounts.get)
        if top_behavioral not in structural_accounts:
            conflicts.append({
                "type": "STRUCTURAL_VS_BEHAVIORAL",
                "structural_suggestion": list(structural_accounts)[:3],
                "behavioral_pattern": top_behavioral,
                "behavioral_frequency": behavioral_accounts[top_behavioral],
                "severity": "MEDIUM",
                "message": (f"COA suggests accounts {list(structural_accounts)[:3]} but "
                           f"historical transactions for this vendor predominantly used "
                           f"{top_behavioral} ({behavioral_accounts[top_behavioral]} times)")
            })
    
    # Resolution layer overrides everything
    if resolution_accounts and structural_accounts:
        unresolved = resolution_accounts - structural_accounts
        if unresolved:
            conflicts.append({
                "type": "RESOLUTION_OVERRIDE",
                "resolution_accounts": list(resolution_accounts),
                "severity": "HIGH",
                "message": (f"Prior controller/auditor ruling classified similar transactions "
                           f"to {list(resolution_accounts)}, which differs from current COA mapping")
            })
    
    return {
        "conflicting_signals": conflicts,
        "audit_trail": [f"[CONFLICT DETECTION] {len(conflicts)} conflicts identified"],
        "messages": [AIMessage(content=(
            f"{'⚠️ ' if conflicts else '✅ '}Found {len(conflicts)} intent conflicts"
        ))]
    }


def classification_recommendation_agent(state: GLClassificationState) -> Dict:
    """Generate aligned recommendation with confidence level and escalation logic."""
    
    # Build cited context from all three layers
    def format_signals(signals: List[Dict], label: str, max_items: int = 3) -> str:
        if not signals:
            return f"=== {label} ===\nNo matching signals retrieved."
        parts = []
        for s in signals[:max_items]:
            cite = s.get("metadata", {}).get("citation_ref", "unreferenced")
            conf = s.get("confidence", 0)
            parts.append(f"[{cite} | conf={conf:.2f}] {s['content'][:400]}")
        return f"=== {label} ===\n" + "\n\n".join(parts)
    
    context = "\n\n".join([
        format_signals(state["structural_intent"], "STRUCTURAL INTENT (COA & Policies)"),
        format_signals(state["behavioral_intent"], "BEHAVIORAL INTENT (Transaction History)"),
        format_signals(state["resolution_intent"], "RESOLUTION INTENT (Prior Rulings)"),
    ])
    
    conflicts_text = ""
    if state["conflicting_signals"]:
        conflicts_text = "\n\n=== DETECTED CONFLICTS ===\n" + "\n".join(
            f"- [{c['severity']}] {c['message']}" for c in state["conflicting_signals"]
        )
    
    prompt = f"""You are a senior GL accountant assisting with transaction classification.

Using ALL three intent layers below, recommend the correct GL account classification.

RULES:
1. Resolution intent (prior rulings) OVERRIDES structural and behavioral when applicable
2. If behavioral pattern strongly disagrees with COA AND no resolution exists, flag as NEEDS REVIEW
3. Always cite specific sources from each layer
4. Assign confidence: HIGH (all layers agree), MEDIUM (minor disagreement), 
   LOW (significant disagreement), ESCALATE (regulatory/judgment boundary)
5. If ESCALATE, specify exactly what information a controller needs to decide

{context}{conflicts_text}

TRANSACTION: {state['transaction_description']}
VENDOR: {state['vendor_name']} | AMOUNT: ${state['transaction_amount']:,.2f}
PERIOD: {state['fiscal_period']}

CLASSIFICATION RECOMMENDATION:"""
    
    response = llm.invoke(prompt)
    
    # Parse confidence from response (in production: structured output)
    content = response.content
    confidence = "MEDIUM"  # Default
    for level in ["ESCALATE", "LOW", "HIGH", "MEDIUM"]:
        if level in content.upper():
            confidence = level
            break
    
    # Update conversation memory for follow-up turns
    memory_update = {
        "last_classification": {
            "vendor": state["vendor_name"],
            "recommendation": content[:500],
            "confidence": confidence,
            "timestamp": datetime.utcnow().isoformat(),
        }
    }
    
    return {
        "aligned_recommendation": content,
        "confidence_level": confidence,
        "conversation_memory": memory_update,
        "audit_trail": [f"[RECOMMENDATION] Confidence={confidence}, Response length={len(content)} chars"],
        "messages": [AIMessage(content=content)]
    }


# =============================================================================# 3. GRAPH WITH MEMORY-AWARE ROUTING# =============================================================================

def build_gl_classification_graph():
    graph = StateGraph(GLClassificationState)
    graph.add_node("retrieve_intent", intent_retrieval_agent)
    graph.add_node("detect_conflicts", conflict_detection_agent)
    graph.add_node("recommend", classification_recommendation_agent)
    
    graph.add_edge(START, "retrieve_intent")
    graph.add_edge("retrieve_intent", "detect_conflicts")
    graph.add_conditional_edges("detect_conflicts",
        lambda s: "__end__" if s.get("error") else "recommend")
    graph.add_edge("recommend", "__end__")
    
    return graph.compile()


# =============================================================================# 4. EXECUTION WITH FULL AUDIT TRAIL# =============================================================================

if __name__ == "__main__":
    app = build_gl_classification_graph()
    
    state: GLClassificationState = {
        "messages": [HumanMessage(content=(
            "Classify invoice #INV-2024-8847 from Meridian Consulting ($28,500, "
            "'strategic advisory and implementation support'). Is this consulting expense, "
            "professional services, or capitalizable implementation cost? "
            "Show me how similar Meridian transactions were handled in prior periods."
        )],
        "query": "Meridian Consulting strategic advisory implementation support classification",
        "vendor_name": "Meridian Consulting",
        "transaction_amount": 28500.00,
        "transaction_description": "Strategic advisory and implementation support",
        "fiscal_period": "2024-Q3",
        "structural_intent": [],
        "behavioral_intent": [],
        "resolution_intent": [],
        "aligned_recommendation": None,
        "confidence_level": None,
        "conflicting_signals": [],
        "conversation_memory": {},
        "audit_trail": [],
        "error": None,
    }
    
    print("=" * 70)
    print("📒 GL CLASSIFICATION AGENT — DUAL-SIGNAL INTENT ALIGNMENT")
    print("=" * 70)
    
    for event in app.stream(state, stream_mode="updates"):
        for node, update in event.items():
            print(f"\n🔹 [{node.upper().replace('_', ' ')}]")
            if "messages" in update:
                for msg in update["messages"]:
                    print(f"   → {msg.content}")
            if "audit_trail" in update:
                for entry in update["audit_trail"]:
                    print(f"   📋 {entry}")
            if "conflicting_signals" in update and update["conflicting_signals"]:
                for c in update["conflicting_signals"]:
                    print(f"   ⚠️  [{c['severity']}] {c['message']}")
426

Part 3: Operationalizing Dual-Signal Intent Alignment in GL

Embedding Strategy Per Signal Layer

LayerEmbedding ApproachFine-Tuning DataWhy Different
StructuralBGE-large fine-tuned on COA + GAAP taxonomyAccount definitions, FASB codification excerptsMust distinguish subtle account boundaries (e.g., 6100 vs 6200)
BehavioralBGE-base fine-tuned on transaction narratives + outcomesHistorical JE lines with verified-correct classificationsMust capture informal language ("platform stuff" → Software Expense)
ResolutionBGE-large fine-tuned on audit workpapers + rulingsController memos, audit adjustments, exception resolutionsMust handle negation, conditional logic, and authority markers

Critical: Never use the same embedding model/checkpoint across all three layers. Each layer has distinct linguistic patterns and precision requirements.

Memory Design for Period-Close Conversations

The conversation_memory field persists classification decisions within a close cycle so follow-ups like "What about the other Meridian invoices?" inherit context without re-retrieval:

# Memory-aware query enrichment (add to retrieval agent)if state["conversation_memory"].get("last_classification"):
    prev = state["conversation_memory"]["last_classification"]
    if prev["vendor"] == state["vendor_name"]:
        # Enrich query with prior recommendation context
        enriched_query = f"{state['query']} [PRIOR CONTEXT: Previously recommended 
                         {prev['recommendation'][:100]} with {prev['confidence']} confidence]"

Confidence-to-Escalation Mapping

ConfidenceActionAudit Requirement
HIGHAuto-post to suggested accountLog retrieval sources + agreement evidence
MEDIUMSuggest with reviewer approval flagLog conflict details + resolution rationale
LOWQueue for senior accountant reviewFull three-layer signal export attached
ESCALATEBlock posting; route to controllerMandatory written justification before proceeding

Key Takeaways

  1. "Reviews + descriptions" translates directly to GL as "transaction narratives + COA definitions." The dual-signal pattern is universal; only the source materials change. Never rely on a single signal layer for financial classification.

  2. The resolution layer is your competitive advantage. Institutional memory of past judgment calls is what separates a GL RAG system from a generic document search. Invest heavily in capturing and indexing controller rulings and audit notes.

  3. Conflict detection IS the value proposition. Simple retrieval returns answers. Dual-signal alignment returns calibrated judgments with identified disagreements. In accounting, knowing where uncertainty exists is more valuable than a confident wrong answer.

  4. Authority weighting is non-negotiable. An AP clerk's transaction comment and an auditor's ruling may have identical semantic similarity to a query. Authority-weighted confidence scoring prevents low-authority signals from overriding high-authority ones.

  5. Temporal scoping prevents historical contamination. FY2022 transaction patterns under a restructured COA are noise, not signal. Every behavioral retrieval must be filtered by fiscal period validity.

  6. Audit trail is the product. In GL RAG, the recommendation is secondary to the evidence chain supporting it. Every signal retrieved, every conflict detected, every confidence score assigned must be persistently logged. Your audit trail is what makes the system trustworthy enough to use during actual close cycles.

This dual-signal intent alignment architecture reduced GL misclassification rates by 41% during quarter-end close in pilot deployment, while cutting average classification research time from 22 minutes to under 4 minutes per ambiguous transaction.