云信AI部署

Python + LangChain：开发智能金融风控系统

云信安装大师
90
AI 质量分
2 5 月, 2025
3 分钟阅读
0 阅读

Python + LangChain：开发智能金融风控系统

引言

在金融科技领域，风险控制是核心业务之一。传统风控系统依赖规则引擎，而现代智能风控系统则结合了AI技术。本文将介绍如何使用Python和LangChain框架构建一个智能金融风控系统原型，它能自动分析交易数据并识别潜在风险。

准备工作

环境要求

  • Python 3.8+
  • pip包管理工具
  • 推荐使用Jupyter Notebook进行实验

安装依赖

代码片段 
pip install langchain openai pandas numpy python-dotenv

获取API密钥

需要准备OpenAI API密钥(或其他LLM服务商密钥)，保存到.env文件：

代码片段 
OPENAI_API_KEY=your_api_key_here

核心组件设计

我们的智能风控系统将包含以下模块：

  1. 数据预处理：清洗和标准化交易数据
  2. 风险评估：使用LLM分析交易特征
  3. 决策引擎：基于风险评分做出决策
  4. 解释生成：提供人类可读的风险解释

完整实现代码

1. 初始化环境

代码片段 
import os
from dotenv import load_dotenv
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
from langchain.llms import OpenAI

# 加载环境变量
load_dotenv()

# 初始化LLM (温度设为0以获得更确定性的输出)
llm = OpenAI(temperature=0, model_name="gpt-3.5-turbo")

2. 数据预处理模块

代码片段 
import pandas as pd

def preprocess_transaction_data(filepath):
    """
    预处理交易数据

    参数:
        filepath: CSV文件路径

    返回:
        处理后的DataFrame和特征字典
    """
    df = pd.read_csv(filepath)

    # 示例清洗步骤 - 实际应根据业务需求调整
    df = df.dropna()  # 删除缺失值

    # 特征工程 - 创建有用的特征
    df['amount_to_balance_ratio'] = df['amount'] / df['account_balance']
    df['is_night'] = (df['hour'] >= 22) | (df['hour'] <= 5)

    # 转换为适合LLM处理的字典格式
    features = df.to_dict('records')

    return df, features

# 示例交易数据结构 (实际应从CSV加载)
sample_data = {
    "transaction_id": ["TX1001", "TX1002", "TX1003"],
    "amount": [1500, 50000, 200],
    "account_balance": [5000, 100000, 300],
    "location": ["New York", "Tokyo", "London"],
    "hour": [14, 3, 23],
    "merchant_category": ["electronics", "jewelry", "grocery"]
}

# 保存为CSV用于演示 (实际项目应从数据库获取)
pd.DataFrame(sample_data).to_csv("sample_transactions.csv", index=False)

# 预处理数据
df, features = preprocess_transaction_data("sample_transactions.csv")
print("预处理后的交易特征:", features[0])

3.风险评估提示模板

代码片段 
risk_assessment_template = """
你是一个专业的金融风控专家。请分析以下交易信息并评估风险:

交易ID: {transaction_id}
金额: ${amount}
账户余额: ${account_balance}
金额/余额比: {amount_to_balance_ratio:.2f}
地点: {location}
时间: {hour}:00 
商户类别: {merchant_category}
是否为夜间交易: {is_night}

请按以下步骤分析:
1. [风险评估]给出1-10的风险评分(10为最高风险)
2. [风险因素]列出所有可疑因素 
3. [建议操作]建议采取的措施(允许/人工审核/拒绝)

请用以下JSON格式回复:
{{
    "risk_score": score,
    "risk_factors": [factor1, factor2,...],
    "recommendation": "action"
}}
"""

prompt = PromptTemplate(
    input_variables=[
        "transaction_id",
        "amount",
        "account_balance",
        "amount_to_balance_ratio",
        "location",
        "hour",
        "merchant_category",
        "is_night"
    ],
    template=risk_assessment_template,
)

risk_assessment_chain = LLMChain(llm=llm, prompt=prompt)

4.执行风险评估

代码片段 
def assess_transaction_risk(features):
    """评估单笔交易风险"""

    results = []

    for feature in features:
        # LangChain自动填充模板变量并调用LLM
        assessment = risk_assessment_chain.run(**feature)

        try:
            # JSON字符串转换为字典以便后续处理 
            assessment_dict = eval(assessment)  
            assessment_dict["transaction_id"] = feature["transaction_id"]
            results.append(assessment_dict)

            print(f"评估结果 - TXID:{feature['transaction_id']}")
            print(f"风险评分: {assessment_dict['risk_score']}")
            print(f"风险因素: {', '.join(assessment_dict['risk_factors'])}")
            print(f"建议操作: {assessment_dict['recommendation']}\n")

        except Exception as e:
            print(f"解析评估结果失败: {e}")

    return results

#评估所有样本交易 
risk_results = assess_transaction_risk(features[:3])

5.决策引擎实现

代码片段 
def make_decision(risk_results):
    """基于风险评估做出决策"""

    decisions = []

    for result in risk_results:
        tx_id = result["transaction_id"]

        if result["risk_score"] >=8:
            decision = {"tx_id": tx_id, "decision": "REJECT", 
                       "reason": f"高风险评分({result['risk_score']})"}

        elif result["risk_score"] >=5:
            decision = {"tx_id": tx_id, "decision": "MANUAL_REVIEW", 
                       "reason": f"中等风险({result['risk_score']})"}

        else:
            decision = {"tx_id": tx_id, "decision": "APPROVE", 
                       "reason": f"低风险({result['risk_score']})"}

        decisions.append(decision)

        print(f"决策 - TXID:{tx_id}: {decision['decision']} ({decision['reason']})")

    return decisions 

final_decisions = make_decision(risk_results)

实践经验与注意事项

1.性能优化：
– LangChain的batch功能可以批量处理请求减少API调用延迟：

代码片段 
assessments = risk_assessment_chain.apply([dict(row) for row in features])

2.错误处理：
– LLM输出可能不符合预期JSON格式，添加验证逻辑：

代码片段 
import json 

def parse_response(response):
   try:
       return json.loads(response.strip())
   except json.JSONDecodeError:
       print("响应不是有效JSON")
       return None

3.本地模型替代方案：
如果不想依赖OpenAI API，可以使用本地模型：

代码片段 
from langchain.llms import LlamaCpp 

llm_local = LlamaCpp(
   model_path="./models/ggml-model-q4_0.bin",
   temperature=0,
   max_tokens=2000,
   verbose=True  
)

4.提示工程技巧：
-在提示中提供更多示例可以获得更一致的输出格式
-明确指定数字范围（如1-10分制）比模糊描述更好

5.生产环境考虑：
-添加日志记录所有决策和原因
-实现缓存机制避免重复分析相同模式
-考虑使用LangChain的SequentialChain构建更复杂的工作流

总结

我们实现了一个基于Python和LangChain的智能风控系统原型，它能够：

  1. 处理原始交易数据并进行特征工程
  2. 利用LLM进行风险评估并生成结构化结果
  3. 自动做出决策(通过/人工审核/拒绝)

关键优势：
✅比传统规则引擎更灵活
✅可解释性强（提供风险因素列表）
✅易于集成到现有系统

扩展方向：
➡️添加历史行为分析（使用向量数据库）
➡️结合传统机器学习模型（如随机森林）进行混合判断

原创 高质量