聊《一个数据分析项目改成 AI 流程后最难的部分完全变了》之前先说一句实在的别急着背概念先看它在真实项目里到底解决什么问题。摘要上周带新人review代码看到一个用LangGraph写的智能分析AgentQuery输入就能自动跑SQL出结果界面还挺好看。新人问这个能上线吗我点了两下发现数据库密码写在环境变量里没做权限隔离查询日志全打到标准输出没做分级工具调用没有熔断和重试。我说这个Demo要接生产得重做一半。这句话让我想到很多从数据分析转大模型的同学。你们的SQL写得比谁都溜指标口径门儿清现在学LangChain、学Agent框架Demo跑通那天觉得自己已经能做了。但真到了要接进现有系统、要让团队接手维护的时候才发现最难的部分根本不是模型调用。目录真实案例一个销售分析Agent的演进从Demo到生产权限改造排查过程日志系统的重构数据工具调用从单步到工作流失败原因三种错误的区分适用边界什么时候不该做Agent总结真实案例一个销售分析Agent的演进事情发生在我现在所在的电商平台数据组。去年Q3我们想做一个销售分析助手让运营同学能用自然语言问数据问题不需要每次都提需求给分析师。初始Demo两周完成输入运营问上周华东区销售额TOP5商品流程LLM → 翻译成SQL → 查ClickHouse → 返回结果输出一个表格Demo跑通那天团队都很兴奋。但当我们准备接入生产环境时问题一个个冒出来。第一个问题权限。这个Agent直接连的是生产ClickHouse运营问的问题如果SQL写得不对可能把整张表的数据拉出来。我们给数据库做了只读账号但没做行级权限——一个运营可以问出所有用户的信息。第二个问题日志。LLM的输入输出、SQL执行情况、工具调用链全部打到了stdout。生产环境日志量暴涨排查问题的时候根本找不到关键信息。第三个问题可观测。Agent调用工具失败时没有重试策略LLM抽风生成非法SQL时没有校验数据库慢查询时整个请求卡死没有超时控制。我们花了三个月重做最终形成了一个可维护的生产级Agent。下面按改造顺序拆解。从Demo到生产权限改造权限问题是数据分析转大模型最容易忽视的部分。你们习惯了直接操作数据库权限意识是在传统BI工具上养成的——看什么表、查什么字段都有审批流程。但Agent不一样它是动态生成SQL的权限边界变得模糊。我们的做法三层权限控制。第一层数据源隔离。给Agent单独创建数据库账号只授予特定库的SELECT权限其他库完全不可见。这步简单但很多团队没做。第二层行级权限。这是最容易被忽略的。我们用RLSRow-Level Security策略根据用户身份过滤数据。运营A只能看华东区的数据运营B只能看华南区。Agent生成的SQL会自动带上这些过滤条件不需要LLM去理解这些业务规则。# 权限验证中间件示例 from functools import wraps from typing import Callable, Any def require_data_permission(allowed_regions: list[str]): 装饰器验证用户是否有权限访问指定区域数据 def decorator(func: Callable) - Callable: wraps(func) def wrapper(user_id: str, region: str, *args, **kwargs) - Any: if region not in allowed_regions: raise PermissionError( f用户 {user_id} 无权访问区域 {region} 的数据 ) return func(user_id, region, *args, **kwargs) return wrapper return decorator class SQLSanitizer: SQL注入防护拦截危险操作 DANGEROUS_KEYWORDS {DROP, DELETE, TRUNCATE, ALTER, INSERT, UPDATE} def sanitize(self, sql: str) - str: upper_sql sql.upper() for keyword in self.DANGEROUS_KEYWORDS: if f {keyword} in upper_sql: raise SecurityError(f检测到危险SQL操作: {keyword}) return sql第三层查询结果脱敏。对于包含用户隐私字段手机号、身份证的查询我们做了自动脱敏——即使权限通过了返回的数据也会把敏感字段mask掉。排查过程日志系统的重构Demo阶段的日志是散乱的我们生产环境第一次排查问题时花了两个小时才定位到一个慢查询的来源。问题出在所有日志都打到stdout没有结构化没有分级没有trace ID。我们的排查链路现象某次促销活动期间Agent响应时间从2秒暴涨到30秒运营反馈系统卡死了。验证动作1. 首先查数据库层面——监控显示ClickHouse有慢查询但具体是哪条SQL不清楚2. 然后查Agent日志——stdout里几千条日志混在一起找不到对应请求3. 最后加trace ID重新排查——每个请求打上唯一ID才能把LLM调用、SQL执行、结果返回串起来排除结果问题不是Agent本身而是LLM生成的SQL没有加LIMIT查了一张千万级表的全量数据。加上LIMIT 1000后恢复正常。# 结构化日志配置 import logging import json from datetime import datetime from typing import Optional class StructuredLogger: 生产级结构化日志每个请求有唯一trace ID def __init__(self, service_name: str): self.logger logging.getLogger(service_name) self.logger.setLevel(logging.INFO) # 生产环境用JSON格式方便ELK采集 formatter logging.Formatter( {timestamp:%(asctime)s, level:%(levelname)s, trace_id:%(trace_id)s, user_id:%(user_id)s, message:%(message)s} ) handler logging.StreamHandler() handler.setFormatter(formatter) self.logger.addHandler(handler) def log_request(self, trace_id: str, user_id: str, action: str, **kwargs): 记录请求日志自动注入trace_id和user_id extra { trace_id: trace_id, user_id: user_id, } self.logger.info( f[{action}] {json.dumps(kwargs, ensure_asciiFalse)}, extraextra )关键改造点引入trace ID贯穿整个请求链路。LLM调用、SQL执行、工具返回每个环节都带上同一个trace ID。这样在日志系统里查一条trace ID就能看到完整的调用链。数据工具调用从单步到工作流Demo阶段的工具调用很简单LLM生成SQL → 执行 → 返回。生产环境需要更复杂的工作流。我们的Agent核心逻辑from langchain_core.tools import tool from langchain_core.prompts import ChatPromptTemplate from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated import operator class AgentState(TypedDict): messages: list current_query: str sql_result: dict has_permission: bool trace_id: str # 工具1权限验证 tool def check_permission(user_id: str, query: str) - bool: 检查用户是否有权限执行此查询 # 实际项目中这里会查权限系统 return True # 工具2SQL生成与校验 tool def generate_and_validate_sql(question: str, schema: dict) - str: 生成SQL并进行安全校验 # LLM生成SQL sql call_llm_for_sql(question, schema) # 安全校验 sanitizer SQLSanitizer() validated_sql sanitizer.sanitize(sql) return validated_sql # 工具3执行查询 tool def execute_query(sql: str, trace_id: str) - dict: 执行查询并返回结果 logger StructuredLogger(agent_service) logger.log_request(trace_id, system, execute_query, sqlsql[:100]) result db.execute(sql) logger.log_request(trace_id, system, query_completed, rowslen(result), duration_ms120) return result # 构建工作流 workflow StateGraph(AgentState) # 节点定义 workflow.add_node(validate_permission, validate_permission_node) workflow.add_node(generate_sql, generate_sql_node) workflow.add_node(execute_query, execute_query_node) workflow.add_node(format_response, format_response_node) # 边定义 workflow.add_edge(validate_permission, generate_sql) workflow.add_conditional_edges( generate_sql, lambda state: execute_query if state[has_permission] else deny, {execute_query: execute_query, deny: END} ) workflow.add_edge(execute_query, format_response) workflow.add_edge(format_response, END) app workflow.compile()代码解释AgentState定义了Agent的完整状态包括消息、当前查询、SQL结果、权限状态和trace ID。这是LangGraph工作流的核心每个节点都会读取和更新这个状态。check_permission权限验证工具。生产环境中这里会对接真实的权限系统检查用户是否有权限访问相关数据表。generate_and_validate_sqlSQL生成工具。关键点是生成了SQL之后要做安全校验拦截危险操作。execute_query查询执行工具。这里做了日志记录每个查询都会打trace_id方便后续排查。工作流边权限不通过时直接拒绝不执行查询。这是一个关键的分叉点很多Demo里没有这个判断。失败原因三种错误的区分做Agent项目失败原因大致分三类新手经常混淆。业务错误LLM生成的SQL逻辑不对或者查询结果不符合业务预期。比如问上周销售额LLM理解成了最近7天而不是上个自然周。这类错误通常是因为指标口径没有明确传给LLM或者few-shot示例不够。配置错误环境变量没配对、数据库连接串写错、模型API key过期。这类错误最容易排查——看日志里的报错信息就能定位。但Demo阶段经常用硬编码绕过了上线就爆。环境错误网络超时、数据库连接池满、LLM服务限流。这类错误需要熔断和重试机制。我们的做法是每个工具调用都有超时控制默认10秒失败后重试3次仍然失败则返回友好错误信息。区分这三类错误的方法很简单业务错误看结果对不对配置错误看启动日志环境错误看超时和限流日志。适用边界什么时候不该做Agent最后说点反向的。不是所有数据分析需求都适合做成Agent。适合的场景问题类型多样无法用固定报表覆盖用户有一定的数据素养能提出相对清晰的问题数据权限边界清晰可以做行级控制不适合的场景查询模式固定90%的问题都是同一张表的固定维度——直接用BI报表更便宜数据敏感度高无法做细粒度权限控制——Agent的动态SQL会带来权限风险用户数据素养低问的问题经常模糊不清——Agent会频繁生成错误SQL体验反而比人工差我们项目里大约60%的需求适合Agent剩下40%用传统报表人工分析覆盖。这个比例是根据实际使用情况调整的初期Agent做得太激进结果运营同学问的问题太随意LLM经常答非所问最后回退到混合模式。总结从数据分析转大模型Demo跑通只是入门。真正考验工程能力的是权限控制、日志追踪、可观测性这些 boring 的部分。你们的SQL写得再好如果Agent在生产环境里权限失控、日志混乱、排查困难这个项目就不会被团队接受。学习顺序建议先掌握Agent框架的基本用法然后重点补权限设计和日志系统的知识。简历上不要只写做了个智能分析Agent要写设计了行级权限控制方案接入trace ID日志系统支持生产环境日均1000次查询。后者才是面试官想听的。Demo能跑通证明你理解了原理权限和日志做扎实了才证明你能做项目。资料展示下面是我整理的AI大模型学习资料和工具包预览适合收藏后按主题逐步学习。需要这份AI大模型资料清单的话在评论区回复「清单」即可我会根据大家的问题继续补充对应的实战内容。