Multi-Agent Loan Processing AgenticAI
In this blog post, we will explore the implementation of a multi-agent loan processing system using the crewAI framework. This innovative approach leverages advanced artificial intelligence to streamline and automate the various stages of loan processing, ensuring efficiency and accuracy.
We will employ a hierarchical architecture, commonly referred to as the Supervisor/Orchestrator pattern. This design effectively organizes agents into a structured framework that mimics a corporate team environment. Within this framework, a manager (the supervisor) oversees a group of specialized agents (the specialists), each with specific expertise.
Supervisor (Parent) Agent
The Supervisor agent serves as the manager of the system. Its primary responsibilities encompass several crucial tasks, including:
Planning the Workflow: The Supervisor outlines the entire loan processing cycle, establishing the order and priority of tasks to be completed.
Delegating Sub-Tasks: Based on the nature of each loan application, the Supervisor assigns specific tasks to the appropriate specialist agents, ensuring that each agent is tasked with functions aligned with their expertise.
Monitoring Progress: Continuous oversight allows the Supervisor to track the status of each task, providing updates on progress and identifying any potential issues or bottlenecks in the process.
Synthesizing Results: After all tasks have been executed, the Supervisor consolidates the findings from the specialists to create a comprehensive overview of the loan application status.
Specialist (Sub-Agent)
Specialist agents are the individual experts within the multi-agent system. Each specialist is programmed to perform a particular task with a high degree of proficiency. Their roles include:
Performing Specific Tasks: Each specialist is designed for tasks such as document validation, credit score retrieval, risk assessment, and compliance checking. This specialization enables them to execute their functions efficiently and effectively.
Executing Under Supervision: Upon receiving a task from the Supervisor, each specialist utilizes its dedicated tools and algorithms to complete the assigned job, ensuring that the output is accurate and meets the necessary requirements.
Returning Results: Once a task is completed, the specialist sends the result back to the Supervisor for further processing and analysis.
Hierarchical Agent system is shown below
Workflow Process
The overall workflow within this multi-agent system is divided into several distinct tasks, each mapped to a specific component of the process:
Document Fetch: The first step involves retrieving the content of the loan application document. This is done at the preprocessing stage, where the document is fetched and made available for subsequent validation and analysis.
Document Validation: Once the document content is fetched, the Document Validation Specialist checks its validity and completeness. This step ensures that all required information is present and adheres to the predefined standards.
Credit Check: Following validation, the Credit Check Agent is tasked with retrieving the borrower’s credit score. This process relies on the borrower’s unique customer ID to access accurate credit information, which is vital for assessing the borrower’s creditworthiness.
Risk Assessment: The next phase is conducted by the Risk Assessment Analyst, who analyzes the data gathered from the document, credit score, and borrower’s income. This analysis determines the risk level associated with granting the loan.
Compliance Check: Finally, the Compliance Check Agent evaluates the entire decision-making process to ensure that it aligns with established lending regulations and legal requirements, safeguarding the institution against potential compliance issues.
By structuring the loan processing system in this way, we can ensure a streamlined and effective approach to handling loan applications. Each agent plays a crucial role in contributing to a well-coordinated effort, ultimately enhancing the quality and speed of the loan processing workflow.
According to our workflow our orchestrator agent and four specialist agents will look like this.
CrewAI framework for building AgenticAI Loan-procesing application
Key concepts in CrewAI encompass several crucial elements that define how the system operates and how agents collaborate effectively:
Agents: Agents are the core components of CrewAI, each uniquely defined by a specific role, goal, and backstory. They utilize a language model (LLM) tailored to their persona, which allows them to communicate in a way that aligns with their expertise or field of knowledge. Additionally, agents may have access to specialized tools that enhance their capabilities. This role-playing aspect not only helps agents embody their defined personas but also enables nuanced interactions and more effective problem-solving based on their areas of expertise.
Tasks: Within CrewAI, tasks are well-defined assignments given to agents, each with a clear description and expected outcomes. Tasks vary in complexity and purpose, and they are assigned to specific agents based on their roles and skill sets. Additionally, tasks can be structured in a manner that allows for chaining—where the output of one task can seamlessly serve as the input for another task. This chaining mechanism facilitates efficient workflows and enhances the overall productivity of the agents.
Tools: Tools serve as essential functions or capabilities that agents can use to interact with external systems or execute specific actions. Examples of these tools include web search capabilities, API access, and data processing functions. In the CrewAI framework, tools often derive from a BaseTool class, ensuring consistency and reliability in their performance. The availability of various tools empowers agents to perform diverse tasks more effectively, expanding their range of capabilities.
Crew: The crew represents the collaborative assembly of agents working together to accomplish a set of objectives or tasks. The composition of the crew is vital, as it determines how well agents can leverage each other's strengths and skills. Effective collaboration within the crew relies on clear communication and the coordinated execution of tasks, enhancing the overall effectiveness of the team.
Process: The process describes the systematic workflow or methodology that the crew adheres to in order to carry out tasks. Various processes may be utilized, including sequential execution, where tasks are completed one after the other, or hierarchical execution, where a lead or manager agent is responsible for delegating tasks to ensure an orderly approach. The defined process is critical for maintaining organization and efficiency in task execution, thereby enabling the crew to meet its objectives strategically and systematically.
Pre-Requisites
VSCode - https://code.visualstudio.com/download
Install Python --- https://www.python.org/downloads/
CrewAI Installation:
Run following commands
python -m .venv ( to create virtual environment)
source .venv/bin/activate ( to activate virtual environment)
pip install uv (Python package installer)
uv --version (to check version)
uv tool install crewai (install the crewai )
uv tool update-shell (to set path)
Create Project
crewai create crew loan_processing ( to create skelaton project loan_processing)
This will create project in structure as shown below
Create requirements.txt file for setting up all libraries and save it at the root of the project
boto3 crewai[tools] crewai-tools[mcp] streamlit==1.49.1 ratelimit tenacityinstall the libraries by running
pip install -r requirements.txt
Update the agents.yml file to add agents for the app as shown below
doc_specialist:
role: >
Document Validation Specialist
goal: >
Validate the completeness and format of a new loan application provided as a JSON string.
backstory: >
You are a meticulous agent responsible for the first step of loan processing.
credit_analyst:
role: >
Credit Check Agent
goal: >
Query the credit bureau API to retrieve an applicant's credit score.
backstory: >
You are a specialized agent that interacts with the Credit Bureau.
risk_assessor:
role: >
Risk Assessment Analyst
goal: >
Calculate the financial risk score for a loan application.
backstory: >
You are a quantitative analyst agent.
compliance_officer:
role: >
Compliance Officer
goal: >
Check the application against all internal lending policies and compliance rules.
backstory: >
You are the final checkpoint for policy and compliance.
manager:
role: >
Loan Processing Manager
goal: >
Manage the loan application workflow and compile the final report.
backstory: >
You are the manager responsible for orchestrating the loan processing pipeline
Update the tasks.yml file to add tasks
task_validate:
description: >
Validate the loan application provided as a JSON string: '{document_content}'.
Pass this string to the 'Validate Document Fields' tool.
expected_output: >
A JSON string with the validation status
agent: doc_specialist
task_credit:
description: >
Extract customer_id and call Query Credit Bureau API.
expected_output: >
A JSON string containing the credit_score.
agent: credit_analyst
context:
- task_validate
task_risk:
description: >
Extract loan details and credit score, then Calculate Risk Score..
expected_output: >
A JSON string containing the risk_score.
agent: risk_assessor
context:
- task_validate
- task_credit
task_compliance:
description: >
Check Lending Compliance based on history and risk score.
expected_output: >
Compliance status JSON.
agent: compliance_officer
context:
- task_validate
- task_risk
task_report:
description: >
Compile a final report with Approve/Deny decision.
expected_output: >
Markdown report.
agent: manager
context:
- task_validate
- task_credit
- task_risk
- task_compliance
Update the crew.py file to include the following configurations in detail:
Create an LLM: Implement the necessary code to initialize and configure the LLM (Language Model), for our case I have used OpenAI LLM.
Configure Agents: Within the @CrewBase section, use the @agent decorator to define and configure all agents that will be part of the crew. Ensure that each agent has clear roles and responsibilities outlined.
Define Tasks: Similarly, utilize the @task decorator to specify all tasks that agents will be responsible for. Make sure each task is well-defined and includes the necessary parameters and execution criteria.
Assemble the Crew: Use the @crew decorator to bring together the agents, tasks, and a designated manager. The assembly should clearly delineate how each component interacts and their roles within the overall crew structure.
Define Tools: Finally, clearly specify all the required tools needed for the operation using the @tools decorator. Ensure that each tool is associated with the relevant agent or task for seamless integration.
With these changes we have created a well-structured and functional crew configuration within the crew.py file.
from crewai import LLM, Agent, Crew, Process, Task
from crewai.project import CrewBase, agent, crew, task
from crewai.agents.agent_builder.base_agent import BaseAgent
from typing import List
from crewai.tools import tool
import json
import sys
from datetime import datetime
import os
#load the OPENAI_API_KEY from environment variable or set it directly
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
llm = LLM(
model=os.getenv("MODEL", "gpt-4o"), # Default to gpt-4o if MODEL env variable is not set
api_key=OPENAI_API_KEY, # Or set OPENAI_API_KEY
temperature=0.0,
max_tokens=1000,
)
@CrewBase
class LoanProcessing():
"""LoanProcessing crew"""
#@title Run CrewAI
agents: List[BaseAgent]
tasks: List[Task]
@agent
def doc_specialist(self) -> Agent:
return Agent(
config=self.agents_config['doc_specialist'], # type: ignore[index]
verbose=True,
tools=[self.ValidateDocumentFieldsTool],
llm=llm
)
@agent
def credit_analyst(self) -> Agent:
return Agent(
config=self.agents_config['credit_analyst'], # type: ignore[index]
verbose=True,
tools=[self.QueryCreditBureauAPITool],
llm=llm
)
@agent
def risk_assessor(self) -> Agent:
return Agent(
config=self.agents_config['risk_assessor'], # type: ignore[index]
verbose=True,
tools=[self.CalculateRiskScoreTool],
llm=llm
)
@agent
def compliance_officer(self) -> Agent:
return Agent(
config=self.agents_config['compliance_officer'], # type: ignore[index]
verbose=True,
tools=[self.CheckLendingComplianceTool],
llm=llm
)
@agent
def manager(self) -> Agent:
return Agent(
config=self.agents_config['manager'], # type: ignore[index]
verbose=True,
llm=llm,
allow_delegation=True
)
@task
def task_validate(self) -> Task:
return Task(
config=self.tasks_config['task_validate'], # type: ignore[index]
)
@task
def task_credit(self) -> Task:
return Task(
config=self.tasks_config['task_credit'], # type: ignore[index]
)
@task
def task_risk(self) -> Task:
return Task(
config=self.tasks_config['task_risk'], # type: ignore[index]
)
@task
def task_compliance(self) -> Task:
return Task(
config=self.tasks_config['task_compliance'], # type: ignore[index]
)
@task
def task_report(self) -> Task:
return Task(
config=self.tasks_config['task_report'], # type: ignore[index]
allow_delegation=False
)
@crew
def crew(self) -> Crew:
"""Creates the LoanProcessing crew"""
# To learn how to add knowledge sources to your crew, check out the documentation:
# https://docs.crewai.com/concepts/knowledge#what-is-knowledge
return Crew(
agents=[self.doc_specialist(), self.credit_analyst(), self.risk_assessor(), self.compliance_officer()], # Automatically created by the @agent decorator
tasks=self.tasks, # Automatically created by the @task decorator
manager_agent=self.manager(), # Automatically created by the @agent decorator
process=Process.hierarchical,
verbose=True
)
@tool
def ValidateDocumentFieldsTool( application_data: str) -> str:
"""Validates JSON application data."""
name: str = "Validate Document Fields"
description: str = "Validates JSON application data."
print(f"--- TOOL: Validating document fields ---"+f" (Data: {application_data}) ---")
try:
data = json.loads(application_data)
required = ["customer_id", "loan_amount", "income", "credit_history"]
missing = [f for f in required if f not in data]
if missing:
return json.dumps({"error": f"Missing fields: {', '.join(missing)}"})
return json.dumps({"status": "validated", "data": data})
except:
return json.dumps({"error": "Invalid JSON"})
@tool
def QueryCreditBureauAPITool(customer_id: str) -> str:
"""Gets credit score for customer_id."""
name: str = "Query Credit Bureau API"
description: str = "Gets credit score for customer_id."
print(f"--- TOOL: Calling Credit Bureau for {customer_id} ---")
scores = {
"CUST-12345": 810, # Good
"CUST-99999": 550, # BAD SCORE (< 600)
"CUST-55555": 620
}
score = scores.get(customer_id)
if score:
return json.dumps({"customer_id": customer_id, "credit_score": score})
return json.dumps({"error": "Customer not found"})
@tool
def CalculateRiskScoreTool(loan_amount: int, income: str, credit_score: int) -> str:
"""Calculates risk based on financial data."""
name: str = "Calculate Risk Score"
description: str = "Calculates risk based on financial data."
print(f"--- TOOL: Calculating Risk (Score: {credit_score}) ---")
# Logic: Credit Score < 600 is automatic HIGH risk
if credit_score < 600:
return json.dumps({"risk_score": 9, "reason": "Credit score too low"})
# Standard logic
try:
inc_val = int(''.join(filter(str.isdigit, income)))
ann_inc = inc_val * 12 if "month" in income.lower() else inc_val
except: ann_inc = 0
risk = 1
if credit_score < 720: risk += 2
if ann_inc > 0 and (loan_amount / ann_inc) > 0.5: risk += 3
return json.dumps({"risk_score": min(risk, 10)})
@tool
def CheckLendingComplianceTool(loan_amount: int, risk_score: int) -> str:
"""Checks if loan complies with lending rules."""
name: str = "Check Lending Compliance"
description: str = "Checks if loan complies with lending rules."
print(f"--- TOOL: Checking Lending Compliance ---")
# Simple compliance logic for demo
if loan_amount > 500000:
return json.dumps({"compliant": False, "reason": "Loan amount exceeds limit"})
if risk_score >= 7:
return json.dumps({"compliant": False, "reason": "Risk score too high"})
return json.dumps({"compliant": True})
We are about to enhance the main.py file of the Crewai application by implementing a run method. This method will serve as the starting point for the Crewai app, taking in the necessary inputs to initialize the application.
The input we will use is a JSON file that features a key named document_content. This content is structured using several important attributes, which include:
customer_id: A unique identifier for each customer.
loan_amount: The total amount of money requested for the loan.
income: The customer's declared income, which helps assess their ability to repay the loan.
credit_history: A record of the customer's credit behavior, indicating their reliability in managing debt.
These attributes will be fetched using a specific identifier known as document_id, ensuring that we are working with the correct data for each individual customer. This setup will allow the Crewai app to effectively process loan requests based on the provided information.
code is shown below
#!/usr/bin/env python
import sys
import warnings
from datetime import datetime
import json
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception
from ratelimit import limits, sleep_and_retry
import time
from loan_processing.crew import LoanProcessing
# --- CONFIGURATION ---
CALLS = 15 # Max calls...
PERIOD = 60 # ...per minute
loan_application_inputs_valid = {
"applicant_id": "borrower_good_780",
"document_id": "document_valid_123"
}
loan_application_inputs_invalid = {
"applicant_id": "borrower_bad_620",
"document_id": "document_invalid_456"
}
warnings.filterwarnings("ignore", category=SyntaxWarning, module="pysbd")
def run():
"""
Run the crew.
"""
inputs = {
'topic': 'AI LLMs',
'current_year': str(datetime.now().year)
}
try:
print("--- KICKING OFF CREWAI (VALID INPUTS) ---")
valid_json = get_document_content(loan_application_inputs_valid['document_id'])
inputs = {'document_content': valid_json}
robust_execute(LoanProcessing().crew().kickoff, inputs=inputs)
except Exception as e:
import traceback
traceback.print_exc()
handle_execution_error(e)
# --- 1. HELPER: Mock Document Fetcher ---
def get_document_content(document_id: str) -> str:
print(f"--- HELPER: Simulating fetch for doc_id: {document_id} ---")
if document_id == "document_valid_123":
# Happy Path: High Income, Good History
return json.dumps({
"customer_id": "CUST-12345",
"loan_amount": 50000,
"income": "USD 120000 a year",
"credit_history": "7 years good standing"
})
elif document_id == "document_risky_789":
# Unhappy Path: Valid Docs, but LOW CREDIT SCORE
return json.dumps({
"customer_id": "CUST-99999",
"loan_amount": 50000,
"income": "USD 40000 a year",
"credit_history": "Recent Missed Payments"
})
elif document_id == "document_invalid_456":
# Broken Path: Missing fields (income)
return json.dumps({
"customer_id": "CUST-55555",
"loan_amount": 200000,
"credit_history": "1 year"
})
else:
return json.dumps({"error": "Document ID not found."})
# --- HELPER: ERROR FILTER ---
def is_rate_limit_error(e):
msg = str(e).lower()
return "429" in msg or "quota" in msg or "resource exhausted" in msg or "serviceunavailable" in msg
# --- ROBUST WRAPPER ---
@sleep_and_retry
@limits(calls=CALLS, period=PERIOD)
@retry(
stop=stop_after_attempt(5),
wait=wait_exponential(multiplier=2, min=4, max=30),
retry=retry_if_exception(is_rate_limit_error),
reraise=True
)
def robust_execute(func, *args, **kwargs):
"""
Executes any function (CrewAI kickoff, LangGraph invoke) with built-in
rate limiting and auto-retries for transient API errors.
"""
print(f" >> [Clock {time.strftime('%X')}] Executing Agent Action (Safe Mode)...")
return func(*args, **kwargs)
# --- ERROR HANDLER ---
def handle_execution_error(e):
"""Prints a clean, professional error report."""
error_msg = str(e)
is_quota = "429" in error_msg or "quota" in error_msg.lower()
print("\n" + "━" * 60)
print(" 🛑 MISSION ABORTED: SYSTEM CRITICAL ERROR")
print("━" * 60)
if is_quota:
print(" ⚠️ CAUSE: QUOTA EXCEEDED (API Refusal)")
print(" 🔍 CONTEXT: The LLM provider rejected the request.")
print("\n 🛠️ ACTION: [1] Wait before retrying")
print(" [2] Check API Limits (Free Tier is ~15 RPM)")
else:
print(f" ⚠️ CAUSE: UNEXPECTED EXCEPTION")
print(f" 📝 DETAILS: {error_msg}")
print("━" * 60 + "\n")
Now run the app with command
crewai run
Github Repo
Code for this project is available at multi-agent-crewai loan processing app
