简介：Grab风控团队将大模型与智能体技术落地到复杂场景数据分析的完整实战方案，适合风控算法工程师、数据产品经理及关注AI应用的技术管理者学习。内容围绕知识注入与流程注入两条主线，重点拆解RAG检索增强生成、SOP驱动的智能体框架、混合检索引擎、树形结构SOP规划、动态工具调用与状态管理等关键机制，并给出Analytics Copilot和RiskOps Autopilot两个真实业务案例，可帮助读者理解如何用技术手段解决海量高维数据、数据孤岛、专家依赖等风控难点。压缩包内共1个文件，为pptx演示文稿，约7.81MB，包含大量架构图、流程图和模块说明，适合直接阅读、分享或二次整理。当前已有59人学习，对想了解大模型与智能体在企业复杂风控场景中从架构设计到落地评估全过程的读者很有参考价值。
1. 把"大模型+智能体"塞进风控数据分析前，先想清楚它到底在解决什么
一份名叫《基于大模型与智能体的复杂场景数据分析在Grab风控场景中的实践与探索》的分享材料，拆开来看，其实是一套"用自然语言问数据、让模型自动分析风险"的落地方法。Grab这类平台，司机刷单、支付盗刷、薅羊毛、虚假评论这些风险数据散落在订单、设备、GPS、文本投诉和交易流水里，传统方式靠的是数据分析师手写SQL、再人工拼一张宽表，排查一次要半天起步。大模型负责把业务问题翻译成分析任务，智能体负责调工具、跑SQL、看报错、反复迭代，最终交付一个带证据链的结论。这套思路适合正在做风控数据分析，又想尝试大模型和智能体框架的从业者。
2. Grab风控里的复杂场景数据难在哪：从规则引擎到大模型智能体的选型理由
2.1 传统风控特征分析为什么处理不了复杂场景
Grab的风控不是单一的"这笔交易是否违规"，而是由一连串行为构成的。比如一个设备在凌晨四点密集切换三个账号，配合新注册的支付账号在一家店铺反复下单。要看穿这个链条，需要同时参照订单事实表、设备指纹、注册信息、GPS轨迹和支付流水，而且这些表之间的关联关系不是一对一，时间窗口还可能跨小时甚至跨天。这种数据形态有四个特点：异构、长尾、时序、对抗。异构意味着不能靠一张宽表描述所有模式；长尾意味着风险正样本少，而且不断变化；时序意味着要看事件顺序，不能用截面数据简单交叉；对抗意味着攻击者会根据规则调整行为，数据分布时刻在漂移。
在这种场景下，传统规则引擎和机器学习模型都有明显短板。规则引擎比如"30分钟内同一设备下单超过3次"，它能覆盖高频已知攻击，但攻击者稍微改一下参数就能绕过。机器学习模型需要人工构造特征，特征工程通常来自复盘，这意味着攻击模式一变，特征就失效，等模型重新训练上线，黑产已经换赛道了。还有一个常被忽略的瓶颈是数据分析本身。风控策略分析师在排查一个可疑玩法时，要手写几十行SQL、多次join、再跑一段Python算窗口特征，最后还要人工解释。一个做风控策略的同事说过，他最怕的不是模型效果差，而是每次排查都像在旧仓库里刨线头，刨了半天才摸到一条线索。
大模型和智能体之所以有机会，恰恰是因为它们能承担一部分"从问题到特征再到结论"的链路。大模型可以读懂"司机是不是在配合刷单"这种业务语言，智能体可以把"用SQL关联两张表、按时间排序、统计设备复用率"这种步骤自动执行出来。两者组合后，策略同学只需要描述问题场景，不需要先学会表结构。这就是整个方向最核心的价值：把数据分析从几天压缩到几分钟，同时让长尾模式的发现不再依赖某位老师傅的个人经验。
2.2 大模型负责理解，智能体负责执行：角色拆分
大模型不是数据库的替代品，它是对业务语言的解释器。同一个词"订单异常"在不同规则下含义完全不同，可能是金额异常、频次异常、设备异常。大模型的任务是把这些模糊表述转成可验证的SQL条件和统计指标。但大模型也有天生缺陷：它没有真实的执行环境，它不知道这张表里到底有没有数据、这个字段是int还是string、这条SQL会不会跑三分钟。这时候智能体的价值就体现出来了。
智能体本质上是一个执行管理器。它维护当前任务的目标、已执行的步骤、工具返回的结果，并决定下一步动作。我常把智能体理解成一个"有手有脚"的插件系统：大模型负责思考，智能体负责让思考落地。落地方式就是工具调用。当模型输出一个调用SQL工具的请求时，智能体负责校验参数、限流、执行、截断结果，再把结果作为新的观察输入给模型。这样形成了一个"推理-行动-观察"的闭环，也就是ReAct范式。在实际项目里，不一定需要引入复杂框架，一个循环加一组工具函数就可以工作，关键是这个循环要能控制步骤数、错误重试和上下文长度。
选型理由可以归纳为：纯Prompt工程解决的是"一次问答"，适合单个问题；纯自动化脚本解决的是"固定流程"，适合已知场景；大模型加智能体适合的是"开放的排查过程"——问题可能是任何形式，分析的中间结果会改变下一步行动。风控复杂场景恰好是后者。而且Grab这种业务往往横跨出行、外卖和支付，同一个用户在不同业务线的行为要串起来看，智能体天然适合做这种跨域数据串联。
2.3 智能体框架与工具调用的基本范式
实现这个闭环，业内常见的路子有三条：直接用LangChain这类通用智能体框架、用Dify这类低代码平台快速搭demo、或者自己实现一个几十行的工具调用循环。我在风控场景里优先推荐第三条，原因有两个：一是风控数据敏感，查询权限、审计日志和超时控制都需要侵入到Agent代码里，通用框架封装太深，出了问题很难控制；二是风控团队往往已经有稳定的数据服务和指标服务，Agent只需要做"调用"和"组合"，不需要重新发明。
工具调用的本质是模型输出一个结构化JSON，而不是自然语言。这个JSON描述了要调用哪个函数、传入什么参数。运行时负责执行函数，再把真实结果返回给模型。这样做的好处是模型不会直接拼接系统命令，所有敏感操作都收敛在少量被注册的函数里。在风控场景这是个安全底线。企业Agent如果要去本地查询业务数据库，也必须走这条路线：大模型在发起端做意图理解，SQL执行永远在受控的数据服务端完成。
无论用哪条路，核心都是Function Calling。在后面的代码里我会跳过LangChain封装，直接写一个最小的循环实现。不是否定框架，而是用最少的代码把原理讲透；实际接入时，你再把权限、日志、超时这些组件填进去就行。
3. 搭一个最小可用的风控数据分析智能体：架构与代码
3.1 整体架构：数据层、智能体层、应用层
我会把整套东西拆成三层。最底下是数据层，包含一张放行的"宽表"或者数据仓库里的多张明细表，以及一个只读账号。中间是智能体层，由一个LLM、一组工具函数和一个循环控制器组成。最上面是应用层，可以是一个Jupyter Notebook、一个Web问答页面，或者一个定时跑批的任务。
在数据层，最好预先把常用表结构整理成一份Schema描述，包含表名、关键字段、枚举值和常见口径。这一步很多人会偷懒，但它直接决定大模型写SQL的准确率。你可以把这份Schema塞到系统Prompt里，或者做成工具让Agent动态查information_schema。我之前做过一个项目，把Schema描述从"一句话"扩充成"字段说明+示例值+口径备注"之后，SQL生成准确率从不到60%升到85%，这个收益远比换一个更大的模型来得快。
3.2 工具层的核心：把SQL查询和统计函数暴露给大模型
工具层的设计原则是"让大模型用最少的操作拿到可分析的结果"。我一般只暴露四类工具：执行SQL、快速统计、获取表结构、返回可画图的数据。不要轻易暴露写操作，也不要让Agent直接去执行任意Python代码，否则风控数据的安全边界会失控。
每个工具都定义成一个函数，并配一个说明。大模型会依据说明决定何时调用。比如run_sql工具的说明里要写清楚"只接受SELECT，返回前100行，禁止无LIMIT的查询"。大模型看到之后会自己收敛行为。这个说明本身也是Prompt工程的一部分，写得越具体，越少翻车。下面这部分代码就是围绕这一个核心设计的。
3.3 一个最小可用的数据分析Agent：完整代码
考虑到风控场景的定制性，我用OpenAI兼容接口直接写循环，不依赖特定框架版本。下面的代码假设你已经有大模型服务的base_url和api_key，并且有一个只读数据库连接。
import json import pandas as pd from openai import OpenAI from sqlalchemy import create_engine # 只读账号连接，应用层永远不要用管理员账号 engine = create_engine("mysql+pymysql://risk_readonly:xxxx@10.0.0.5:3306/risk_db?charset=utf8mb4") client = OpenAI(base_url="http://your-llm-service:8000/v1", api_key="local-key") # 工具定义：结构化的函数声明，模型会按JSON Schema输出调用意图 TOOLS = [ { "type": "function", "function": { "name": "run_sql", "description": "执行只读SQL查询，返回前50行结果。只允许SELECT语句，必须带LIMIT。", "parameters": { "type": "object", "properties": { "query": {"type": "string", "description": "完整SQL语句"} }, "required": ["query"] } } }, { "type": "function", "function": { "name": "describe_table", "description": "返回指定表的字段名、类型和备注，用于确认表结构。", "parameters": { "type": "object", "properties": { "table": {"type": "string", "description": "表名"} }, "required": ["table"] } } } ] def run_sql(query: str) -> str: # 防御式校验：宁可拒掉也不让危险SQL进入源库 if not query.strip().lower().startswith("select"): return "错误：只允许SELECT查询" if ";" in query.rstrip(";"): return "错误：禁止多语句执行" if "limit" not in query.lower(): query += " LIMIT 50" try: df = pd.read_sql(query, engine) return df.head(50).to_json(orient="records", force_ascii=False) except Exception as e: # 把结构化错误信息返回给模型，而不是让它瞎猜 return f"SQL执行失败: {str(e)}" def describe_table(table: str) -> str: df_schema = pd.read_sql(f"DESCRIBE {table}", engine) return df_schema.to_json(orient="records", force_ascii=False) TOOL_MAP = { "run_sql": run_sql, "describe_table": describe_table } def run_agent(user_question: str, max_steps: int = 8): # 系统Prompt里放角色边界和口径，用户消息放具体问题 messages = [ {"role": "system", "content": "你是一个风控数据分析助手。先确认表和字段，再写SQL，执行后基于真实结果回答。"}, {"role": "user", "content": user_question} ] for step in range(max_steps): resp = client.chat.completions.create( model="qwen-plus", # 本地方案可替换为Qwen、DeepSeek等支持function calling的模型 messages=messages, tools=TOOLS, temperature=0.1, # 风控分析要低随机性，避免结论漂移 top_p=0.9, max_tokens=1024 ) msg = resp.choices[0].message # 没有工具调用意图，说明模型认为可以给出最终答复 if msg.tool_calls is None: return msg.content # 工具调用结果要挂到消息历史里，模型才能在下一轮看到结果 messages.append(msg) for tc in msg.tool_calls: args = json.loads(tc.function.arguments) result = TOOL_MAP[tc.function.name](**args) # 强制截断，防止查询结果撑爆上下文 messages.append({ "role": "tool", "tool_call_id": tc.id, "content": str(result)[:2000] }) return "达到最大步数，任务未完成，请拆分问题后重试"这段代码是最小闭环的逻辑说明：模型每一轮都可以选择继续调工具或者直接回答；工具结果以"tool"角色的消息回传，让模型知道"刚才那条SQL真的执行了，结果是什么"；循环通过max_steps兜底。注意temperature设成了0.1，因为在风控分析里我们不希望同样的问法每次跑出不同结论；top_p保持0.9，避免采样过于激进；max_tokens给1024，用来限制单次生成长度，防止模型一次输出过长导致后续轮次可用token不足。
3.4 大模型选型与关键参数说明
选型时优先看三个能力：是否支持function calling、上下文长度是否足够容纳工具结果、中文业务语义理解是否靠谱。如果数据不能出内网，就用本地部署的开源模型；如果安全边界允许，也可以走云上API。初期不建议直接做大模型微调实战，先把你Schema和口径写进Prompt，通常能解决大部分问题。等积累了足够多的"问题-SQL"标注pair，再考虑微调也不迟。
参数上，除了上面提到的temperature和max_tokens，还有一个容易被忽略的max_steps。模型在复杂分析里可能会先describe_table再写SQL，再根据报错修改SQL，这一步就到3轮了。我一版把max_steps设置在6到12之间，太低完不成任务，太高容易出现无效循环。另外，工具返回的截断长度也要控制，2000字符对大多数风控分析足够了，太长反而稀释模型对任务的注意力。
4. 让智能体在风控数据上干具体活：意图解析、SQL生成与结果校验
4.1 把业务问题翻译成Agent任务：Prompt模板设计
系统Prompt不能只是"你是一个助手"这种空话，要把表结构和业务口径都写进去。我一般会在Prompt里放一个固定模板，把数据字典和口径说明放在最前面。下面是我经常使用的精简模板结构：
SYSTEM_PROMPT = """ 你是Grab风控部的数据分析Agent，负责把业务问题转成SQL并给出结论。 ## 可用表 - orders: 订单表。字段：order_id, driver_id, user_id, city, amount, create_time, status - device_login: 设备登录表。字段：id, user_id, device_id, login_time - payment: 支付表。字段：payment_id, order_id, pay_time, pay_method, pay_result ## 重要口径 - 刷单嫌疑：同一device_id在1小时内关联超过3个user_id，或同一user_id在1小时内创建超过2笔订单。 - 薅羊毛：新注册用户7天内使用首单优惠券超过3次。 ## 分析步骤要求 1. 先根据用户问题确定所需表和关键时间窗口。 2. 如果表结构不确定，先调用describe_table确认。 3. 执行SQL前先在心里过一遍：字段名是否存在，JOIN条件是否正确。 4. 基于SQL真实返回结果回答，严禁编造数字。 ## 输出格式 先用一段话概述结论，再用列表给出关键数据支撑。 """这个模板解决了两件事：一是让模型知道"刷单嫌疑"在公司内部是怎么定义的，而不是自己去按字面理解；二是通过输出格式约束，保证它回答时不会丢证据。实践中，很多Agent跑偏都是因为口径不写清楚，模型把"刷单"理解成了"重复下单"，拿到结果自然跟业务对不上。
4.2 给智能体划定边界：表结构约束、会话隔离与权限校验
智能体在风控场景里必须被当成一个"半可信执行者"。我在生产环境里会再加三道防线。
第一，数据库账号只读，且只能访问风险分析库，不用主库。这是最底线的保护。如果被授权了宽表，那就在这一层把不需要的敏感字段脱敏掉。第二，在run_sql函数里做语句层面的拦截，防止模型拼出多语句或非查询操作。第三，按业务线做会话隔离，订单分析Agent不能去访问支付团队的原始流水表，除非在高级别风险排查里由用户显式授权。
拦截逻辑可以这样加在工具函数里：
import re def safe_run_sql(query: str) -> str: # 只允许单条SELECT语句 if not re.match(r"^\s*select\s", query, re.IGNORECASE): return "错误：工具只能执行SELECT语句" if ";" in query.rstrip(";"): return "错误：不能执行多语句" # 禁止明显全表扫描 if re.search(r"\bjoin\s+\w+\s+without\s+limit", query, re.IGNORECASE): return "错误：JOIN查询必须携带LIMIT" return run_sql(query)这个例子里，我用正则做了一次粗校验，实际生产可以挂SQL解析器或直接用MySQL的EXPLAIN判断是否全表扫描。校验失败时，错误信息要原样返回给模型，它才能在下一轮修正自己的SQL。这里的关键是：错误信息要具体，比如"字段city不存在，可选的字段有city_name"，模型才能自我纠正。
4.3 结果校验：用代码强制校验，而不是让模型自说自话
大模型生成的结论再流畅，也不能取代真实数据的验证。我习惯在Agent输出最终结论前，先让结果过一遍校验器。校验器的职责是检查"模型声称的数字"是否真的存在于SQL返回的数据里。
def validate_answer(answer: str, sql_result: list) -> bool: # 提取模型中出现的数字和SQL结果的关键指标做交叉验证 import re if len(sql_result) == 0 and "无数据" not in answer: return False numbers_in_answer = re.findall(r"\d+", answer) if not numbers_in_answer: return False # 简单的检查：结论里必须包含SQL结果中top1的实体名或数值 top_item = sql_result[0] if sql_result else {} for value in top_item.values(): if str(value) in answer: return True return False这只是最粗粒度的校验，但能拦住最常见的问题：模型拿到空查询结果后编故事。更强的做法是要求模型把结论输出成JSON，里面带上"evidence"字段，指向上一步SQL结果里的具体行号。校验器拿到这个JSON后去查那行数据是否匹配，不匹配就打回重试。
这一步很重要。智能体看起来在执行数据分析，但如果执行结果不可信，它做得再快也没有意义。校验器就是那个"不近人情的QA"，强制模型每一步都对得上数。
5. 复杂场景数据分析的避坑记录：5个让智能体翻车的真实问题
5.1 SQL幻觉：查了半天查了个寂寞
现象：智能体在分析时经常生成引用错误字段的SQL，比如把login_time写成device_login_time；或者查询返回空结果后，模型没有意识到是SQL写错了，反而编一个"该时间段无风险数据"的结论。第一次跑通demo时，十个问题里有四五个都在这里翻车。
原因：模型对表结构的理解停留在"猜"的层面，系统Prompt里的Schema描述太长，工具结果和报错信息又不够具体，模型没拿到线索去修正自己。
解决：我给每个Agent配了一个describe_table工具，并明确要求它在不确定字段名时先调用；同时在run_sql的报错信息里带上"错误字段、正确字段、错误类型"三段结构化文本。经过这两步，SQL幻觉比例能降到一个可控水平。用真实表结构做预校验也很有效，在SQL执行前先查information_schema，比对字段是否存在。
5.2 工具循环：智能体停不下来
现象：同一个Agent在同一张表上反复调用describe_table，或者重复执行同一条SQL。有一次排查时，Agent在8轮里3次调用了同一个工具，烧掉大量token，最后也没给出结论。
原因：工具反馈里没有"这一步已经做过"的记忆，模型不记得自己已经查过这张表。我最初没有设置最大轮数，导致它无限循环下去。
解决：一个兜底是设置max_steps，超时后终止并返回已完成的分析片段；另一个更有效的办法是在系统Prompt里写"如果你已经获取过某张表结构，直接使用，不要重复调用describe_table"。同时在工具结果里附带一个简短缓存摘要，比如"表device_login结构已在第1步查询过"，模型看到后通常会停止重复动作。
5.3 上下文窗口被数据撑爆
现象：Agent执行一条SQL后返回了上千行原始数据，模型在下一轮生成时开始答非所问，甚至忘记最初的问题。把工具返回的JSON直接全部塞进消息历史，是这里最常见的错误。
原因：上下文窗口有限，数据量一大，系统Prompt里的口径说明和任务目标就被挤掉了。大模型注意力被一大堆订单明细稀释，自然抓不住重点。
解决：在run_sql里强制limit 50，并且把结果截断到2000字符后再回传。如果确实需要看全局分布，不要回传明细，而是让Agent自己先做聚合统计，比如按城市group by后再取top 10。复杂的多表关联分析，我一般会拆成多个子任务，每个子任务只回传中间统计量。
5.4 实时风控的延迟要求与推理延迟的矛盾
现象：智能体数据分析在离线场景很好用，但一旦有同事想把它接到实时风控决策链路里，发现一次Agent分析要好几秒甚至几十秒，线上交易等不了。
原因：大模型推理本身就慢，多轮工具调用又放大了延迟。实时风控的预算通常在几十毫秒到几百毫秒级别，Agent这种"慢慢思考"的模式天生不合适。
解决：把Agent定位成离线排查、策略回放和事后的复杂场景归因工具，不要让它参与实时拦截。实时链路继续用规则和轻量模型，Agent负责在事后把可疑样本解释清楚。如果你想用这方向替代部分人工数据分析，那就聚焦在"给策略分析师提效"这件事上，而不是去改造线上决策链，一上来就瞄着实时场景做，基本没有成功可能。
5.5 数据权限和审计：智能体不是DBA
现象：Agent在开放权限下会生成一些很"野"的SQL，比如跨库join，或者不带任何过滤条件扫描超大表。有次同事发现一个Agent任务把分析库的CPU跑满了，查日志才知道它执行了一条SELECT * FROM all_logs。
原因：我给Agent的数据库账号权限太大，而且没有做事后审计。模型本身没有"这条SQL会让数据库承受多大压力"的意识。
解决：最小权限原则，账号只允许访问分析宽表；在run_sql里强制所有查询必须带时间范围和limit；同时在Agent执行层记录每一次工具调用的入参、返回结果hash、耗时和调用人工号，保留审计日志。曾经有一个排查需求需要跨部门数据，我没有放开权限，而是把涉及的数据先同步到风险分析库，再做脱敏和授权。智能体的定位是用来做分析，不是用来当DBA。
6. 从demo到可评估：用影子模式和回放验证智能体的分析质量
把Agent跑通只算完成了20%，剩下的是证明"它真的比人快、和人的结论一致"。我的习惯是先在离线环境搭一套影子模式：把过去三个月真实风控事件的问题和最终人工分析报告整理出来，构造一个评估集。每个case包含三样东西：原始问题、期望的分析步骤、最终的结论摘要。
验证时跑三个指标：一是SQL执行正确率，看Agent生成的SQL能不能在只读库上跑通并返回有效数据；二是结论一致率，让Agent的结论和当时人工报告的结论对比，看关键实体和指标是否对得上；三是平均耗时，记录从提出问题到输出结论用了多少轮、多少秒。由于历史case已经有标准答案，这一步做得越充分，后面上生产越有底气。
回放的价值在于把"好像能用"变成"确实能复用"。我当时用一个月的司机端风险case做回放，发现Agent对"设备复用"类问题的复现率很高，但对"关系图谱"类问题几乎全部跑偏，因为模型缺少图数据库工具。于是我在工具层加了一个简单的共现查询接口，把同设备、同IP、同支付方式的关联搜索做成工具暴露给Agent，复现率才明显提上来。这件事如果不做回放，直接上Web系统给分析师用，第一周就会被一堆badcase淹死。
我的教训是：不要急着宣传"大模型已经替代了数据分析"，先把评估集建好，把失败case归类，再决定是调Prompt、加工具还是做微调。一套智能体数据分析方案的成熟度，不是看demo演示多流畅，而是看它在历史回放中的坏case能不能收敛。希望帮到你。
本文还有配套的精品资源，点击获取
