第1篇：桌面应用AI集成的“意图上云、数据留本地“安全架构设计

第1篇桌面应用AI集成的意图上云、数据留本地安全架构设计在政务信息化项目中档案数据往往涉及敏感信息。当大模型成为系统标配一个核心矛盾浮出水面既想让AI理解用户的复杂需求又绝不能让业务数据流入第三方云端。本文分享一套在PyQt6桌面应用中落地的意图上云、数据留本地架构用四层隔离机制彻底斩断数据外泄通道同时保留大模型的自然语言理解能力。一、问题的提出AI集成中的数据安全困局去年接到一个政务信息化项目档案管理系统的需求用户明确提出一个硬性约束系统中的项目信息、财务数据、责任主体等敏感内容绝不能通过大模型API传输到外部服务器。这个约束合情合理但实现起来却面临三重矛盾第一重矛盾理解力与隔离性的冲突。大模型的价值在于理解用户模糊、口语化的需求比如帮我看看预算超支的项目有哪些附件还没归档。如果不把业务数据传给模型它如何知道哪些项目超支、哪些附件未归档第二重矛盾交互自然度与系统严谨性的冲突。政务系统要求操作可追溯、结果可审计。如果让大模型直接生成SQL查询并执行一旦出现数据泄露或误操作责任边界完全模糊。第三重矛盾功能完整性与成本控制。系统需要支持查询、统计、图表生成、多步工作流等多种能力。如果每个功能都写死规则维护成本极高如果完全依赖大模型Token费用和数据风险又不可控。我们的解决方案是让大模型只处理意图所有业务数据的获取和加工全部在本地完成。大模型收到的是用户说了什么、需要做什么返回的是该调用哪个本地功能、传入什么参数。数据从不出本地AI却好像什么都能查。二、核心设计四层数据隔离机制整个架构可以用一句话概括大模型是大脑负责理解本地系统是手脚负责执行。二者之间传递的永远只有指令没有数据。具体实现上我们设计了四层隔离机制隔离层作用大模型能看到什么大模型看不到什么第一层System Prompt隔离只暴露能力清单不暴露数据结构Skill名称、参数定义、使用示例数据库表结构、字段含义、数据样本第二层消息内容隔离历史消息截断防止上下文累积最近20轮对话的摘要每轮最多500字符完整的查询结果、业务数据详情第三层执行过程隔离Skill在本地运行数据不入云JSON格式的调用指令数据库查询结果、文件内容、统计数值第四层答复生成隔离模板本地渲染大模型不接触结果无最终呈现给用户的数据表格、图表、文本四层隔离自上而下形成一道完整防线。接下来逐层拆解实现细节。三、第一层隔离System Prompt 的能力清单设计System Prompt 是大模型理解系统能力的唯一信息源。设计得当模型就能精准选择Skill设计不当要么频繁误调要么需要大量示例消耗Token。我们的 System Prompt 分为五个模块角色定义、技能摘要、工作流规则、输出格式要求、字段映射表。3.1 技能摘要的动态生成系统中每个Skill都继承自BaseSkill基类通过get_description_for_llm()方法自动生成供大模型阅读的能力描述classBaseSkill(ABC):abstractmethoddefdefine(self)-SkillDefinition:定义 Skill 的元数据名称、描述、参数等passdefget_description_for_llm(self)-str:dself.definition descf【{d.display_name}】({d.name})\ndescf 功能:{d.description}\ndescf 分类:{d.category.value}\ndescf 参数:\n{d.get_params_schema()}\nifd.example_query:descf 用户提问示例:{d.example_query}\nreturndescSkillRegistry在初始化时遍历所有已注册的Skill拼接成完整的技能摘要注入System Prompt。关键点在于摘要中只包含Skill能做什么、需要什么参数完全不涉及数据库表名、字段类型、数据量等信息。3.2 字段映射表的强制约束这是最容易被忽视却最致命的一点。大模型擅长理解中文但本地数据库的字段名是英文。如果任由模型自由发挥用户说责任者为某某的附件模型可能输出{责任者: {contains: 某某}}导致本地过滤时找不到字段查询结果为空。我们在System Prompt中硬编码了完整的字段映射表并标注极其重要违反将导致查询无结果必须将用户提到的中文术语转换为对应的英文数据库字段名 - 项目名称 → project_name - 预算 → budget - 责任单位/责任者 → responsible_unit - ... 示例用户说责任者为张三的附件正确的filter条件为 {responsible_unit: {contains: 张三}} 错误写法绝对禁止{责任者: {contains: 张三}}这条规则从根本上杜绝了因字段名不匹配导致的空结果问题。后续在代码层面我们还会对模型返回的JSON做二次校验。3.3 输出格式的严格限定大模型必须按严格的JSON格式输出调用指令我们定义了五种标准动作action使用场景call_skill单个Skill即可完成的需求execute_workflow需要多步组合由大模型动态编排execute_predefined_workflow匹配预定义工作流Token消耗更低unknown超出Skill范围礼貌拒绝greeting打招呼或闲聊每种动作都有固定的JSON Schema大模型只需要填空不需要创造新格式。这大幅提升了输出稳定性也为后续的JSON解析降低了难度。四、第二层隔离消息内容的截断与脱敏对话历史是提升多轮交互体验的关键但也是数据泄露的高风险通道。如果直接把前几轮的查询结果作为上下文传给大模型敏感数据就随着Token一起上了云。我们的策略是**“传意图不传数据”**。def_build_messages(self,user_message:str)-List[Dict[str,str]]:messages[{role:system,content:system_prompt}]# 添加对话历史仅传简短的意图摘要不传完整业务数据formsginself.message_history[-self.max_history:]:contentmsg.get(content,)# 对历史消息进行截断避免token爆炸iflen(content)500:contentcontent[:500]...messages.append({role:msg[role],content:content})messages.append({role:user,content:user_message})returnmessages这里有两个关键控制点数量控制max_history默认保留最近20轮对话。政务场景下单次会话通常不会太长20轮足够覆盖绝大多数上下文需求。长度控制每轮消息最多500字符超出部分直接截断并追加…。这意味着即使上一轮查询返回了一个包含100条记录的表格传给大模型的也只是前500个字符的摘要。模型能感知到上一轮查过项目列表这个意图但看不到具体的项目信息。实际运行中500字符的摘要通常只包含表头或前两三行数据足以让模型理解上下文主题却不会泄露完整信息。五、第三层隔离Skill执行的本地闭环这是整个架构的核心。大模型输出的只是一串JSON指令真正的数据查询、计算、加工全部由本地Skill完成。5.1 从JSON到Skill调用的完整链路当用户输入查询预算小于200的项目的全部附件时大模型返回这样的JSON{action:execute_workflow,steps:[{action:call_skill,skill_name:get_all_projects,params:{},result_key:projects},{action:filter,source:${projects},condition:{budget:{lt:200}},result_key:filtered_projects},{action:loop,source:${filtered_projects},item_key:project,call:{skill_name:get_files_by_project,params:{project_id:${project.id}}},result_key:all_files,merge:true}],reply_hint:已为您查询预算小于200元的项目的全部附件,template_key:workflow_files_by_projects,template_data_keys:[filtered_projects,all_files]}DialogManager解析这个JSON后调用_execute_workflow()方法。工作流引擎按顺序执行每个步骤call_skill调用get_all_projects获取所有项目列表filter在本地内存中按budget 200过滤loop遍历过滤结果对每个项目调用get_files_by_project整个过程中大模型只知道要查预算小于200的项目的附件这个意图以及三个步骤的抽象描述。它不知道数据库里有多少个项目、不知道具体预算数值、不知道附件文件名更不会接触到附件的文件内容。5.2 Skill的线程安全设计SkillRegistry采用单例模式双重检查锁定确保多线程环境下安全classSkillRegistry:_instanceNone_lockthreading.Lock()def__new__(cls):ifcls._instanceisNone:withcls._lock:ifcls._instanceisNone:cls._instancesuper().__new__(cls)cls._instance._skills{}cls._instance._registry_lockthreading.Lock()returncls._instancedefcall_skill(self,name:str,**params)-Dict[str,Any]:# 获取skill在锁内完成withself._registry_lock:skillself._skills.get(name)# execute在锁外执行避免长时间持有锁ifskillisNone:return{success:False,error:fSkill {name} 未找到}returnskill.execute(**params)关键点在于call_skill方法获取Skill实例时加锁执行Skill时释放锁。这避免了复杂查询长时间占用锁导致的并发阻塞。5.3 七步工作流引擎对于多步组合需求我们实现了七种标准步骤类型步骤类型功能典型用法call_skill调用Skill获取数据查询所有项目filter条件过滤支持AND/OR/NOT嵌套筛选预算超支的项目sort多字段排序按预算降序、按编号升序aggregate聚合统计count/sum/avg/max/min统计各部门项目数量distinct去重获取所有不重复的状态值loop循环调用Skill遍历项目列表查附件extract提取指定字段只取项目名称列表步骤间通过${result_key}语法传递数据支持字段级引用如${project.id}。整个工作流在本地内存中执行不经过网络传输。为了防止恶意或错误的工作流导致无限循环引擎设置了MAX_LOOP_ITERATIONS 100的硬性上限。六、第四层隔离模板引擎的本地渲染Skill执行完成后需要将结果呈现给用户。如果让大模型生成答复文本就必须把查询结果传给模型第四层隔离就此破裂。我们的方案是本地模板引擎每个Skill在define()中声明自己的结果模板在execute()中返回模板数据和模板键名由TemplateEngine在本地填充渲染。classTemplateEngine:_templates:Dict[str,Any]{}classmethoddefrender(cls,key:str,data:Dict[str,Any])-str:render_funccls._templates.get(key)ifrender_funcisNone:# 安全兜底模板未找到时不输出原始datareturnf模板 {key} 未定义try:returnrender_func(data)exceptExceptionase:return模板渲染失败请稍后重试模板渲染函数是纯本地函数示例def_render_project_list(data:Dict[str,Any])-str:itemsdata.get(items,[])ifnotitems:returnp未找到符合条件的项目。/phtmltabletrth编号/thth名称/thth预算/th/trforiteminitems:htmlftrtd{_safe_str(item.get(project_code))}/tdhtmlftd{_safe_str(item.get(project_name))}/tdhtmlftd{_format_money(item.get(budget))}/td/trhtml/tablereturnhtml注意两个安全细节数据转义_safe_str()对用户提供的内容做HTML转义防止XSS注入失败兜底模板未找到或渲染失败时绝不回退到输出原始数据而是返回固定提示语模板引擎还内置了字典翻译功能。数据库中项目状态存的是ACCEPTANCE这样的英文枚举值模板渲染时自动翻译为验收翻译映射从本地数据库加载并缓存5分钟。大模型永远看不到这些翻译规则也看不到原始枚举值。七、完整交互流程一个多步查询的完整生命周期以用户输入查询预算小于200的项目的全部附件为例完整生命周期如下阶段一意图上云用户输入自然语言需求DialogManager._build_messages()组装消息System Prompt 截断后的历史消息 用户输入消息发送到云端大模型API大模型理解意图输出JSON调用指令只含Skill名和参数不含任何业务数据阶段二本地执行DialogManager._parse_llm_response()解析JSON支持四层容错直接解析、Markdown代码块提取、正则提取、嵌套JSON匹配识别为execute_workflow动作调用_execute_workflow()工作流引擎顺序执行三步查项目 - 过滤 - 循环查附件每个步骤调用本地SkillSkill通过Service层访问本地SQLite数据库所有查询结果存储在本地内存的context字典中阶段三数据留本地工作流执行完毕DialogManager从结果中提取template_key和template_dataTemplateEngine.render()在本地查找模板函数并填充数据生成HTML格式的表格答复包含项目信息和附件列表如果结果中包含chart_type额外生成本地图表Plotly渲染为HTML最终答复文本返回给UI层展示全程未向云端传输任何业务数据阶段四对话历史更新将用户查询预算小于200的项目的附件和系统返回了X条结果的摘要加入历史摘要长度控制在500字符以内为下一轮对话提供上下文八、安全防护的额外措施四层隔离是主体防线我们还增加了三道辅助安全措施。8.1 LLM调用费用控制政务系统的预算审计同样严格。LLMClient实现了月度费用上限控制classLLMClient:def__init__(self):self.monthly_cost0.0self._cost_monthdatetime.now().strftime(%Y-%m)self._load_cost_from_file()# 启动时加载本月已用费用defchat(self,messages,**kwargs):self._check_month_reset()# 跨月自动重置ifself.monthly_costMAX_MONTH_COST:return{content:本月大模型调用费用已达上限请联系管理员。}# ... 调用API并计算费用 ...self._save_cost_to_file()# 持久化到本地文件费用记录保存在本地JSON文件中每月自动重置。当费用达到上限时系统直接拒绝调用不会向云端发送请求。8.2 JSON解析的四层容错大模型不总是完美遵守JSON格式要求。DialogManager实现了四层容错解析def_parse_llm_response(self,response:str)-Dict[str,Any]:# 第一层直接解析try:returnjson.loads(response)exceptjson.JSONDecodeError:pass# 第二层从markdown代码块提取json_matchre.search(r(?:json)?\s*(\{.*?\})\s*,response,re.DOTALL)ifjson_match:try:returnjson.loads(json_match.group(1))exceptjson.JSONDecodeError:pass# 第三层从文本中提取第一个JSON对象json_matchre.search(r\{[^{}]*\},response,re.DOTALL)ifjson_match:try:returnjson.loads(json_match.group(0))exceptjson.JSONDecodeError:pass# 第四层尝试提取更复杂的嵌套JSON基于大括号计数brace_count0start_idx-1fori,charinenumerate(response):ifchar{:ifbrace_count0:start_idxi brace_count1elifchar}:brace_count-1ifbrace_count0andstart_idx0:try:returnjson.loads(response[start_idx:i1])exceptjson.JSONDecodeError:start_idx-1continue# 全部失败返回unknownreturn{action:unknown,reply_hint:我理解了您的需求但当前无法处理。}四层容错覆盖了绝大多数模型输出异常的情况确保系统不会因为格式问题而直接崩溃。8.3 未知请求的兜底拒绝System Prompt中明确规定“超出Skill能力的请求返回actionunknown”。同时我们在代码层也做了兜底UNKNOWN_REPLY(很抱歉我目前无法处理该类问题只能协助您进行项目档案管理相关操作包括项目/分卷/附件的查询和管理、项目周期管理、用户/部门/角色/字典的管理、操作日志查询与清理、系统配置查看、数据统计分析和图表生成等。)当大模型返回actionunknown或解析失败时系统返回这段预设文本明确告知用户能力边界。这比让模型自由发挥、可能编造出不存在的功能要安全得多。九、踩过的坑与解决方案9.1 大模型使用中文字段名导致查询为空早期版本没有强制字段映射规则大模型经常输出{责任者: {contains: 某某}}这样的条件。本地过滤时找不到责任者这个字段结果永远为空。解决方案在System Prompt中硬编码完整映射表并在提示词中标注极其重要违反将导致查询无结果。同时准备在引擎层增加字段名别名容错机制自动将常见中文别名映射到英文字段名。9.2 历史消息中的数据残留第一版实现中对话历史直接存储了完整的系统答复文本包含HTML表格。当历史消息累积后敏感数据随着上下文反复上传。解决方案引入500字符截断机制并考虑在存储历史时仅保留回答了什么问题的摘要而非完整答复内容。9.3 工作流步骤间类型不一致大模型输出的参数通常是字符串类型但Skill执行时需要int、float或dict。例如project_id应该是整数模型却输出字符串1。解决方案在_execute_skill_call()中增加参数类型自动转换根据Skill定义中的param_type将字符串转为目标类型def_convert_params(self,skill_name:str,params:Dict[str,Any])-Dict[str,Any]:skillregistry.get_skill(skill_name)ifnotskill:returnparams definitionskill.definitionforparam_defindefinition.params:ifparam_def.nameinparams:valueparams[param_def.name]ifparam_def.param_typeint:params[param_def.name]int(value)elifparam_def.param_typefloat:params[param_def.name]float(value)elifparam_def.param_typedictandisinstance(value,str):params[param_def.name]json.loads(value)returnparams9.4 图表渲染导致界面卡死使用Plotly生成图表时to_html(full_htmlFalse)默认内嵌整个Plotly.js库约4.7MB导致QWebEngineView解析超时显示空白。解决方案设置include_plotlyjsFalse在HTML中通过本地文件引用Plotly.js同时改用setUrl()加载临时HTML文件而不是setHtml()直接注入安全沙箱限制会阻止file://协议的script引用。十、总结意图上云、数据留本地不是一句空洞的口号而是一套可落地、可审计、可量化的工程实践。其核心在于明确划分大模型和本地系统的职责边界大模型负责理解自然语言、选择合适Skill、提取参数、编排多步工作流本地系统负责执行数据库查询、加工业务数据、渲染答复模板、控制访问权限四层隔离机制System Prompt隔离、消息内容隔离、执行过程隔离、答复生成隔离确保业务数据在任何环节都不会流入云端。辅助措施费用控制、JSON容错、兜底拒绝进一步提升了系统的鲁棒性和安全性。这套架构已经在政务信息化项目档案管理系统中完整落地支持Windows和银河麒麟V10双平台运行。从实际使用效果来看用户可以用自然语言完成90%以上的查询和统计需求而敏感数据始终停留在本地SQLite数据库中满足政务场景对数据安全的严格要求。如果你的桌面应用也在探索AI集成又面临数据合规的压力不妨参考这个意图与数据分离的思路。大模型的理解能力和本地系统的执行能力各司其职才是真正的安全之道。