Multi-Model Risk Assessment Agent
Build a LangGraph agent that routes loan queries to ML models with SHAP explainability.
Introduction
30 Second Summary
When a bank evaluates your loan application, one AI model alone rarely makes the final call. Production systems route your data through multiple specialized models, each handling a different piece of the puzzle.
In this project, you will build a multi-model risk assessment agent that takes natural-language questions about loan applicants and intelligently routes them to the right combination of ML models for answers.
What You'll Build
A LangGraph-powered agent that uses a local LLM to classify intent, then dynamically orchestrates three distinct models through routing, ensemble, and pipeline patterns.
By the end of this project, you'll have:
- A LangGraph StateGraph with conditional routing logic that directs queries to the correct model based on classified intent.
- Three trained models (Logistic Regression, Random Forest, and a deterministic rule engine) orchestrated through routing, ensemble, and pipeline modes.
- SHAP explainability attached to every prediction, with a local Ollama LLM synthesizing human-readable explanations.
- Secret Mission: A confidence-calibrated disagreement detector that flags cases where models strongly disagree and triggers a human-in-the-loop escalation path.
Are there any prerequisites?
You should be familiar with Python and pandas. You also need Ollama installed locally on your machine.
Set Up the Project and Install Dependencies
Your packages are installed and Qwen3 is ready to go.
Now it's time to create the project file where all the pieces come together. You'll lay the foundation with imports, a typed state schema that defines how data flows through the agent, and a connection to your local LLM.
In this step, get ready to:
- Create main.py with all required imports.
- Define the AgentState schema to track data through the graph.
- Connect to the local Qwen3 model via ChatOllama.
Create main.py and add the imports
The agent relies on several libraries working together. LangGraph handles orchestration, scikit-learn provides the ML models, SHAP powers explainability, and langchain-ollama connects to your local LLM.
- In Cursor, create a new file called main.py in your project folder.
- Add the following imports at the top of the file:
import json
from typing import TypedDict, Literal
import numpy as np
import pandas as pd
from sklearn.linear_model import LogisticRegression
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
import shap
from langgraph.graph import StateGraph, START, END
from langchain_ollama import ChatOllama
from langchain.messages import SystemMessage, HumanMessageWhat do these imports do?
- json and typing are Python standard library tools for serializing data and adding type annotations.
- numpy and pandas handle numerical operations and structured dataframes for the loan dataset.
- sklearn provides two ML classifiers (Logistic Regression and Random Forest) plus a utility for splitting data into training and test sets.
- shap generates feature-level explanations for tree-based model predictions.
- langgraph.graph gives you StateGraph, START, and END to build the agent's execution graph.
- ChatOllama from langchain_ollama connects to your local Qwen3 model for intent classification and response synthesis.
- Save main.py.
You won't see any output yet. The next chunk defines the state schema that these imports will power.
Seeing an import error?
- Make sure all packages are installed. Re-run pip install langgraph langchain-ollama scikit-learn shap pandas if any import fails.
- Confirm you're using the same Python environment where the packages were installed.
Help me debug my import errors in main.py
Define the agent state and connect the LLM
LangGraph uses a typed state dictionary to pass data between nodes in the graph. Each field represents a piece of information that flows through the system as the agent processes a query.
- Below the imports in main.py, add the state definition and LLM connection:
# --- State Definition ---
class AgentState(TypedDict):
query: str
intent: str
applicant_data: dict
predictions: dict
shap_explanation: dict
final_response: str
# --- LLM Setup ---
llm = ChatOllama(model="qwen3", temperature=0)What does this code do?
AgentState defines the shape of data that every node in your graph can read and write.
- query stores the natural-language question from the user.
- intent holds the classified intent (like "compliance" or "risk_score").
- applicant_data contains the loan applicant's financial details as a dictionary.
- predictions collects outputs from whichever ML models run.
- shap_explanation stores feature contribution data from SHAP.
- final_response holds the LLM's human-readable summary at the end of the pipeline.
ChatOllama(model="qwen3", temperature=0) creates a connection to your local Qwen3 model with temperature set to 0 for deterministic, reproducible responses.
- Save main.py.
- Verify the file runs cleanly by running this in your terminal:
python main.pyThe script should exit without any output or errors. A clean exit means all imports resolved and the LLM connection was created successfully.
Getting a connection error from ChatOllama?
- Make sure Ollama is running. Open a separate terminal and run ollama list to check that Qwen3 appears in the model list.
- If Ollama isn't running, start it with ollama serve in a separate terminal window.
- Confirm the model name in your code matches exactly. It should be qwen3, not qwen-3 or qwen3:latest.
Help me fix my ChatOllama connection error
✔️ Awesome, I've got everything!
Great. Double check that you've saved main.py and that it runs cleanly with no errors.
ⓧ I'd like to double check the full code
Here's the complete main.py so far:
import json
from typing import TypedDict, Literal
import numpy as np
import pandas as pd
from sklearn.linear_model import LogisticRegression
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
import shap
from langgraph.graph import StateGraph, START, END
from langchain_ollama import ChatOllama
from langchain.messages import SystemMessage, HumanMessage
# --- State Definition ---
class AgentState(TypedDict):
query: str
intent: str
applicant_data: dict
predictions: dict
shap_explanation: dict
final_response: str
# --- LLM Setup ---
llm = ChatOllama(model="qwen3", temperature=0)Your project foundation is set. You have imports, a typed state schema, and a live LLM connection. Next up, you'll generate a synthetic loan dataset and train the ML models that your agent will orchestrate.
Train the ML Models
Your project structure is ready and all dependencies are installed. Now you need the models that will power the agent's predictions.
A risk assessment system needs multiple perspectives on a loan applicant. In this step, you will build three different models that each evaluate risk in a different way. The LLM you set up in Step 1 will never make predictions itself. Instead, it will orchestrate these deterministic models that produce consistent, auditable results.
In this step, get ready to:
- Generate a synthetic loan dataset with pandas.
- Train a Logistic Regression and Random Forest classifier with scikit-learn.
- Build a deterministic rule engine that checks regulatory thresholds.
Load a synthetic loan dataset and prepare features
Before training any model, you need data. You will generate a synthetic dataset of 1,000 loan applicants with features like income, debt-to-income ratio, credit score, employment history, and loan amount.
- In main.py, find the # --- LLM Setup --- comment.
- Add the following function above that line to generate synthetic loan data:
# --- Synthetic Dataset ---
def generate_loan_data(n=1000):
np.random.seed(42)
data = pd.DataFrame({
"income": np.random.normal(55000, 15000, n).clip(20000, 150000),
"debt_to_income": np.random.uniform(0.1, 0.7, n),
"credit_score": np.random.normal(680, 50, n).clip(300, 850).astype(int),
"employment_years": np.random.exponential(5, n).clip(0, 30),
"loan_amount": np.random.normal(25000, 10000, n).clip(5000, 100000),
})
default_prob = (
0.3 * (data["debt_to_income"] > 0.43).astype(float)
+ 0.25 * (data["credit_score"] < 620).astype(float)
+ 0.2 * (data["income"] < 35000).astype(float)
+ 0.15 * (data["employment_years"] < 1).astype(float)
+ 0.1 * (data["loan_amount"] > 50000).astype(float)
)
data["default"] = (default_prob + np.random.normal(0, 0.15, n) > 0.45).astype(int)
return dataWhat does this code do?
This function creates a realistic-looking loan dataset without needing any external data files. Each applicant gets five numeric features drawn from statistical distributions that mimic real-world lending data.
The default_prob formula is the key part. It assigns weighted risk factors to each feature. For example, a debt-to-income ratio above 0.43 contributes 30% toward a default signal. The final default column is the binary label (1 = defaulted, 0 = did not default) that the models will learn to predict.
You will not see a change yet. The next chunk calls this function and trains the models on the generated data.
Train a Logistic Regression and Random Forest classifier
With the data generator ready, you can now train two ML models. Each model learns different patterns from the same data, which is what makes ensemble comparisons valuable later.
- Directly below the generate_loan_data function you just added, paste the following training code (still above # --- LLM Setup ---):
# --- Train Models ---
data = generate_loan_data()
features = ["income", "debt_to_income", "credit_score", "employment_years", "loan_amount"]
X = data[features]
y = data["default"]
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)
logistic_model = LogisticRegression(max_iter=1000, random_state=42)
logistic_model.fit(X_train, y_train)
rf_model = RandomForestClassifier(n_estimators=100, max_depth=5, random_state=42)
rf_model.fit(X_train, y_train)
# SHAP explainer for Random Forest
shap_explainer = shap.TreeExplainer(rf_model)What does this code do?
This block generates the dataset, splits it into training (80%) and test (20%) sets, then trains two classifiers.
- LogisticRegression produces calibrated probabilities. It models a linear relationship between the features and the log-odds of default. max_iter=1000 gives the solver enough iterations to converge (the default of 100 can be too few for some datasets).
- RandomForestClassifier captures non-linear interactions between features. It builds 100 decision trees, each limited to depth 5, then averages their predictions.
- shap.TreeExplainer(rf_model) pre-computes the internal tree structure needed to calculate fast, exact SHAP values for the Random Forest. You will use this explainer later to generate feature-level explanations.
Why two different models?
Logistic Regression is chosen for its calibrated probabilities. When it says 70% risk, that closely matches real-world default rates. This matters for ensemble agreement scoring.
Random Forest is chosen for its ability to capture non-linear feature interactions (like high debt combined with short employment). It is also directly compatible with SHAP's fast TreeExplainer, which gives exact explanations without sampling.
- Save main.py.
- Run the file to confirm the models train without errors:
python main.pyThe script should complete in a few seconds with no output. If you see no errors, both models trained successfully.
Seeing a convergence warning?
If you see a warning about LogisticRegression not converging, double check that you set max_iter=1000 (not the default 100). This gives the solver enough iterations for your dataset.
If you see an import error, make sure you installed all packages in Step 1.
help me debug a convergence warning when training LogisticRegression in my LangGraph risk assessment project
Build a deterministic rule engine
ML models are great at finding patterns, but some decisions need hard rules. Regulatory compliance checks must be deterministic. A rule engine guarantees the same input always produces the same pass/fail result.
- Below the shap_explainer line you just added, paste the rule engine function (still above # --- LLM Setup ---):
# --- Rule Engine ---
def rule_engine(applicant: dict) -> dict:
flags = []
passed = True
if applicant["debt_to_income"] > 0.43:
flags.append("FAIL: Debt-to-income ratio exceeds 0.43 regulatory threshold")
passed = False
if applicant["credit_score"] < 580:
flags.append("FAIL: Credit score below 580 minimum")
passed = False
if applicant["employment_years"] < 0.5:
flags.append("WARNING: Less than 6 months employment history")
if passed and not flags:
flags.append("PASS: All regulatory checks passed")
return {"passed": passed, "flags": flags}What does this code do?
The rule engine performs three hard-coded regulatory checks on an applicant dictionary:
- Debt-to-income above 0.43 is an automatic fail. This threshold comes from the Qualified Mortgage rule.
- Credit score below 580 is an automatic fail.
- Employment under 6 months triggers a warning but does not fail the applicant.
The function returns a dictionary with a passed boolean and a flags list. The agent will route compliance questions to this engine instead of asking an ML model to guess at regulatory rules.
- Save main.py.
- Test the rule engine by running this command:
python -c "from main import rule_engine; print(rule_engine({'debt_to_income': 0.5, 'credit_score': 600, 'employment_years': 2, 'income': 50000, 'loan_amount': 20000}))"You should see output showing 'passed': False with a flag about the debt-to-income ratio exceeding 0.43.
Getting a ModuleNotFoundError?
Make sure you are running the command from the same directory where main.py lives. Use ls to confirm main.py is in your current folder.
If you see a connection error related to Ollama, make sure Ollama is still running in the background (it was started in Step 1). The ChatOllama line at the bottom of the file instantiates the LLM client on import.
help me fix an import error when testing my rule engine function
✔️ Awesome, I've got everything!
Great. Double check you have saved main.py with all three sections: the generate_loan_data function, the training block, and the rule_engine function.
ⓧ I'd like to double check the full code
Here is the complete main.py at this point:
import json
from typing import TypedDict, Literal
import numpy as np
import pandas as pd
from sklearn.linear_model import LogisticRegression
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
import shap
from langgraph.graph import StateGraph, START, END
from langchain_ollama import ChatOllama
from langchain.messages import SystemMessage, HumanMessage
# --- State Definition ---
class AgentState(TypedDict):
query: str
intent: str
applicant_data: dict
predictions: dict
shap_explanation: dict
final_response: str
# --- Synthetic Dataset ---
def generate_loan_data(n=1000):
np.random.seed(42)
data = pd.DataFrame({
"income": np.random.normal(55000, 15000, n).clip(20000, 150000),
"debt_to_income": np.random.uniform(0.1, 0.7, n),
"credit_score": np.random.normal(680, 50, n).clip(300, 850).astype(int),
"employment_years": np.random.exponential(5, n).clip(0, 30),
"loan_amount": np.random.normal(25000, 10000, n).clip(5000, 100000),
})
default_prob = (
0.3 * (data["debt_to_income"] > 0.43).astype(float)
+ 0.25 * (data["credit_score"] < 620).astype(float)
+ 0.2 * (data["income"] < 35000).astype(float)
+ 0.15 * (data["employment_years"] < 1).astype(float)
+ 0.1 * (data["loan_amount"] > 50000).astype(float)
)
data["default"] = (default_prob + np.random.normal(0, 0.15, n) > 0.45).astype(int)
return data
# --- Train Models ---
data = generate_loan_data()
features = ["income", "debt_to_income", "credit_score", "employment_years", "loan_amount"]
X = data[features]
y = data["default"]
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)
logistic_model = LogisticRegression(max_iter=1000, random_state=42)
logistic_model.fit(X_train, y_train)
rf_model = RandomForestClassifier(n_estimators=100, max_depth=5, random_state=42)
rf_model.fit(X_train, y_train)
# SHAP explainer for Random Forest
shap_explainer = shap.TreeExplainer(rf_model)
# --- Rule Engine ---
def rule_engine(applicant: dict) -> dict:
flags = []
passed = True
if applicant["debt_to_income"] > 0.43:
flags.append("FAIL: Debt-to-income ratio exceeds 0.43 regulatory threshold")
passed = False
if applicant["credit_score"] < 580:
flags.append("FAIL: Credit score below 580 minimum")
passed = False
if applicant["employment_years"] < 0.5:
flags.append("WARNING: Less than 6 months employment history")
if passed and not flags:
flags.append("PASS: All regulatory checks passed")
return {"passed": passed, "flags": flags}
# --- LLM Setup ---
llm = ChatOllama(model="qwen3", temperature=0)You now have three distinct models ready to go: a Logistic Regression for calibrated probabilities, a Random Forest for non-linear patterns with SHAP explainability, and a rule engine for deterministic compliance checks. Next up, you will build the LangGraph agent that orchestrates all three.
Build the LangGraph Agent
Your models are trained and ready to make predictions. Now you need the brain that decides which model to consult for any given question.
LangGraph's StateGraph lets you define nodes (processing functions) and edges (transitions between them). This gives you explicit control over how queries flow through the system. In this step, you'll build the intent classifier that reads a natural-language question and the routing logic that sends it to the right place.
In this step, get ready to:
- Create an intent classifier node that uses the LLM to categorize queries.
- Wire up conditional edges that route queries to the correct orchestration mode.
Create the intent classifier node
The intent classifier is the first node in your graph. It takes a user's natural-language query and categorizes it into one of four intents: compliance, risk_score, full_assessment, or screen. Each intent maps to a different orchestration mode downstream.
- In main.py, scroll to the bottom of the file (below the llm = ChatOllama(...) line) and add the intent classifier function:
# --- Node: Intent Classifier ---
def classify_intent(state: AgentState) -> dict:
system_prompt = """You are a query classifier for a loan risk assessment system.
Classify the user query into exactly one of these intents:
- "compliance" - questions about regulatory compliance or rule checks
- "risk_score" - questions about credit risk or default probability
- "full_assessment" - requests for a complete evaluation using all models
- "screen" - requests to screen/filter an applicant through the full pipeline
Respond with ONLY the intent word, nothing else. Do not use any thinking tags."""
response = llm.invoke([
SystemMessage(content=system_prompt),
HumanMessage(content=state["query"])
])
intent = response.content.strip().strip('"').lower()
valid_intents = ["compliance", "risk_score", "full_assessment", "screen"]
if intent not in valid_intents:
intent = "full_assessment"
return {"intent": intent}What does this code do?
This function is a LangGraph node. It receives the current graph state and returns a dictionary with the fields to update.
- The system_prompt tells Qwen3 to act as a strict classifier. The instruction "Do not use any thinking tags" prevents the model from emitting internal reasoning tokens in its response.
- llm.invoke() sends the system prompt and the user's query to Qwen3 via Ollama.
- The fallback logic (if intent not in valid_intents) defaults to full_assessment so the system never gets stuck if the LLM produces an unexpected response.
- Save main.py.
Why a thin classifier instead of full tool calling?
The LLM's only job here is to understand the user's intent. It never touches predictions or data. This separation keeps the LLM layer lightweight and auditable. If the classifier ever misroutes, you can inspect its single-word output without digging through tool-call traces.
Wire up conditional edges that route to the correct orchestration mode
Conditional edges are what make your graph dynamic. Instead of a fixed sequence of steps, the graph inspects the current state and picks a different path at runtime.
- Below the classify_intent function, add the route decision function:
# --- Node: Route Decision ---
def route_decision(state: AgentState) -> str:
intent = state["intent"]
if intent == "compliance":
return "routing_node"
elif intent == "risk_score":
return "routing_node"
elif intent == "full_assessment":
return "ensemble_node"
elif intent == "screen":
return "pipeline_node"
return "ensemble_node"What does this code do?
- route_decision reads the classified intent from state and returns a string naming the next node to execute.
- Both compliance and risk_score route to routing_node because they each need only a single model. The routing node itself will pick which one.
- full_assessment sends the query to ensemble_node (all models), and screen sends it to pipeline_node (sequential chain).
- Save main.py.
Now you have the classifier and the router. Time to wire them into an actual graph.
- Below the route_decision function, add the graph construction code:
# --- Build the Graph ---
workflow = StateGraph(AgentState)
workflow.add_node("classify_intent", classify_intent)
workflow.add_edge(START, "classify_intent")
workflow.add_conditional_edges("classify_intent", route_decision, {
"routing_node": "routing_node",
"ensemble_node": "ensemble_node",
"pipeline_node": "pipeline_node",
})What's happening here?
- StateGraph(AgentState) creates a new graph that uses your AgentState TypedDict as its state schema. Every node reads from and writes to this shared state.
- add_node registers a function as a named node in the graph.
- add_edge(START, "classify_intent") makes classify_intent the first node that runs when the graph starts.
- add_conditional_edges connects classify_intent to the routing function. The dictionary maps each possible return value to the target node name.
- Save main.py.
You won't see a change yet. The target nodes (routing_node, ensemble_node, pipeline_node) don't exist yet. You'll build them in the next step.
Seeing a syntax error?
- Check that the triple-quoted system prompt string uses matching triple double-quotes at both ends (""").
- Make sure the add_conditional_edges dictionary uses colons, not equals signs, for key-value pairs.
- Verify that your classify_intent and route_decision functions are defined BEFORE the graph construction code that references them.
Help me debug a syntax error in my LangGraph agent code.
✔️ Awesome, I've got everything!
Great. Double check you've saved your file before moving on.
ⓧ I'd like to double check the full code
Here is the full main.py as it should look at the end of this step:
import json
from typing import TypedDict, Literal
import numpy as np
import pandas as pd
from sklearn.linear_model import LogisticRegression
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
import shap
from langgraph.graph import StateGraph, START, END
from langchain_ollama import ChatOllama
from langchain.messages import SystemMessage, HumanMessage
# --- State Definition ---
class AgentState(TypedDict):
query: str
intent: str
applicant_data: dict
predictions: dict
shap_explanation: dict
final_response: str
# --- Synthetic Dataset ---
def generate_loan_data(n=1000):
np.random.seed(42)
data = pd.DataFrame({
"income": np.random.normal(55000, 15000, n).clip(20000, 150000),
"debt_to_income": np.random.uniform(0.1, 0.7, n),
"credit_score": np.random.normal(680, 50, n).clip(300, 850).astype(int),
"employment_years": np.random.exponential(5, n).clip(0, 30),
"loan_amount": np.random.normal(25000, 10000, n).clip(5000, 100000),
})
default_prob = (
0.3 * (data["debt_to_income"] > 0.43).astype(float)
+ 0.25 * (data["credit_score"] < 620).astype(float)
+ 0.2 * (data["income"] < 35000).astype(float)
+ 0.15 * (data["employment_years"] < 1).astype(float)
+ 0.1 * (data["loan_amount"] > 50000).astype(float)
)
data["default"] = (default_prob + np.random.normal(0, 0.15, n) > 0.45).astype(int)
return data
# --- Train Models ---
data = generate_loan_data()
features = ["income", "debt_to_income", "credit_score", "employment_years", "loan_amount"]
X = data[features]
y = data["default"]
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)
logistic_model = LogisticRegression(max_iter=1000, random_state=42)
logistic_model.fit(X_train, y_train)
rf_model = RandomForestClassifier(n_estimators=100, max_depth=5, random_state=42)
rf_model.fit(X_train, y_train)
# SHAP explainer for Random Forest
shap_explainer = shap.TreeExplainer(rf_model)
# --- Rule Engine ---
def rule_engine(applicant: dict) -> dict:
flags = []
passed = True
if applicant["debt_to_income"] > 0.43:
flags.append("FAIL: Debt-to-income ratio exceeds 0.43 regulatory threshold")
passed = False
if applicant["credit_score"] < 580:
flags.append("FAIL: Credit score below 580 minimum")
passed = False
if applicant["employment_years"] < 0.5:
flags.append("WARNING: Less than 6 months employment history")
if passed and not flags:
flags.append("PASS: All regulatory checks passed")
return {"passed": passed, "flags": flags}
# --- LLM Setup ---
llm = ChatOllama(model="qwen3", temperature=0)
# --- Node: Intent Classifier ---
def classify_intent(state: AgentState) -> dict:
system_prompt = """You are a query classifier for a loan risk assessment system.
Classify the user query into exactly one of these intents:
- "compliance" - questions about regulatory compliance or rule checks
- "risk_score" - questions about credit risk or default probability
- "full_assessment" - requests for a complete evaluation using all models
- "screen" - requests to screen/filter an applicant through the full pipeline
Respond with ONLY the intent word, nothing else. Do not use any thinking tags."""
response = llm.invoke([
SystemMessage(content=system_prompt),
HumanMessage(content=state["query"])
])
intent = response.content.strip().strip('"').lower()
valid_intents = ["compliance", "risk_score", "full_assessment", "screen"]
if intent not in valid_intents:
intent = "full_assessment"
return {"intent": intent}
# --- Node: Route Decision ---
def route_decision(state: AgentState) -> str:
intent = state["intent"]
if intent == "compliance":
return "routing_node"
elif intent == "risk_score":
return "routing_node"
elif intent == "full_assessment":
return "ensemble_node"
elif intent == "screen":
return "pipeline_node"
return "ensemble_node"
# --- Build the Graph ---
workflow = StateGraph(AgentState)
workflow.add_node("classify_intent", classify_intent)
workflow.add_edge(START, "classify_intent")
workflow.add_conditional_edges("classify_intent", route_decision, {
"routing_node": "routing_node",
"ensemble_node": "ensemble_node",
"pipeline_node": "pipeline_node",
})Your agent can now understand a question and pick a path. Next up, you'll build the three destination nodes that actually run the models.
Implement the Three Orchestration Modes
Your agent can now classify intent and decide which path to take. The conditional edges are wired up, but the destination nodes don't exist yet.
In this step, you'll build the three execution nodes that do the actual work: routing to a single model, consulting all models as an ensemble, and chaining models in a sequential pipeline.
In this step, get ready to:
- Build a routing node that dispatches to the single most relevant model.
- Build an ensemble node that queries all three models and compares outputs.
- Build a pipeline node that chains models sequentially with early termination.
Build the routing node
Routing is the simplest orchestration pattern. The agent looks at the classified intent and sends the query to exactly one model.
- In main.py, find the route_decision function you wrote in the previous step.
- Add the following routing_node function below it:
# --- Node: Routing (single model) ---
def routing_node(state: AgentState) -> dict:
applicant = state["applicant_data"]
intent = state["intent"]
predictions = {}
if intent == "compliance":
predictions["rule_engine"] = rule_engine(applicant)
elif intent == "risk_score":
X_input = pd.DataFrame([applicant])[features]
rf_prob = rf_model.predict_proba(X_input)[0][1]
predictions["random_forest"] = {"default_probability": round(float(rf_prob), 4)}
return {"predictions": predictions}What does this code do?
This node reads the intent from state and calls only the model that matches. A "compliance" query goes to the rule engine. A "risk_score" query goes to the Random Forest.
The node returns a predictions dictionary that gets merged into the agent state automatically by LangGraph.
- Save main.py.
You won't see a change yet. The next two nodes complete the orchestration layer, and then you'll register all three with the graph.
Why use the Random Forest for risk scoring?
The Random Forest captures non-linear feature interactions (like income combined with debt ratio) better than Logistic Regression. It also pairs with SHAP TreeExplainer for fast, exact explainability in the next step.
Build the ensemble node
The ensemble pattern consults every model on the same applicant and produces a transparency report. This shows where models agree and where they diverge.
- In main.py, add the following ensemble_node function below routing_node:
# --- Node: Ensemble (all models) ---
def ensemble_node(state: AgentState) -> dict:
applicant = state["applicant_data"]
X_input = pd.DataFrame([applicant])[features]
lr_prob = logistic_model.predict_proba(X_input)[0][1]
rf_prob = rf_model.predict_proba(X_input)[0][1]
rule_result = rule_engine(applicant)
predictions = {
"logistic_regression": {"default_probability": round(float(lr_prob), 4)},
"random_forest": {"default_probability": round(float(rf_prob), 4)},
"rule_engine": rule_result,
"agreement": abs(lr_prob - rf_prob) < 0.15,
"average_probability": round(float((lr_prob + rf_prob) / 2), 4),
}
return {"predictions": predictions}What does this code do?
This node runs all three models against the same applicant data. It collects each model's default probability plus the rule engine's pass/fail result.
- agreement is True when the two ML models predict probabilities within 0.15 of each other.
- average_probability gives a simple combined score for quick decision-making.
- Save main.py.
You won't see a change yet. The pipeline node below completes the set.
Why track model agreement?
When Logistic Regression and Random Forest disagree significantly, it signals the applicant sits in a region where the models have learned different boundaries. A loan officer can use this flag to know when extra scrutiny is needed.
Build the pipeline node and register all orchestration nodes
The pipeline pattern chains models sequentially. Each stage acts as a gate. If the applicant fails an early stage, the pipeline short-circuits and skips the rest.
- In main.py, add the following pipeline_node function below ensemble_node:
# --- Node: Pipeline (sequential) ---
def pipeline_node(state: AgentState) -> dict:
applicant = state["applicant_data"]
X_input = pd.DataFrame([applicant])[features]
# Stage 1: Rule engine filter
rule_result = rule_engine(applicant)
if not rule_result["passed"]:
return {"predictions": {
"pipeline_stage": "rejected_at_rules",
"rule_engine": rule_result,
}}
# Stage 2: Logistic Regression scoring
lr_prob = logistic_model.predict_proba(X_input)[0][1]
# Stage 3: Random Forest refinement
rf_prob = rf_model.predict_proba(X_input)[0][1]
predictions = {
"pipeline_stage": "completed",
"rule_engine": rule_result,
"logistic_regression": {"default_probability": round(float(lr_prob), 4)},
"random_forest": {"default_probability": round(float(rf_prob), 4)},
"final_score": round(float(rf_prob), 4),
}
return {"predictions": predictions}What does this code do?
The pipeline runs three stages in order, with an early exit if the applicant fails regulatory checks.
- Stage 1 runs the rule engine. If the applicant fails (e.g. debt-to-income exceeds 0.43), the function returns immediately with pipeline_stage: "rejected_at_rules".
- Stage 2 scores with Logistic Regression (calibrated probabilities).
- Stage 3 refines with Random Forest and sets the final_score.
Now register all three orchestration nodes with the graph. The conditional edges from the previous step already reference these node names.
- Scroll down to the # --- Build the Graph --- section at the bottom of main.py.
- Find this line:
workflow.add_node("classify_intent", classify_intent)- Add these three lines directly below it:
workflow.add_node("routing_node", routing_node)
workflow.add_node("ensemble_node", ensemble_node)
workflow.add_node("pipeline_node", pipeline_node)What does this do?
Each add_node call registers a Python function as a named node in the StateGraph. The string name must match the targets declared in your add_conditional_edges call from the previous step.
- Save main.py.
- Verify there are no syntax errors by running this in your terminal:
python main.pyThe script should finish without errors. You won't see output because the graph isn't invoked yet.
Seeing a NameError or KeyError?
- If you see NameError: name 'routing_node' is not defined, make sure the function definitions appear above the # --- Build the Graph --- section. Python reads top to bottom.
- If you see an IndentationError, check that the if/elif blocks inside each function use consistent 4-space indentation.
- Check that the add_node string names match your function names exactly (e.g. "routing_node" matches routing_node).
Still stuck? Ask for help: Why is my LangGraph node registration failing with a NameError?
✔️ Awesome, I've got everything!
Great. Double check you've saved your file before moving on.
ⓧ I'd like to double check the full code
Here is the complete main.py at this point:
import json
from typing import TypedDict, Literal
import numpy as np
import pandas as pd
from sklearn.linear_model import LogisticRegression
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
import shap
from langgraph.graph import StateGraph, START, END
from langchain_ollama import ChatOllama
from langchain.messages import SystemMessage, HumanMessage
# --- State Definition ---
class AgentState(TypedDict):
query: str
intent: str
applicant_data: dict
predictions: dict
shap_explanation: dict
final_response: str
# --- Synthetic Dataset ---
def generate_loan_data(n=1000):
np.random.seed(42)
data = pd.DataFrame({
"income": np.random.normal(55000, 15000, n).clip(20000, 150000),
"debt_to_income": np.random.uniform(0.1, 0.7, n),
"credit_score": np.random.normal(680, 50, n).clip(300, 850).astype(int),
"employment_years": np.random.exponential(5, n).clip(0, 30),
"loan_amount": np.random.normal(25000, 10000, n).clip(5000, 100000),
})
default_prob = (
0.3 * (data["debt_to_income"] > 0.43).astype(float)
+ 0.25 * (data["credit_score"] < 620).astype(float)
+ 0.2 * (data["income"] < 35000).astype(float)
+ 0.15 * (data["employment_years"] < 1).astype(float)
+ 0.1 * (data["loan_amount"] > 50000).astype(float)
)
data["default"] = (default_prob + np.random.normal(0, 0.15, n) > 0.45).astype(int)
return data
# --- Train Models ---
data = generate_loan_data()
features = ["income", "debt_to_income", "credit_score", "employment_years", "loan_amount"]
X = data[features]
y = data["default"]
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)
logistic_model = LogisticRegression(max_iter=1000, random_state=42)
logistic_model.fit(X_train, y_train)
rf_model = RandomForestClassifier(n_estimators=100, max_depth=5, random_state=42)
rf_model.fit(X_train, y_train)
# SHAP explainer for Random Forest
shap_explainer = shap.TreeExplainer(rf_model)
# --- Rule Engine ---
def rule_engine(applicant: dict) -> dict:
flags = []
passed = True
if applicant["debt_to_income"] > 0.43:
flags.append("FAIL: Debt-to-income ratio exceeds 0.43 regulatory threshold")
passed = False
if applicant["credit_score"] < 580:
flags.append("FAIL: Credit score below 580 minimum")
passed = False
if applicant["employment_years"] < 0.5:
flags.append("WARNING: Less than 6 months employment history")
if passed and not flags:
flags.append("PASS: All regulatory checks passed")
return {"passed": passed, "flags": flags}
# --- LLM Setup ---
llm = ChatOllama(model="qwen3", temperature=0)
# --- Node: Intent Classifier ---
def classify_intent(state: AgentState) -> dict:
system_prompt = """You are a query classifier for a loan risk assessment system.
Classify the user query into exactly one of these intents:
- "compliance" - questions about regulatory compliance or rule checks
- "risk_score" - questions about credit risk or default probability
- "full_assessment" - requests for a complete evaluation using all models
- "screen" - requests to screen/filter an applicant through the full pipeline
Respond with ONLY the intent word, nothing else. Do not use any thinking tags."""
response = llm.invoke([
SystemMessage(content=system_prompt),
HumanMessage(content=state["query"])
])
intent = response.content.strip().strip('"').lower()
valid_intents = ["compliance", "risk_score", "full_assessment", "screen"]
if intent not in valid_intents:
intent = "full_assessment"
return {"intent": intent}
# --- Node: Route Decision ---
def route_decision(state: AgentState) -> str:
intent = state["intent"]
if intent == "compliance":
return "routing_node"
elif intent == "risk_score":
return "routing_node"
elif intent == "full_assessment":
return "ensemble_node"
elif intent == "screen":
return "pipeline_node"
return "ensemble_node"
# --- Node: Routing (single model) ---
def routing_node(state: AgentState) -> dict:
applicant = state["applicant_data"]
intent = state["intent"]
predictions = {}
if intent == "compliance":
predictions["rule_engine"] = rule_engine(applicant)
elif intent == "risk_score":
X_input = pd.DataFrame([applicant])[features]
rf_prob = rf_model.predict_proba(X_input)[0][1]
predictions["random_forest"] = {"default_probability": round(float(rf_prob), 4)}
return {"predictions": predictions}
# --- Node: Ensemble (all models) ---
def ensemble_node(state: AgentState) -> dict:
applicant = state["applicant_data"]
X_input = pd.DataFrame([applicant])[features]
lr_prob = logistic_model.predict_proba(X_input)[0][1]
rf_prob = rf_model.predict_proba(X_input)[0][1]
rule_result = rule_engine(applicant)
predictions = {
"logistic_regression": {"default_probability": round(float(lr_prob), 4)},
"random_forest": {"default_probability": round(float(rf_prob), 4)},
"rule_engine": rule_result,
"agreement": abs(lr_prob - rf_prob) < 0.15,
"average_probability": round(float((lr_prob + rf_prob) / 2), 4),
}
return {"predictions": predictions}
# --- Node: Pipeline (sequential) ---
def pipeline_node(state: AgentState) -> dict:
applicant = state["applicant_data"]
X_input = pd.DataFrame([applicant])[features]
# Stage 1: Rule engine filter
rule_result = rule_engine(applicant)
if not rule_result["passed"]:
return {"predictions": {
"pipeline_stage": "rejected_at_rules",
"rule_engine": rule_result,
}}
# Stage 2: Logistic Regression scoring
lr_prob = logistic_model.predict_proba(X_input)[0][1]
# Stage 3: Random Forest refinement
rf_prob = rf_model.predict_proba(X_input)[0][1]
predictions = {
"pipeline_stage": "completed",
"rule_engine": rule_result,
"logistic_regression": {"default_probability": round(float(lr_prob), 4)},
"random_forest": {"default_probability": round(float(rf_prob), 4)},
"final_score": round(float(rf_prob), 4),
}
return {"predictions": predictions}
# --- Build the Graph ---
workflow = StateGraph(AgentState)
workflow.add_node("classify_intent", classify_intent)
workflow.add_node("routing_node", routing_node)
workflow.add_node("ensemble_node", ensemble_node)
workflow.add_node("pipeline_node", pipeline_node)
workflow.add_edge(START, "classify_intent")
workflow.add_conditional_edges("classify_intent", route_decision, {
"routing_node": "routing_node",
"ensemble_node": "ensemble_node",
"pipeline_node": "pipeline_node",
})Your agent now has three distinct ways to process a query. Next up, you'll add SHAP explainability and an LLM synthesis node to turn raw predictions into human-readable assessments.
Add SHAP Explainability and LLM Synthesis
Your agent can now route queries to the right models and collect predictions. The missing piece is transparency. A loan officer needs to understand why the model flagged an applicant, not just that it did.
In this step, you will attach SHAP explainability to the Random Forest predictions, then build a synthesis node where the LLM converts raw numbers into a human-readable assessment.
In this step, get ready to:
- Build a SHAP explanation node that computes feature-level contributions for every prediction.
- Create an LLM synthesis node that converts raw predictions and SHAP values into a plain-English explanation.
- Wire the final graph edges and run the agent end-to-end.
Build the SHAP explanation node
SHAP (SHapley Additive exPlanations) breaks down a model's prediction into contributions from each input feature. Instead of a single "high risk" label, you get a breakdown showing that income pushed the score up by 0.12 while credit score pushed it down by 0.08.
The shap_explainer you created in Step 2 is a TreeExplainer that computes exact SHAP values efficiently for tree-based models like your Random Forest.
- In main.py, find the # --- Build the Graph --- comment. Add the following function directly above it:
# --- Node: SHAP Explanation ---
def explain_with_shap(state: AgentState) -> dict:
applicant = state["applicant_data"]
X_input = pd.DataFrame([applicant])[features]
shap_values = shap_explainer.shap_values(X_input)
# For binary classification, shap_values shape is (n_samples, n_features, 2)
if len(shap_values.shape) == 3:
contributions = shap_values[0, :, 1]
else:
contributions = shap_values[0]
explanation = {
"feature_contributions": {
feat: round(float(val), 4)
for feat, val in zip(features, contributions)
},
"base_value": round(float(shap_explainer.expected_value[1]
if hasattr(shap_explainer.expected_value, '__len__')
else shap_explainer.expected_value), 4),
}
return {"shap_explanation": explanation}What does this code do?
This node takes the applicant data and feeds it through the SHAP explainer to get per-feature contribution scores.
- shap_explainer.shap_values(X_input) computes how much each feature pushed the prediction toward default (positive) or away from it (negative).
- The shape check (if len(shap_values.shape) == 3) handles scikit-learn's binary classification format, which returns a 3D array with values for each class.
- base_value is the model's average prediction across the training set. Feature contributions add or subtract from this baseline to produce the final prediction.
- Save main.py.
You won't see output yet. The next chunk adds the synthesis node that uses these SHAP values.
Seeing a red underline on shap_explainer?
Make sure shap_explainer = shap.TreeExplainer(rf_model) exists earlier in your file (from Step 2). The function references it as a module-level variable.
help me fix my shap_explainer reference
Create the LLM synthesis node
Raw SHAP values are numbers. A loan officer needs a sentence like "the applicant's low income was the primary risk driver." This node hands the predictions and SHAP breakdown to Qwen3 and asks it to write a concise summary.
- In main.py, add the following function directly below the explain_with_shap function you just wrote:
# --- Node: LLM Synthesis ---
def synthesize_response(state: AgentState) -> dict:
system_prompt = """You are a risk assessment analyst. Given prediction results and SHAP explanations,
produce a clear, concise summary for a loan officer. Include:
1. The decision (approve/deny/review)
2. Key risk factors from SHAP values
3. Which models were consulted and whether they agree
Keep it under 150 words. Do not use any thinking tags."""
context = json.dumps({
"predictions": state["predictions"],
"shap_explanation": state.get("shap_explanation", {}),
"orchestration_mode": state["intent"],
}, indent=2)
response = llm.invoke([
SystemMessage(content=system_prompt),
HumanMessage(content=f"Assessment results:\n{context}")
])
return {"final_response": response.content}What does this code do?
- The system prompt tells the LLM to act as a risk analyst and produce a structured summary with a decision, key factors, and model agreement status.
- json.dumps serializes the predictions and SHAP explanation into a formatted string the LLM can read.
- The LLM never predicts risk itself. It only synthesizes results that the deterministic models already computed. This keeps the system auditable.
- Save main.py.
You won't see a change yet. The next chunk wires these two nodes into the graph and runs everything.
Why doesn't the LLM do the prediction itself?
LLMs are non-deterministic. The same input can produce different outputs on different runs. For financial decisions that require auditability and reproducibility, deterministic ML models handle the computation. The LLM handles natural language understanding and explanation only.
Syntax error on the triple-quoted string?
Make sure you have three double-quotes (""") at both the opening and closing of the system prompt. A missing quote will cause a syntax error across multiple lines.
help me fix my synthesize_response function
Wire the graph and run the agent
With both new nodes written, you need to register them in the StateGraph and connect edges so every orchestration path flows through SHAP explanation, then synthesis, then exits.
- In main.py, scroll down to the # --- Build the Graph --- section. Find the line:
workflow.add_node("pipeline_node", pipeline_node)- Add these two lines directly below it to register the new nodes:
workflow.add_node("explain_with_shap", explain_with_shap)
workflow.add_node("synthesize_response", synthesize_response)- Now find the add_conditional_edges call (the last line in the graph section). Add these edges directly below it:
workflow.add_edge("routing_node", "explain_with_shap")
workflow.add_edge("ensemble_node", "explain_with_shap")
workflow.add_edge("pipeline_node", "explain_with_shap")
workflow.add_edge("explain_with_shap", "synthesize_response")
workflow.add_edge("synthesize_response", END)
agent = workflow.compile()What do these edges do?
- All three orchestration nodes (routing, ensemble, pipeline) converge into the same SHAP explanation node. Every prediction path gets explainability.
- explain_with_shap feeds into synthesize_response, which feeds into END. This creates a linear tail after the branching logic.
- workflow.compile() validates the graph structure and produces a runnable agent.
- Save main.py.
The graph is complete. Now add the test harness so you can actually run the agent.
- At the very bottom of main.py, add the following block:
# --- Run the Agent ---
if __name__ == "__main__":
sample_applicant = {
"income": 45000,
"debt_to_income": 0.38,
"credit_score": 650,
"employment_years": 3.5,
"loan_amount": 30000,
}
queries = [
"Is this applicant compliant with lending regulations?",
"What is the default risk for this customer?",
"Give me a full risk assessment with all models.",
"Screen this applicant through the complete pipeline.",
]
for query in queries:
print(f"\n{'='*60}")
print(f"QUERY: {query}")
print(f"{'='*60}")
result = agent.invoke({
"query": query,
"applicant_data": sample_applicant,
"intent": "",
"predictions": {},
"shap_explanation": {},
"final_response": "",
})
print(f"INTENT: {result['intent']}")
print(f"PREDICTIONS: {json.dumps(result['predictions'], indent=2)}")
print(f"\nRESPONSE:\n{result['final_response']}")What does this code do?
- sample_applicant is a test profile with moderate income, reasonable debt-to-income ratio, and a middling credit score. It is designed to produce interesting (non-obvious) results.
- The four queries exercise all three orchestration modes: compliance and risk_score trigger routing, full assessment triggers ensemble, and screen triggers the pipeline.
- agent.invoke() runs the full graph from START to END, passing through intent classification, the selected orchestration node, SHAP explanation, and LLM synthesis.
- Save main.py.
- Run the agent by executing this command in your terminal:
python main.pyYou should see four query/response pairs printed to your terminal. Each one shows the classified intent, the raw predictions (as JSON), and a natural-language response from the LLM.
What should I see?
The output includes four sections separated by lines of equals signs. Each section prints the query, the classified intent (like "compliance" or "full_assessment"), a JSON block of predictions, and a plain-English response from the LLM summarizing the risk assessment.
The first run may take 30-60 seconds as Ollama loads Qwen3 into memory. Subsequent runs are faster.
Seeing a connection error or timeout?
Make sure Ollama is running in the background. Open a separate terminal and run ollama list to confirm the qwen3 model is available. If Ollama is not running, start it before retrying.
help me debug my Ollama connection
✔️ Awesome, I've got everything!
Great. Double check you have saved your file.
ⓧ I'd like to double check the full code
Here is the complete main.py as it should look at the end of this step:
import json
from typing import TypedDict, Literal
import numpy as np
import pandas as pd
from sklearn.linear_model import LogisticRegression
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
import shap
from langgraph.graph import StateGraph, START, END
from langchain_ollama import ChatOllama
from langchain.messages import SystemMessage, HumanMessage
# --- State Definition ---
class AgentState(TypedDict):
query: str
intent: str
applicant_data: dict
predictions: dict
shap_explanation: dict
final_response: str
# --- Synthetic Dataset ---
def generate_loan_data(n=1000):
np.random.seed(42)
data = pd.DataFrame({
"income": np.random.normal(55000, 15000, n).clip(20000, 150000),
"debt_to_income": np.random.uniform(0.1, 0.7, n),
"credit_score": np.random.normal(680, 50, n).clip(300, 850).astype(int),
"employment_years": np.random.exponential(5, n).clip(0, 30),
"loan_amount": np.random.normal(25000, 10000, n).clip(5000, 100000),
})
default_prob = (
0.3 * (data["debt_to_income"] > 0.43).astype(float)
+ 0.25 * (data["credit_score"] < 620).astype(float)
+ 0.2 * (data["income"] < 35000).astype(float)
+ 0.15 * (data["employment_years"] < 1).astype(float)
+ 0.1 * (data["loan_amount"] > 50000).astype(float)
)
data["default"] = (default_prob + np.random.normal(0, 0.15, n) > 0.45).astype(int)
return data
# --- Train Models ---
data = generate_loan_data()
features = ["income", "debt_to_income", "credit_score", "employment_years", "loan_amount"]
X = data[features]
y = data["default"]
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)
logistic_model = LogisticRegression(max_iter=1000, random_state=42)
logistic_model.fit(X_train, y_train)
rf_model = RandomForestClassifier(n_estimators=100, max_depth=5, random_state=42)
rf_model.fit(X_train, y_train)
# SHAP explainer for Random Forest
shap_explainer = shap.TreeExplainer(rf_model)
# --- Rule Engine ---
def rule_engine(applicant: dict) -> dict:
flags = []
passed = True
if applicant["debt_to_income"] > 0.43:
flags.append("FAIL: Debt-to-income ratio exceeds 0.43 regulatory threshold")
passed = False
if applicant["credit_score"] < 580:
flags.append("FAIL: Credit score below 580 minimum")
passed = False
if applicant["employment_years"] < 0.5:
flags.append("WARNING: Less than 6 months employment history")
if passed and not flags:
flags.append("PASS: All regulatory checks passed")
return {"passed": passed, "flags": flags}
# --- LLM Setup ---
llm = ChatOllama(model="qwen3", temperature=0)
# --- Node: Intent Classifier ---
def classify_intent(state: AgentState) -> dict:
system_prompt = """You are a query classifier for a loan risk assessment system.
Classify the user query into exactly one of these intents:
- "compliance" - questions about regulatory compliance or rule checks
- "risk_score" - questions about credit risk or default probability
- "full_assessment" - requests for a complete evaluation using all models
- "screen" - requests to screen/filter an applicant through the full pipeline
Respond with ONLY the intent word, nothing else. Do not use any thinking tags."""
response = llm.invoke([
SystemMessage(content=system_prompt),
HumanMessage(content=state["query"])
])
intent = response.content.strip().strip('"').lower()
valid_intents = ["compliance", "risk_score", "full_assessment", "screen"]
if intent not in valid_intents:
intent = "full_assessment"
return {"intent": intent}
# --- Node: Route Decision ---
def route_decision(state: AgentState) -> str:
intent = state["intent"]
if intent == "compliance":
return "routing_node"
elif intent == "risk_score":
return "routing_node"
elif intent == "full_assessment":
return "ensemble_node"
elif intent == "screen":
return "pipeline_node"
return "ensemble_node"
# --- Node: Routing (single model) ---
def routing_node(state: AgentState) -> dict:
applicant = state["applicant_data"]
intent = state["intent"]
predictions = {}
if intent == "compliance":
predictions["rule_engine"] = rule_engine(applicant)
elif intent == "risk_score":
X_input = pd.DataFrame([applicant])[features]
rf_prob = rf_model.predict_proba(X_input)[0][1]
predictions["random_forest"] = {"default_probability": round(float(rf_prob), 4)}
return {"predictions": predictions}
# --- Node: Ensemble (all models) ---
def ensemble_node(state: AgentState) -> dict:
applicant = state["applicant_data"]
X_input = pd.DataFrame([applicant])[features]
lr_prob = logistic_model.predict_proba(X_input)[0][1]
rf_prob = rf_model.predict_proba(X_input)[0][1]
rule_result = rule_engine(applicant)
predictions = {
"logistic_regression": {"default_probability": round(float(lr_prob), 4)},
"random_forest": {"default_probability": round(float(rf_prob), 4)},
"rule_engine": rule_result,
"agreement": abs(lr_prob - rf_prob) < 0.15,
"average_probability": round(float((lr_prob + rf_prob) / 2), 4),
}
return {"predictions": predictions}
# --- Node: Pipeline (sequential) ---
def pipeline_node(state: AgentState) -> dict:
applicant = state["applicant_data"]
X_input = pd.DataFrame([applicant])[features]
# Stage 1: Rule engine filter
rule_result = rule_engine(applicant)
if not rule_result["passed"]:
return {"predictions": {
"pipeline_stage": "rejected_at_rules",
"rule_engine": rule_result,
}}
# Stage 2: Logistic Regression scoring
lr_prob = logistic_model.predict_proba(X_input)[0][1]
# Stage 3: Random Forest refinement
rf_prob = rf_model.predict_proba(X_input)[0][1]
predictions = {
"pipeline_stage": "completed",
"rule_engine": rule_result,
"logistic_regression": {"default_probability": round(float(lr_prob), 4)},
"random_forest": {"default_probability": round(float(rf_prob), 4)},
"final_score": round(float(rf_prob), 4),
}
return {"predictions": predictions}
# --- Node: SHAP Explanation ---
def explain_with_shap(state: AgentState) -> dict:
applicant = state["applicant_data"]
X_input = pd.DataFrame([applicant])[features]
shap_values = shap_explainer.shap_values(X_input)
# For binary classification, shap_values shape is (n_samples, n_features, 2)
if len(shap_values.shape) == 3:
contributions = shap_values[0, :, 1]
else:
contributions = shap_values[0]
explanation = {
"feature_contributions": {
feat: round(float(val), 4)
for feat, val in zip(features, contributions)
},
"base_value": round(float(shap_explainer.expected_value[1]
if hasattr(shap_explainer.expected_value, '__len__')
else shap_explainer.expected_value), 4),
}
return {"shap_explanation": explanation}
# --- Node: LLM Synthesis ---
def synthesize_response(state: AgentState) -> dict:
system_prompt = """You are a risk assessment analyst. Given prediction results and SHAP explanations,
produce a clear, concise summary for a loan officer. Include:
1. The decision (approve/deny/review)
2. Key risk factors from SHAP values
3. Which models were consulted and whether they agree
Keep it under 150 words. Do not use any thinking tags."""
context = json.dumps({
"predictions": state["predictions"],
"shap_explanation": state.get("shap_explanation", {}),
"orchestration_mode": state["intent"],
}, indent=2)
response = llm.invoke([
SystemMessage(content=system_prompt),
HumanMessage(content=f"Assessment results:\n{context}")
])
return {"final_response": response.content}
# --- Build the Graph ---
workflow = StateGraph(AgentState)
workflow.add_node("classify_intent", classify_intent)
workflow.add_node("routing_node", routing_node)
workflow.add_node("ensemble_node", ensemble_node)
workflow.add_node("pipeline_node", pipeline_node)
workflow.add_node("explain_with_shap", explain_with_shap)
workflow.add_node("synthesize_response", synthesize_response)
workflow.add_edge(START, "classify_intent")
workflow.add_conditional_edges("classify_intent", route_decision, {
"routing_node": "routing_node",
"ensemble_node": "ensemble_node",
"pipeline_node": "pipeline_node",
})
workflow.add_edge("routing_node", "explain_with_shap")
workflow.add_edge("ensemble_node", "explain_with_shap")
workflow.add_edge("pipeline_node", "explain_with_shap")
workflow.add_edge("explain_with_shap", "synthesize_response")
workflow.add_edge("synthesize_response", END)
agent = workflow.compile()
# --- Run the Agent ---
if __name__ == "__main__":
sample_applicant = {
"income": 45000,
"debt_to_income": 0.38,
"credit_score": 650,
"employment_years": 3.5,
"loan_amount": 30000,
}
queries = [
"Is this applicant compliant with lending regulations?",
"What is the default risk for this customer?",
"Give me a full risk assessment with all models.",
"Screen this applicant through the complete pipeline.",
]
for query in queries:
print(f"\n{'='*60}")
print(f"QUERY: {query}")
print(f"{'='*60}")
result = agent.invoke({
"query": query,
"applicant_data": sample_applicant,
"intent": "",
"predictions": {},
"shap_explanation": {},
"final_response": "",
})
print(f"INTENT: {result['intent']}")
print(f"PREDICTIONS: {json.dumps(result['predictions'], indent=2)}")
print(f"\nRESPONSE:\n{result['final_response']}")Secret mission
Disagreement Detector with Human-in-the-Loop
Your agent trusts its models blindly. But what happens when the Logistic Regression says 'approve' and the Random Forest says 'deny'? In this mission, you'll build a confidence-calibrated disagreement detector that catches these conflicts and escalates them to a human reviewer before a decision is made.
Clean Up Your Resources
Clean Up Your Resources
This project runs entirely locally, so there are no ongoing costs. Decide whether to keep your resources as-is or free up disk space by removing the Qwen3 model.
Resources you used:
- Qwen3 model stored on disk via Ollama.
- Project folder containing main.py on your Desktop.
- Python packages installed via pip (langgraph, langchain-ollama, scikit-learn, shap, pandas).
Keep everything running
No action needed. Choose this if you want to keep experimenting with the agent or tweak the orchestration modes.
- The Qwen3 model stays on disk, ready for instant use next time you run the agent.
- Your Python packages remain installed in your environment.
- There are no cloud resources or subscriptions generating charges.
Pause - I'll come back to this later
Shut down any running processes to free up memory. Your files and model stay on disk for later.
- If Ollama is still running in the background, stop it by closing the Ollama application or stopping the service.
- Close any terminal sessions that were running the agent.
- Your project files and the Qwen3 model remain on disk. You can pick up right where you left off.
Delete - I don't want to use this again
Remove all project resources and free up disk space.
- Delete the Qwen3 model from Ollama by running this command in your terminal:
ollama rm qwen3- Delete your project folder. In Finder (macOS) or File Explorer (Windows), navigate to your Desktop and delete the folder containing main.py.
- Uninstall the Python packages by running:
pip uninstall langgraph langchain-ollama scikit-learn shap pandas -yWhat about Ollama itself?
Ollama was a prerequisite you installed before this project. If you still want it for other projects, leave it installed. If you want to remove it entirely, follow the uninstall instructions on the official Ollama website for your operating system.
Nice Work!
Nice Work!
You just built a fully local, zero-cost hybrid AI agent that separates reasoning from computation. Your LangGraph agent orchestrates multiple ML models, explains every prediction with SHAP, and synthesizes human-readable summaries through a local LLM.
You've learned how to:
- Build a LangGraph StateGraph that orchestrates multiple ML models via conditional routing, dispatching queries to the right model based on classified intent.
- Implement three orchestration patterns (routing, ensemble, and pipeline) that give you flexible ways to combine deterministic and probabilistic models.
- Integrate SHAP TreeExplainer for feature-level prediction explanations, with a local Qwen3 LLM synthesizing the results into plain-language summaries.
- Secret Mission: Added a disagreement detector that flags cases where models diverge and triggers a human-in-the-loop escalation path using LangGraph's interrupt mechanism.
Ready to quiz yourself?