Building Multi-Agent AI Systems for Banking: Advanced Workflows and Agent Coordination with CrewAI (Part 3)
Implementing customer service automation and credit risk assessment with hierarchical agent teams
In Part 1, we explored the theory behind agentic AI and why multi-agent systems matter in banking.
In Part 2, we built a sequential fraud detection system where agents worked one after another like an assembly line. That pattern works well for linear workflows, but it breaks down when you need dynamic task allocation, parallel processing, or adaptive coordination based on intermediate results.
Today, we move to hierarchical workflows where a manager agent coordinates multiple specialists, delegates tasks based on complexity, and synthesizes results from parallel execution paths. This mirrors how actual banking operations work. A supervisor does not process every task sequentially. They assess what needs to be done, assign specialists, monitor progress, and make decisions based on collective insights.
We will build two complete systems that demonstrate advanced coordination patterns.
- First, a customer service automation system with five agents handling query routing, account access, problem resolution, and escalation.
- Second, a credit risk assessment system with four agents performing parallel analysis of financial data, industry research, policy compliance, and final decision synthesis.
By the end of this guide, you will understand when to use hierarchical versus sequential workflows, how to implement manager agents that coordinate specialists, and how to optimize costs by mixing different LLM providers based on task complexity.
This is part of the series Building Production-Ready Multi-Agent AI Systems for Banking: A Complete Guide, where we walk through practical architectures, orchestration patterns, memory systems, tool integrations, and production deployment strategies for building enterprise-grade banking AI systems using CrewAI and modern agentic AI frameworks.
If you have not read Part 1 and Part 2, start there. This article assumes you understand agent fundamentals and have experience with sequential workflows.
Let’s build something more sophisticated.
Sequential vs Hierarchical Workflows: When to Use Each
The workflow pattern you choose fundamentally shapes system behavior and performance. Let me explain the tradeoffs.
Sequential Workflows
In sequential execution, tasks run in a fixed order. Agent A completes its task, passes results to Agent B, who completes its task and passes results to Agent C. Each agent waits for the previous agent to finish before starting.
When sequential works well:
- Linear processes with clear dependencies (document analysis, report generation)
- Tasks where order matters and cannot be parallelized (validation before execution)
- Simple workflows with 2–3 steps
- Predictable resource usage and timing
Sequential limitations:
- No parallelism means longer total execution time
- Cannot adapt to intermediate results (stuck with the original plan)
- All agents must complete even if early results suggest a different path
- Rigid structure does not match dynamic real-world processes
We used sequential workflow in Part 2 because fraud detection naturally follows investigation, analysis, and decision steps in that order.
Hierarchical Workflows
In hierarchical execution, a manager agent receives the high-level goal, breaks it into subtasks, delegates to specialist agents, monitors their work, and synthesizes results. Specialists can work in parallel. The manager can dynamically adjust the plan based on intermediate findings.
When hierarchical excels:
- Complex problems requiring multiple perspectives (credit assessment, compliance review)
- Tasks that can be parallelized (multiple data sources, independent analyses)
- Scenarios where strategy adapts based on findings (investigation with unknown scope)
- Systems needing central coordination and quality control
Hierarchical considerations:
- More LLM calls for manager coordination (higher cost)
- Manager must be skilled at delegation and synthesis
- Requires robust context management across agents
- More complex to debug when things go wrong
Most production banking systems I build use hierarchical workflows because real banking operations are rarely linear.
The Manager Agent Pattern
A manager agent is not just another specialist. It operates at a higher level of abstraction, coordinating work rather than performing it directly.
Manager Responsibilities
A good manager agent handles five key functions:
- Planning: Breaks down the high-level goal into concrete subtasks that specialists can execute.
- Delegation: Assigns tasks to appropriate specialists based on their expertise and current workload.
- Monitoring: Tracks progress, identifies blockers, and determines when tasks are complete versus when they need iteration.
- Synthesis: Combines outputs from multiple specialists into a coherent result that addresses the original goal.
- Adaptation: Adjusts strategy based on intermediate findings, adding new tasks or changing priorities as needed.
Configuring Manager Agents in CrewAI
CrewAI provides built-in support for hierarchical workflows. Here is how to configure a manager:
from crewai import Crew, Process, Agent
# Create specialist agents (we will define these fully later)
specialist_agents = [agent1, agent2, agent3]
# Create crew with hierarchical process
crew = Crew(
agents=specialist_agents,
tasks=task_list,
process=Process.hierarchical,
manager_llm=get_openai_llm(model="gpt-4") # Manager needs strong reasoning
)
The key difference is process=Process.hierarchical and specifying manager_llm. CrewAI automatically creates a manager agent that coordinates the specialist agents. You can also create the manager explicitly for more control:
manager = Agent(
role="Operations Manager",
goal="Coordinate specialist agents to achieve the overall objective efficiently",
backstory="""You are an experienced operations manager who understands
how to break down complex problems, delegate to specialists, and synthesize
their findings into actionable results.""",
llm=get_openai_llm(model="gpt-4"),
allow_delegation=True # Critical: manager can delegate to other agents
)
The allow_delegation=True flag is what makes an agent a manager. Without it, the agent cannot assign work to others.
Use Case 1: Customer Service Automation System
Let me show you a realistic customer service system that routes queries, accesses accounts, resolves problems, and escalates when needed.
System Architecture
We need five agents:
- Intake Agent: Receives customer queries, classifies urgency and category
- Account Agent: Accesses customer data and account history
- Resolution Agent: Solves common problems (password reset, statement requests)
- Technical Agent: Handles complex technical issues
- Manager Agent: Coordinates the workflow and makes escalation decisions
Building the Tools
Create customer_service_tools.py:
from crewai_tools import tool
import json
from datetime import datetime, timedelta
@tool("Classify Customer Query")
def classify_query(query_text: str) -> dict:
"""
Classify customer query by category and urgency.
Returns category, urgency level, and suggested routing.
"""
query_lower = query_text.lower()
# Category classification
if any(word in query_lower for word in ["password", "login", "access", "locked"]):
category = "authentication"
urgency = "high"
elif any(word in query_lower for word in ["statement", "balance", "transaction"]):
category = "account_inquiry"
urgency = "medium"
elif any(word in query_lower for word in ["fraud", "unauthorized", "suspicious"]):
category = "fraud_concern"
urgency = "critical"
elif any(word in query_lower for word in ["loan", "credit", "rate", "application"]):
category = "credit_services"
urgency = "medium"
elif any(word in query_lower for word in ["technical", "error", "bug", "not working"]):
category = "technical_issue"
urgency = "high"
else:
category = "general_inquiry"
urgency = "low"
# Routing suggestion
routing = {
"authentication": "resolution_agent",
"account_inquiry": "account_agent",
"fraud_concern": "escalate_to_fraud_team",
"credit_services": "credit_specialist",
"technical_issue": "technical_agent",
"general_inquiry": "resolution_agent"
}
return {
"category": category,
"urgency": urgency,
"suggested_routing": routing.get(category, "manager_review"),
"keywords_detected": [word for word in ["password", "fraud", "error", "loan"]
if word in query_lower]
}
@tool("Access Customer Account")
def access_customer_account(customer_id: str) -> dict:
"""
Retrieve customer account information and recent activity.
Returns account status, balance, recent transactions, and alerts.
"""
# Simulated account data
accounts = {
"CUST12345": {
"customer_id": "CUST12345",
"name": "John Smith",
"account_type": "Premium Checking",
"status": "active",
"balance": 15420.50,
"last_login": "2024-03-24T10:15:00Z",
"recent_transactions": [
{"date": "2024-03-25", "description": "ATM Withdrawal", "amount": -200.00},
{"date": "2024-03-24", "description": "Salary Deposit", "amount": 5500.00},
{"date": "2024-03-23", "description": "Online Purchase", "amount": -89.99}
],
"alerts": [],
"customer_since": "2019-06-15",
"contact_email": "john.smith@email.com",
"contact_phone": "+1-555-0123"
},
"CUST67890": {
"customer_id": "CUST67890",
"name": "Sarah Johnson",
"account_type": "Business Checking",
"status": "active",
"balance": 45230.75,
"last_login": "2024-03-25T14:30:00Z",
"recent_transactions": [
{"date": "2024-03-25", "description": "Wire Transfer", "amount": -15000.00},
{"date": "2024-03-24", "description": "Deposit", "amount": 22000.00}
],
"alerts": ["High-value wire transfer pending verification"],
"customer_since": "2021-03-10",
"contact_email": "sarah.j@business.com",
"contact_phone": "+1-555-0456"
}
}
return accounts.get(customer_id, {"error": "Customer not found"})
@tool("Execute Account Action")
def execute_account_action(customer_id: str, action: str, parameters: dict = None) -> dict:
"""
Execute common account actions like password reset, statement generation, etc.
Returns confirmation and next steps.
"""
if parameters is None:
parameters = {}
valid_actions = {
"reset_password": "Password reset initiated. Temporary password sent to registered email.",
"request_statement": f"Statement for period {parameters.get('period', 'last 30 days')} generated and sent to email.",
"update_contact": "Contact information update request submitted. Verification required.",
"report_fraud": "Fraud report filed. Case number FR-2024-12345 created. Fraud team will contact within 2 hours.",
"dispute_transaction": "Transaction dispute initiated. Reference number DT-2024-67890. Resolution within 10 business days."
}
if action not in valid_actions:
return {
"success": False,
"message": f"Unknown action: {action}",
"available_actions": list(valid_actions.keys())
}
return {
"success": True,
"action": action,
"customer_id": customer_id,
"message": valid_actions[action],
"timestamp": datetime.now().isoformat(),
"reference_number": f"REF-{hash(action + customer_id) % 1000000}"
}
@tool("Check Knowledge Base")
def check_knowledge_base(query: str) -> dict:
"""
Search internal knowledge base for solutions and documentation.
Returns relevant articles and step-by-step solutions.
"""
# Simulated knowledge base
knowledge_base = {
"password reset": {
"article_id": "KB001",
"title": "How to Reset Your Password",
"solution": "1. Click 'Forgot Password' on login page\n2. Enter registered email\n3. Check email for reset link\n4. Create new password (min 12 characters)",
"related_articles": ["KB002: Account Security", "KB015: Two-Factor Authentication"]
},
"transaction dispute": {
"article_id": "KB045",
"title": "Disputing a Transaction",
"solution": "1. Identify the disputed transaction\n2. Gather supporting documentation\n3. Submit dispute form within 60 days\n4. Wait for investigation (5-10 business days)",
"related_articles": ["KB046: Fraud Protection", "KB047: Chargeback Process"]
},
"mobile app error": {
"article_id": "KB123",
"title": "Troubleshooting Mobile App Issues",
"solution": "1. Clear app cache\n2. Update to latest version\n3. Check internet connection\n4. Restart device\n5. Reinstall if problem persists",
"related_articles": ["KB124: App Features", "KB125: Mobile Security"]
}
}
query_lower = query.lower()
for key, article in knowledge_base.items():
if key in query_lower:
return article
return {
"article_id": None,
"title": "No exact match found",
"solution": "Please contact customer support for personalized assistance.",
"related_articles": []
}
Defining the Agent Team
Create customer_service_agents.py:
from crewai import Agent
from llm_config import get_openai_llm, get_anthropic_llm, get_ollama_llm
from customer_service_tools import (
classify_query,
access_customer_account,
execute_account_action,
check_knowledge_base
)
def create_customer_service_agents():
"""
Create a hierarchical customer service team.
Uses different LLMs based on task complexity for cost optimization.
"""
# Intake Agent - Simple classification task, use fast/cheap model
intake_agent = Agent(
role="Customer Query Intake Specialist",
goal="Quickly classify and route customer queries to appropriate handlers",
backstory="""You are a front-line customer service agent who excels at
understanding customer needs and routing them efficiently. You identify
urgency levels and suggest the best specialist to handle each query.""",
tools=[classify_query],
llm=get_ollama_llm(model="llama3.1:8b"), # Fast local model for simple task
verbose=True,
allow_delegation=False
)
# Account Agent - Data retrieval, use mid-tier model
account_agent = Agent(
role="Account Information Specialist",
goal="Access and explain customer account information accurately",
backstory="""You are an account specialist with 8 years of experience.
You retrieve customer data, explain account activity, and identify any
issues that need attention. You present information clearly and highlight
anything unusual.""",
tools=[access_customer_account],
llm=get_anthropic_llm(model="claude-3-5-haiku-20241022"), # Fast, affordable
verbose=True,
allow_delegation=False
)
# Resolution Agent - Common problem solving
resolution_agent = Agent(
role="Problem Resolution Specialist",
goal="Resolve common customer issues efficiently using standard procedures",
backstory="""You are a problem-solving expert who handles routine customer
requests like password resets, statement requests, and contact updates.
You follow documented procedures, execute actions accurately, and provide
clear confirmation to customers.""",
tools=[execute_account_action, check_knowledge_base],
llm=get_anthropic_llm(model="claude-3-5-haiku-20241022"),
verbose=True,
allow_delegation=False
)
# Technical Agent - Complex issues, use stronger model
technical_agent = Agent(
role="Technical Support Specialist",
goal="Diagnose and resolve technical issues with banking systems and applications",
backstory="""You are a senior technical support engineer with deep knowledge
of banking systems, mobile applications, and online platforms. You troubleshoot
complex technical problems, provide detailed solutions, and escalate system-wide
issues when necessary.""",
tools=[check_knowledge_base],
llm=get_anthropic_llm(model="claude-3-5-sonnet-20241022"), # Stronger reasoning
verbose=True,
allow_delegation=False
)
# Manager Agent - Coordination and decision-making, use best model
manager_agent = Agent(
role="Customer Service Manager",
goal="Coordinate specialists to resolve customer queries efficiently and ensure quality",
backstory="""You are an experienced customer service manager who oversees
a team of specialists. You assess situations, delegate tasks appropriately,
make escalation decisions, and ensure customers receive excellent service.
You balance efficiency with quality and know when to escalate to human staff.""",
tools=[], # Manager coordinates but doesn't execute actions directly
llm=get_openai_llm(model="gpt-4"), # Best model for coordination
verbose=True,
allow_delegation=True # This makes it a manager
)
return {
"intake": intake_agent,
"account": account_agent,
"resolution": resolution_agent,
"technical": technical_agent,
"manager": manager_agent
}
Notice the LLM selection strategy. The intake agent uses a local Llama model because classification is straightforward. Account and resolution agents use Claude Haiku for good performance at low cost. The technical agent uses Claude Sonnet for complex reasoning. The manager uses GPT-4 for sophisticated coordination. This mixed approach optimizes for both quality and cost.
Creating the Hierarchical Workflow
Create customer_service_tasks.py:
from crewai import Task
def create_customer_service_tasks(agents, customer_query, customer_id):
"""
Create tasks for hierarchical customer service workflow.
Manager will coordinate these tasks dynamically.
"""
# Task 1: Classify and Route Query
intake_task = Task(
description=f"""Analyze this customer query and determine routing:
Customer Query: "{customer_query}"
Customer ID: {customer_id}
Use the classification tool to:
1. Identify the query category
2. Assess urgency level
3. Recommend routing to appropriate specialist
4. Extract key information for the handling agent""",
expected_output="""Query classification containing:
- Category (authentication, account_inquiry, fraud_concern, etc.)
- Urgency level (low, medium, high, critical)
- Recommended routing
- Key information extracted from query""",
agent=agents["intake"]
)
# Task 2: Retrieve Account Context
account_task = Task(
description=f"""Retrieve account information for customer {customer_id}.
Provide:
1. Current account status and balance
2. Recent transaction history
3. Any active alerts or issues
4. Relevant account details for addressing the query""",
expected_output="""Account summary containing:
- Account status and balance
- Recent activity
- Any alerts or concerns
- Customer profile information""",
agent=agents["account"],
context=[intake_task] # Depends on intake classification
)
# Task 3: Resolve or Escalate
resolution_task = Task(
description="""Based on the query classification and account context,
either resolve the customer issue or prepare for escalation.
If resolvable:
1. Execute appropriate account action
2. Provide clear confirmation to customer
3. Document resolution
If escalation needed:
1. Summarize the situation
2. Identify escalation path
3. Prepare handoff information""",
expected_output="""Resolution report containing:
- Action taken (if resolved)
- Customer communication (what to tell the customer)
- Escalation recommendation (if needed)
- Reference numbers or case IDs""",
agent=agents["resolution"],
context=[intake_task, account_task]
)
# Task 4: Technical Support (conditional, used if query is technical)
technical_task = Task(
description="""Provide technical support for complex system or application issues.
1. Review the query and account context
2. Search knowledge base for solutions
3. Provide step-by-step troubleshooting
4. Escalate to engineering if system-wide issue""",
expected_output="""Technical support response containing:
- Problem diagnosis
- Step-by-step solution
- Knowledge base article references
- Escalation flag if needed""",
agent=agents["technical"],
context=[intake_task, account_task]
)
return [intake_task, account_task, resolution_task, technical_task]
Running the Customer Service System
Create customer_service_main.py:
from crewai import Crew, Process
from customer_service_agents import create_customer_service_agents
from customer_service_tasks import create_customer_service_tasks
def handle_customer_query(query: str, customer_id: str):
"""
Process customer service query using hierarchical workflow.
"""
print(f"\n{'='*80}")
print(f"CUSTOMER SERVICE - Query Processing")
print(f"Customer: {customer_id}")
print(f"Query: {query}")
print(f"{'='*80}\n")
# Create the agent team
agents = create_customer_service_agents()
# Create tasks
tasks = create_customer_service_tasks(agents, query, customer_id)
# Create hierarchical crew
# Manager agent coordinates specialists automatically
crew = Crew(
agents=[
agents["intake"],
agents["account"],
agents["resolution"],
agents["technical"]
],
tasks=tasks,
process=Process.hierarchical,
manager_llm=agents["manager"].llm, # Use manager's LLM for coordination
verbose=True
)
# Execute
try:
result = crew.kickoff()
print(f"\n{'='*80}")
print("FINAL RESPONSE TO CUSTOMER")
print(f"{'='*80}\n")
print(result)
return result
except Exception as e:
print(f"\nError processing query: {str(e)}")
return None
if __name__ == "__main__":
# Example 1: Password reset request
result1 = handle_customer_query(
query="I can't log into my account. I think I forgot my password.",
customer_id="CUST12345"
)
# Example 2: Account inquiry
# result2 = handle_customer_query(
# query="What's my current balance and recent transactions?",
# customer_id="CUST12345"
# )
# Example 3: Technical issue
# result3 = handle_customer_query(
# query="The mobile app keeps crashing when I try to view statements.",
# customer_id="CUST67890"
# )
When you run this, the manager agent receives the high-level goal (resolve customer query), breaks it into subtasks (classify, retrieve account, resolve), assigns them to specialists, monitors execution, and synthesizes the final response. If the query requires technical support, the manager routes it appropriately. If fraud is detected, the manager escalates immediately.
This is dynamic coordination, not a fixed sequence.
Use Case 2: Credit Risk Assessment System
Now let’s build a credit assessment system that analyzes loan applications from multiple angles simultaneously.
System Architecture
We need four agents operating in parallel:
- Financial Analyzer: Examines applicant financials, debt ratios, cash flow
- Industry Researcher: Analyzes business sector trends and risks
- Compliance Checker: Validates against lending policies and regulations
- Manager: Synthesizes findings and makes credit decision
Building Advanced Tools
Create credit_assessment_tools.py:
from crewai_tools import tool
import json
@tool("Analyze Financial Statements")
def analyze_financials(applicant_data: dict) -> dict:
"""
Analyze applicant financial data and calculate key metrics.
Returns financial ratios, cash flow analysis, and risk indicators.
"""
income = applicant_data.get("annual_income", 0)
existing_debt = applicant_data.get("existing_debt", 0)
requested_amount = applicant_data.get("requested_amount", 0)
assets = applicant_data.get("assets", 0)
credit_score = applicant_data.get("credit_score", 0)
# Calculate key ratios
total_debt = existing_debt + requested_amount
debt_to_income = (total_debt / income * 100) if income > 0 else 999
loan_to_value = (requested_amount / assets * 100) if assets > 0 else 999
# Assess financial health
if debt_to_income < 36:
debt_assessment = "Strong - Low debt burden"
elif debt_to_income < 43:
debt_assessment = "Acceptable - Moderate debt burden"
else:
debt_assessment = "Concerning - High debt burden"
# Credit score evaluation
if credit_score >= 750:
credit_assessment = "Excellent"
elif credit_score >= 700:
credit_assessment = "Good"
elif credit_score >= 650:
credit_assessment = "Fair"
else:
credit_assessment = "Poor"
return {
"debt_to_income_ratio": round(debt_to_income, 2),
"loan_to_value_ratio": round(loan_to_value, 2),
"debt_assessment": debt_assessment,
"credit_score": credit_score,
"credit_assessment": credit_assessment,
"monthly_debt_service": round(total_debt * 0.05 / 12, 2), # Assuming 5% rate
"affordability_score": 100 - min(debt_to_income, 100),
"recommendation": "approve" if debt_to_income < 43 and credit_score >= 650 else "review"
}
@tool("Research Industry Trends")
def research_industry(industry: str, business_age_years: int) -> dict:
"""
Analyze industry trends and business sector risk.
Returns industry outlook, risk factors, and market position.
"""
# Simulated industry research
industry_data = {
"technology": {
"outlook": "positive",
"growth_rate": 8.5,
"risk_level": "medium",
"key_risks": ["rapid change", "competition", "talent acquisition"],
"stability_score": 70
},
"healthcare": {
"outlook": "positive",
"growth_rate": 6.2,
"risk_level": "low",
"key_risks": ["regulatory changes", "reimbursement rates"],
"stability_score": 85
},
"retail": {
"outlook": "mixed",
"growth_rate": 2.1,
"risk_level": "high",
"key_risks": ["e-commerce competition", "consumer spending", "margins"],
"stability_score": 55
},
"manufacturing": {
"outlook": "stable",
"growth_rate": 3.5,
"risk_level": "medium",
"key_risks": ["supply chain", "automation", "global competition"],
"stability_score": 70
},
"default": {
"outlook": "neutral",
"growth_rate": 3.0,
"risk_level": "medium",
"key_risks": ["economic conditions", "competition"],
"stability_score": 65
}
}
data = industry_data.get(industry.lower(), industry_data["default"])
# Business age factor
if business_age_years < 2:
maturity = "startup"
maturity_risk = "high"
elif business_age_years < 5:
maturity = "early_stage"
maturity_risk = "medium"
else:
maturity = "established"
maturity_risk = "low"
data["business_maturity"] = maturity
data["maturity_risk"] = maturity_risk
return data
@tool("Check Lending Compliance")
def check_compliance(applicant_data: dict, loan_terms: dict) -> dict:
"""
Validate loan application against lending policies and regulations.
Returns compliance status, policy violations, and approval requirements.
"""
violations = []
warnings = []
approvals_needed = []
# Check loan amount limits
requested = loan_terms.get("amount", 0)
if requested > 500000:
approvals_needed.append("Executive approval required for loans over $500K")
# Check credit score requirements
credit_score = applicant_data.get("credit_score", 0)
if credit_score < 620:
violations.append("Credit score below minimum threshold (620)")
elif credit_score < 680:
warnings.append("Credit score in review range - additional documentation required")
# Check debt-to-income
income = applicant_data.get("annual_income", 0)
existing_debt = applicant_data.get("existing_debt", 0)
dti = (existing_debt + requested) / income * 100 if income > 0 else 999
if dti > 50:
violations.append(f"Debt-to-income ratio ({dti:.1f}%) exceeds maximum (50%)")
elif dti > 43:
warnings.append(f"Debt-to-income ratio ({dti:.1f}%) requires compensating factors")
# Check documentation requirements
if applicant_data.get("income_verified", False) == False:
violations.append("Income verification required")
if applicant_data.get("employment_verified", False) == False:
warnings.append("Employment verification recommended")
# Determine compliance status
if len(violations) > 0:
status = "non_compliant"
can_approve = False
elif len(warnings) > 0:
status = "conditional"
can_approve = True
else:
status = "compliant"
can_approve = True
return {
"compliance_status": status,
"can_approve": can_approve,
"violations": violations,
"warnings": warnings,
"approvals_needed": approvals_needed,
"additional_documentation": warnings + violations
}
Creating the Credit Assessment Agents
Create credit_assessment_agents.py:
from crewai import Agent
from llm_config import get_openai_llm, get_anthropic_llm
from credit_assessment_tools import (
analyze_financials,
research_industry,
check_compliance
)
def create_credit_assessment_agents():
"""
Create specialized credit assessment agents.
These agents work in parallel under manager coordination.
"""
financial_analyst = Agent(
role="Senior Financial Analyst",
goal="Analyze applicant financial health and calculate credit risk metrics",
backstory="""You are a seasoned financial analyst with 15 years in commercial
lending. You excel at interpreting financial statements, calculating ratios,
and assessing borrower capacity. You identify both strengths and concerns in
financial profiles and explain them clearly.""",
tools=[analyze_financials],
llm=get_anthropic_llm(model="claude-3-5-sonnet-20241022"),
verbose=True,
allow_delegation=False
)
industry_researcher = Agent(
role="Industry Research Analyst",
goal="Evaluate business sector trends and market position",
backstory="""You are an industry research specialist who tracks market trends,
competitive dynamics, and sector-specific risks. You provide context on whether
a business operates in a growing or declining sector and what challenges they face.""",
tools=[research_industry],
llm=get_anthropic_llm(model="claude-3-5-sonnet-20241022"),
verbose=True,
allow_delegation=False
)
compliance_officer = Agent(
role="Lending Compliance Officer",
goal="Ensure loan applications meet regulatory and policy requirements",
backstory="""You are a compliance specialist who knows lending regulations,
internal policies, and documentation requirements inside and out. You identify
policy violations, flag missing documentation, and determine approval authority
levels. You protect the bank from regulatory risk.""",
tools=[check_compliance],
llm=get_anthropic_llm(model="claude-3-5-sonnet-20241022"),
verbose=True,
allow_delegation=False
)
credit_manager = Agent(
role="Credit Decision Manager",
goal="Synthesize analyses and make sound credit decisions",
backstory="""You are an experienced credit manager who makes final lending
decisions. You review financial analysis, industry research, and compliance
findings to make balanced decisions that manage risk while supporting good
business opportunities. You explain decisions clearly with supporting rationale.""",
llm=get_openai_llm(model="gpt-4"),
verbose=True,
allow_delegation=True
)
return {
"financial": financial_analyst,
"industry": industry_researcher,
"compliance": compliance_officer,
"manager": credit_manager
}
Creating Parallel Analysis Tasks
Create credit_assessment_tasks.py:
from crewai import Task
def create_credit_assessment_tasks(agents, application_data):
"""
Create tasks for parallel credit analysis.
Financial, industry, and compliance analyses run simultaneously.
"""
applicant = application_data["applicant"]
loan_request = application_data["loan_request"]
# Task 1: Financial Analysis (parallel)
financial_task = Task(
description=f"""Conduct comprehensive financial analysis of loan application.
Applicant Data:
- Annual Income: ${applicant['annual_income']:,}
- Existing Debt: ${applicant['existing_debt']:,}
- Assets: ${applicant['assets']:,}
- Credit Score: {applicant['credit_score']}
Loan Request:
- Amount: ${loan_request['amount']:,}
- Purpose: {loan_request['purpose']}
- Term: {loan_request['term_months']} months
Analyze:
1. Debt-to-income ratios
2. Loan-to-value metrics
3. Affordability assessment
4. Credit profile evaluation
5. Risk indicators""",
expected_output="""Financial analysis report containing:
- Key financial ratios calculated
- Debt capacity assessment
- Credit quality evaluation
- Financial risk rating (low/medium/high)
- Recommendation with supporting data""",
agent=agents["financial"]
)
# Task 2: Industry Research (parallel)
industry_task = Task(
description=f"""Research industry trends and business viability.
Business Information:
- Industry: {applicant['industry']}
- Business Age: {applicant['business_age_years']} years
- Business Type: {applicant['business_type']}
Analyze:
1. Industry outlook and growth trends
2. Sector-specific risks
3. Business maturity assessment
4. Competitive environment
5. Market stability""",
expected_output="""Industry analysis report containing:
- Industry outlook (positive/stable/negative)
- Growth rate and trends
- Key risk factors
- Business maturity assessment
- Sector risk rating""",
agent=agents["industry"]
)
# Task 3: Compliance Check (parallel)
compliance_task = Task(
description=f"""Verify compliance with lending policies and regulations.
Review Requirements:
- Loan amount: ${loan_request['amount']:,}
- Credit score: {applicant['credit_score']}
- Income verification: {applicant.get('income_verified', 'Not specified')}
- Employment verification: {applicant.get('employment_verified', 'Not specified')}
Check:
1. Minimum credit score requirements
2. Debt-to-income thresholds
3. Loan amount approval authorities
4. Documentation completeness
5. Regulatory compliance""",
expected_output="""Compliance review report containing:
- Compliance status (compliant/conditional/non-compliant)
- Policy violations identified
- Warnings and conditions
- Required approvals
- Missing documentation""",
agent=agents["compliance"]
)
# Task 4: Final Decision (depends on all parallel tasks)
decision_task = Task(
description="""Synthesize all analyses and make final credit decision.
Review findings from:
- Financial analysis
- Industry research
- Compliance check
Make Decision:
1. Weigh all factors and risks
2. Consider compensating strengths
3. Apply credit judgment
4. Determine: APPROVE, APPROVE_WITH_CONDITIONS, or DECLINE
5. Provide clear justification
6. Specify any conditions or monitoring requirements""",
expected_output="""Credit decision report containing:
- Final Decision: APPROVE / APPROVE_WITH_CONDITIONS / DECLINE
- Decision rationale with key factors
- Risk rating (low/medium/high)
- Conditions (if approved with conditions)
- Pricing recommendations (rate, terms)
- Monitoring requirements""",
agent=agents["manager"],
context=[financial_task, industry_task, compliance_task]
)
return [financial_task, industry_task, compliance_task, decision_task]
Running the Credit Assessment System
Create credit_assessment_main.py:
from crewai import Crew, Process
from credit_assessment_agents import create_credit_assessment_agents
from credit_assessment_tasks import create_credit_assessment_tasks
def assess_credit_application(application_data):
"""
Process credit application using hierarchical workflow with parallel analysis.
"""
print(f"\n{'='*80}")
print(f"CREDIT ASSESSMENT - Loan Application Review")
print(f"Applicant: {application_data['applicant']['name']}")
print(f"Requested Amount: ${application_data['loan_request']['amount']:,}")
print(f"{'='*80}\n")
# Create agents
agents = create_credit_assessment_agents()
# Create tasks
tasks = create_credit_assessment_tasks(agents, application_data)
# Create hierarchical crew
crew = Crew(
agents=[
agents["financial"],
agents["industry"],
agents["compliance"]
],
tasks=tasks,
process=Process.hierarchical,
manager_llm=agents["manager"].llm,
verbose=True
)
# Execute
try:
result = crew.kickoff()
print(f"\n{'='*80}")
print("CREDIT DECISION")
print(f"{'='*80}\n")
print(result)
return result
except Exception as e:
print(f"\nError in credit assessment: {str(e)}")
return None
if __name__ == "__main__":
# Sample loan application
application = {
"applicant": {
"name": "TechStart Solutions Inc",
"industry": "technology",
"business_type": "LLC",
"business_age_years": 3,
"annual_income": 850000,
"existing_debt": 120000,
"assets": 450000,
"credit_score": 710,
"income_verified": True,
"employment_verified": True
},
"loan_request": {
"amount": 250000,
"purpose": "equipment_purchase",
"term_months": 60
}
}
result = assess_credit_application(application)
The key difference here is that financial analysis, industry research, and compliance checking happen in parallel. The manager coordinates all three, then synthesizes results for the final decision. This cuts total execution time significantly compared to sequential processing.
Cost Optimization Strategies
Running hierarchical workflows with multiple agents increases LLM costs. Here are proven strategies to optimize:
1. Model Selection by Task Complexity
Use cheaper models for simple tasks, reserve expensive models for complex reasoning:
- Data retrieval/classification: Llama 3.1 8B (local, free)
- Standard analysis: Claude Haiku ($0.25 per million tokens)
- Complex reasoning: Claude Sonnet ($3 per million tokens)
- Coordination: GPT-4 ($30 per million tokens) or Claude Opus
2. Parallel vs Sequential Execution
Parallel execution costs more in tokens but saves wall-clock time. Calculate tradeoffs:
- Sequential: Lower token cost, longer total time
- Parallel: Higher token cost, faster total time
- Hybrid: Critical path parallel, rest sequential
3. Context Management
Minimize token usage by passing only essential context between agents:
# Instead of passing entire previous outputs
context=[task1, task2, task3]
# Pass only what's needed
context=[task1] # Next agent only needs task1 results
4. Caching and Memoization
Cache results for repeated analyses:
from functools import lru_cache
@lru_cache(maxsize=1000)
def get_industry_data(industry: str):
# Expensive research operation
# Results cached for repeated requests
pass
5. Batch Processing
For high-volume operations, batch similar requests:
# Process 100 credit applications in one crew execution
# Rather than 100 separate executions
Real cost example from production: Processing 1,000 credit applications per day:
- All GPT-4: ~$150/day
- Optimized mix: ~$35/day (77% savings)
- Mostly local models: ~$8/day (95% savings, some quality tradeoff)
Error Handling and Fallback Strategies
Hierarchical workflows need robust error handling because failures can occur at multiple levels.
Agent-Level Error Handling
from crewai import Agent
def create_resilient_agent(role, goal, backstory, tools, primary_llm, fallback_llm):
"""Create agent with automatic LLM fallback"""
try:
agent = Agent(
role=role,
goal=goal,
backstory=backstory,
tools=tools,
llm=primary_llm,
verbose=True
)
return agent
except Exception as e:
print(f"Primary LLM failed, using fallback: {e}")
return Agent(
role=role,
goal=goal,
backstory=backstory,
tools=tools,
llm=fallback_llm,
verbose=True
)
Task-Level Retry Logic
def execute_task_with_retry(crew, max_retries=3):
"""Execute crew with exponential backoff retry"""
import time
for attempt in range(max_retries):
try:
result = crew.kickoff()
return result
except Exception as e:
wait_time = 2 ** attempt # Exponential backoff
print(f"Attempt {attempt + 1} failed: {e}")
if attempt < max_retries - 1:
print(f"Retrying in {wait_time} seconds...")
time.sleep(wait_time)
else:
print("Max retries exceeded")
raise
Graceful Degradation
def assess_credit_with_fallback(application_data):
"""Try full analysis, fall back to simplified version if needed"""
try:
# Attempt full hierarchical analysis
return assess_credit_application(application_data)
except Exception as e:
print(f"Full analysis failed: {e}")
print("Falling back to simplified assessment...")
# Simplified fallback: single agent, basic analysis
from credit_assessment_tools import analyze_financials
financial_result = analyze_financials(application_data['applicant'])
return f"Simplified assessment: {financial_result['recommendation']}"
Key Takeaways
We built two sophisticated multi-agent systems today:
- Customer Service Automation: Five agents with hierarchical coordination, mixed LLM strategy for cost optimization, dynamic routing based on query classification, and manager oversight for quality and escalation.
- Credit Risk Assessment: Four agents performing parallel analysis, simultaneous financial, industry, and compliance review, synthesis of multiple perspectives into final decision, and structured decision-making with clear justification.
These patterns apply to countless banking workflows. The key principles are:
- Use hierarchical workflows when tasks can be parallelized or require dynamic coordination
- Optimize costs by matching model capability to task complexity
- Implement robust error handling and fallback strategies
- Design manager agents specifically for coordination, not execution
- Pass minimal necessary context between agents to control token costs
In Part 4, we will build a production-ready transaction reconciliation system with complete deployment architecture, monitoring, audit trails, and integration with existing banking infrastructure.
What You Should Build Next
Take these implementations and adapt them to your specific banking use cases. Try building:
- Loan underwriting workflow with document analysis agents
- Trade surveillance system with pattern detection agents
- Compliance monitoring with regulatory checking agents
- Customer onboarding with verification and validation agents
The architecture patterns you learned today scale to these scenarios. Start with 3–4 agents, add hierarchical coordination, optimize costs, and iterate based on results.
If you found this valuable, give it a clap and share your implementation experiences in the comments. What workflows are you building? Which cost optimization strategies worked best? What challenges did you encounter with hierarchical coordination?
Follow me for Part 4 where we deploy these systems to production with Docker containers, monitoring dashboards, and enterprise integration patterns.
About This Series: This is Part 3 of a 4-part series on building multi-agent AI systems for banking with CrewAI.
Part 1: Foundation and concepts of agentic AI Part 2: Basic implementation with fraud detection Part 3 (this article): Advanced workflows with hierarchical coordination Part 4 (coming soon): Production-ready transaction reconciliation system
This is part of the series Building Production-Ready Multi-Agent AI Systems for Banking: A Complete Guide, where we walk through practical architectures, orchestration patterns, memory systems, tool integrations, and production deployment strategies for building enterprise-grade banking AI systems using CrewAI and modern agentic AI frameworks.
Follow me to get notified when Part 4 is published. Share this with teams exploring AI automation in banking or financial services.
Building Multi-Agent AI Systems for Banking: Advanced Workflows and Agent Coordination with CrewAI… was originally published in Towards AI on Medium, where people are continuing the conversation by highlighting and responding to this story.