构建基于本地大模型的隐私优先 AI Agent Harness Engineering
构建基于本地大模型的隐私优先 AI Agent Harness Engineering
引言
痛点引入:云端Agent的隐私与信任危机
你有没有过这样的经历?为了让AI帮你整理孩子的成长日记、分析家庭理财的Excel表格、生成一封包含敏感公司内部业务信息的周报草稿,你只能把这些核心隐私数据(甚至包括身份证号扫描件OCR后的文本、客户联系方式的截图描述、项目未公开的预算数据)一股脑儿上传到OpenAI、Claude、Gemini这些第三方云服务平台?
或许你安慰自己:“这些大厂有隐私政策,不会泄露我的数据吧?”但现实却一次又一次给我们敲响警钟:
- 数据泄露事件频发:2023年ChatGPT出现“历史会话泄露漏洞”,数千名用户的对话、API密钥被他人访问;同年某国产大模型平台被曝存储用户上传的医疗影像、法律合同等敏感数据长达180天以上,且无主动删除机制;
- 数据主权与二次利用隐患:大多数云服务商的隐私政策明确写着“为了改进服务质量,可能会对用户上传的非匿名化数据进行分析、训练”——你的家庭隐私、商业机密,很可能成为大模型迭代的“燃料”;
- 合规风险:欧盟《通用数据保护条例(GDPR)》、美国《加州消费者隐私法案(CCPA)》、中国《个人信息保护法(PIPL)》都对“个人敏感信息的跨境传输”“非必要数据收集”做出了严格规定,盲目使用云端Agent处理隐私数据,企业可能面临最高年营收4%或2000万欧元(GDPR) 的巨额罚款,个人也可能遭遇隐私被滥用的法律纠纷;
- 网络依赖与成本不可控:云端Agent必须联网才能使用,一旦断网就完全“瘫痪”;同时,大模型API调用的成本随使用量呈指数级增长——假设你的企业每天使用GPT-4o处理1000条包含隐私数据的业务请求,每月的API费用可能高达数万元甚至数十万元。
解决方案概述:隐私优先的本地大模型AI Agent Harness
那有没有办法既能享受AI Agent带来的便利,又能100%保护隐私数据不离开本地设备?答案是肯定的——构建一套基于本地大模型(Local Large Language Model, LLLM)的隐私优先AI Agent Harness(线束工程/集成框架)。
首先,我们得明确两个核心定义:
- 本地大模型(LLLM):指完全部署在本地硬件(PC、服务器、边缘设备如Jetson AGX Orin、Mac Studio等)上的大语言模型,所有推理、训练、数据处理都在本地完成,不产生任何网络流量(除非你主动配置联网工具,但数据本身仍不会上传到第三方),例如Meta的Llama 3 8B/70B、阿里云的Qwen 2.5 7B/14B、智谱AI的GLM-4 9B Chat、Anthropic的Claude 3 Haiku(通过Ollama本地部署)等;
- 隐私优先的AI Agent Harness:这里的“Harness”不是“ harnessing power”(利用力量)的抽象动词,而是软件工程中“线束集成系统”的具象比喻——就像汽车的线束把发动机、电池、车灯、传感器等所有零件连接成一个完整的、能协同工作的系统一样,AI Agent Harness是一套轻量、可定制、模块化的集成框架,它把本地大模型、工具链、记忆系统、向量检索模块、隐私合规引擎等“AI零件”安全、高效地组装成一个完整的、隐私优先的AI Agent,用户可以像拼乐高一样根据自己的需求添加或删除模块,不需要完全依赖LangChain、LlamaIndex这些臃肿的第三方框架,同时也能避免现有框架在隐私保护方面的不足。
这套Harness的核心优势可以概括为“4个100%,2个可控”:
- 4个100%:
- 100%数据隐私:所有隐私数据(包括用户输入、本地文件内容、工具执行结果、记忆系统存储的信息)完全在本地设备的内存或本地存储介质中流转、存储,不产生任何非必要的网络流量;
- 100%数据主权:你拥有所有数据的完全控制权,可以随时、彻底地删除任何数据,没有任何第三方能够访问或使用;
- 100%网络离线可用:除了可选的联网工具(如天气查询、股票查询,但查询参数本身是匿名化的,且只发送到公开的API接口,不包含隐私数据),整个Agent可以完全离线运行;
- 100%可定制:所有模块都是开源(或你可以自己实现)的,你可以根据自己的需求修改工具链、记忆系统、指令对齐逻辑、隐私合规规则等;
- 2个可控:
- 成本可控:只需要一次性购买本地硬件(或者甚至不需要——现在很多消费级PC如MacBook Pro M3 Max、RTX 4090台式机都能流畅运行Llama 3 70B 4-bit量化版、Qwen 2.5 14B 4-bit量化版),不需要支付任何API调用费用;
- 性能可控:你可以根据本地硬件的性能调整大模型的量化精度、工具调用的并发数、记忆系统的缓存大小等,从而获得最佳的性能体验。
最终效果展示
在开始讲解Harness的构建之前,我们先来看一下最终实现的本地隐私优先笔记助手Agent的效果:
- 场景1:整理本地Markdown成长日记:用户在Agent界面输入“帮我整理一下今年1-6月的孩子成长日记,重点记录身高、体重、第一次说话、第一次走路的时间和内容”,Agent会自动从本地存储的
/Users/xxx/Documents/GrowthDiary/目录下读取所有2024年1-6月的Markdown文件,然后通过本地大模型整理成一份结构化的Word文档(也可以是Markdown、PDF),整个过程没有任何网络流量(你可以在Mac的活动监视器、Windows的任务管理器中查看网络使用情况); - 场景2:分析本地Excel家庭理财表格:用户输入“帮我分析一下2024年的家庭理财Excel表格,统计一下每个月的收入、支出、结余,以及支出最多的前5个类别”,Agent会自动读取本地的
FamilyFinance_2024.xlsx文件,通过本地的Pandas工具进行数据分析,然后通过本地大模型生成一份可视化的分析报告(包含柱状图、饼图),整个过程Excel文件完全在本地处理; - 场景3:查询本地SQLite业务数据并生成周报草稿:假设用户是一家小型电商公司的运营,本地有一个SQLite数据库存储着2024年的订单数据、用户数据,用户输入“帮我查询一下2024年第30周的订单数据,统计一下订单量、GMV、客单价、复购率,然后生成一封包含这些数据的周报草稿,不要包含任何客户的真实姓名和手机号”,Agent会自动连接本地的SQLite数据库,进行匿名化查询(只查询统计数据,不查询个人敏感信息),然后通过本地大模型生成周报草稿,整个过程数据库完全在本地访问,且隐私合规引擎会自动过滤掉个人敏感信息。
准备工作
环境/工具
在开始构建Harness之前,我们需要准备以下环境和工具:
- 硬件环境:
- 最低配置(适合运行Llama 3 8B 4-bit量化版、Qwen 2.5 7B 4-bit量化版,推理速度约为10-20 tokens/s):
- CPU:Intel Core i7-12700H/AMD Ryzen 7 7735H或以上;
- 内存:32GB DDR4/DDR5或以上;
- 存储:512GB NVMe SSD或以上(用于存储本地大模型、本地文件、记忆系统数据);
- GPU(可选但强烈推荐):NVIDIA RTX 3060 12GB/RTX 4060 Ti 16GB/AMD Radeon RX 7900 XT 20GB或以上(用于加速大模型推理,GPU推理速度约为50-100 tokens/s);
- 推荐配置(适合运行Llama 3 70B 4-bit量化版、Qwen 2.5 14B 8-bit量化版,推理速度约为30-60 tokens/s):
- CPU:Intel Core i9-14900K/AMD Ryzen 9 7950X或以上;
- 内存:64GB DDR5或以上;
- 存储:1TB NVMe SSD或以上;
- GPU:NVIDIA RTX 4090 24GB/RTX A6000 48GB/MacBook Pro M3 Max 128GB或以上;
- 边缘设备配置(适合部署在小型服务器、边缘设备如Jetson AGX Orin 64GB上,运行Llama 3 8B 4-bit量化版,推理速度约为5-10 tokens/s):
- Jetson AGX Orin 64GB;
- 存储:512GB NVMe SSD或以上;
- 最低配置(适合运行Llama 3 8B 4-bit量化版、Qwen 2.5 7B 4-bit量化版,推理速度约为10-20 tokens/s):
- 软件环境:
- 操作系统:macOS 13.0+(Ventura/Sonoma,推荐,因为M系列芯片的GPU加速效果最好)、Ubuntu 22.04+、Windows 11+;
- 编程语言:Python 3.10+(推荐3.11或3.12,因为这两个版本的Python性能更好);
- 本地大模型部署工具:Ollama(推荐,因为它是目前最流行、最易用的本地大模型部署工具,支持一键下载、部署、运行几乎所有主流的开源大模型)、vLLM(适合高并发场景,性能更好,但配置稍微复杂)、llama.cpp(适合无GPU的环境,纯CPU推理,性能较差但兼容性最好);
- 本地向量库:ChromaDB(推荐,因为它是开源的、轻量的、易用的本地向量库,支持多种嵌入模型)、FAISS(Meta开发的,性能更好,但需要手动配置)、Milvus Lite(Milvus的本地版,适合大规模向量数据);
- 本地结构化记忆库:SQLite(推荐,因为它是Python内置的、轻量的、无需安装的关系型数据库,适合存储结构化的记忆数据如用户会话历史、工具执行记录、对齐约束等);
- 工具链库:Pydantic(用于定义工具的输入输出格式,推荐Pydantic v2,因为性能更好)、Pandas(用于处理Excel/CSV文件)、python-docx(用于生成Word文档)、matplotlib/seaborn(用于生成可视化图表)、sqlalchemy(用于连接本地SQLite/MySQL/PostgreSQL数据库);
- 界面库(可选):Streamlit(用于快速构建Web界面,推荐,因为它是目前最流行的Python Web界面库,不需要前端知识)、Gradio(适合快速原型开发)、PyQt(适合构建桌面应用)。
基础知识
在开始构建Harness之前,我们需要读者具备以下前置知识:
- Python基础编程:熟悉Python的变量、函数、类、模块、异常处理等基础语法;
- 大模型基础概念:了解大语言模型的基本工作原理(如Transformer架构、自注意力机制、Tokenization、推理过程等),了解Prompt Engineering的基本技巧(如Few-Shot Learning、Chain-of-Thought、Role Prompting等);
- AI Agent基础概念:了解AI Agent的基本组成(如LLM核心、工具链、记忆系统、向量检索模块等),了解工具调用的基本流程;
- 隐私保护基础概念:了解《个人信息保护法(PIPL)》《通用数据保护条例(GDPR)》的基本要求,了解敏感信息的分类、匿名化、脱敏的基本方法;
- SQL基础语法:熟悉SQL的基本查询语句(如SELECT、FROM、WHERE、GROUP BY、ORDER BY等),因为我们会使用SQLite存储结构化的记忆数据;
- Git基础操作(可选):如果读者想使用我们提供的开源代码,需要熟悉Git的基本操作(如clone、pull、push等)。
如果读者不具备以上前置知识,可以参考以下学习资源:
- Python基础编程:廖雪峰的Python教程、Python官方文档;
- 大模型基础概念:李沐老师的《动手学深度学习》大模型篇、Andrej Karpathy的《Let’s build GPT: from scratch, in code, spelled out.》、Prompt Engineering Guide;
- AI Agent基础概念:LangChain官方文档、LlamaIndex官方文档、OpenAI的GPT-4o Agent介绍;
- 隐私保护基础概念:《个人信息保护法(PIPL)》官方解读、《通用数据保护条例(GDPR)》中文译本、数据匿名化与脱敏技术指南;
- SQL基础语法:廖雪峰的SQL教程、SQLite官方文档;
- Git基础操作:廖雪峰的Git教程。
核心步骤
步骤1:定义隐私优先AI Agent Harness的核心架构
在开始编写代码之前,我们首先需要明确Harness的核心架构——只有架构清晰,后续的代码实现才能高效、可维护、可扩展。
1.1 现有AI Agent框架的隐私保护不足
在设计我们自己的Harness之前,我们先来看一下现有主流AI Agent框架(如LangChain、LlamaIndex)在隐私保护方面的不足:
- LangChain:
- LangChain默认会调用OpenAI、Anthropic等第三方云服务的嵌入模型和大模型,即使你配置了本地大模型,LangChain的某些模块(如LangSmith追踪、LangChain Hub Prompt模板)仍然会产生网络流量,可能会泄露你的Prompt模板和部分会话数据;
- LangChain的工具调用模块默认没有沙箱隔离机制,如果你的Agent调用了恶意工具(或者用户通过Prompt Injection让Agent调用了恶意工具),可能会破坏你的本地设备;
- LangChain的记忆系统默认没有隐私合规机制,可能会存储个人敏感信息;
- LlamaIndex:
- LlamaIndex同样默认会调用第三方云服务的嵌入模型和大模型;
- LlamaIndex的向量索引模块默认没有隐私保护机制,可能会将隐私数据的向量表示存储在第三方云服务的向量库中;
- LlamaIndex的工具调用模块也默认没有沙箱隔离机制。
正是因为现有框架在隐私保护方面的这些不足,我们才需要构建一套完全自主可控、隐私优先、轻量可定制的Harness。
1.2 隐私优先AI Agent Harness的核心架构设计
我们的Harness采用分层模块化架构,从上到下分为用户交互层、指令解析与对齐层、核心业务层、本地资源层四个层次,每个层次之间通过清晰的接口进行通信,保证了架构的解耦性和可扩展性。
同时,我们在每个层次都加入了隐私合规引擎,确保所有数据的流转、存储、处理都符合隐私保护的要求。
我们可以用下面的Mermaid架构图来清晰地展示Harness的核心架构:
1.3 隐私优先AI Agent Harness的核心模块组成
从上面的架构图中,我们可以看到Harness的核心模块组成,接下来我们逐个介绍这些模块的功能:
- 用户交互层:
- Web/桌面界面:负责与用户进行交互,接收用户的输入,展示Agent的输出;
- 输入预处理模块:负责对用户的输入进行预处理,例如去除多余的空格、换行符,转换为统一的编码格式等;
- 输出后处理模块:负责对Agent的输出进行后处理,例如格式化输出、生成可视化图表、转换为不同的文件格式等;
- 指令解析与对齐层:
- 意图识别模块:负责识别用户的意图,例如“整理本地Markdown文件”“分析本地Excel文件”“查询本地数据库”“生成周报草稿”等;
- 安全过滤模块(防Prompt Injection):负责过滤掉用户输入中的恶意Prompt,防止用户通过Prompt Injection控制Agent调用恶意工具、破坏本地设备或泄露隐私数据;
- 对齐约束检查模块:负责检查Agent的输出是否符合用户预设的对齐约束,例如“不要包含任何客户的真实姓名和手机号”“不要修改本地的系统文件”“不要访问任何第三方云服务”等;
- 对齐约束记忆(SQLite):负责存储用户预设的对齐约束;
- 核心业务层:
- 本地大模型抽象层:负责封装不同的本地大模型部署工具(如Ollama、vLLM、llama.cpp)的API,提供统一的接口供上层调用,这样上层不需要关心底层使用的是哪种部署工具,可以灵活切换;
- 工具管理模块:负责管理所有的工具,包括工具的注册、调用、参数验证等;
- 工具沙箱隔离模块:负责在沙箱环境中执行工具,防止工具(尤其是用户自定义的工具)破坏本地设备;
- 分层记忆系统:负责存储Agent的记忆数据,分为短期记忆(Session Memory,存储当前会话的历史信息,存储在内存中,会话结束后自动清除)、中期记忆(Short-Term Memory,存储最近7天的会话历史信息,存储在SQLite中,7天后自动清除)、长期记忆(Long-Term Memory,存储重要的、需要长期保留的信息,存储在SQLite中,需要用户手动删除)三个层次;
- 向量检索模块:负责将本地文件(如Markdown、PDF、Word、Excel)转换为向量表示,存储在本地向量库中,然后根据用户的输入检索相关的本地文件内容,提供给本地大模型作为上下文;
- 本地资源层:
- 本地大模型(Ollama/vLLM/llama.cpp):负责进行推理、生成文本;
- 本地嵌入模型(Ollama/Sentence-Transformers):负责将文本转换为向量表示;
- 本地向量库(ChromaDB/FAISS):负责存储向量表示,提供向量检索功能;
- 本地结构化记忆库(SQLite):负责存储对齐约束、中期记忆、长期记忆、审计日志等结构化数据;
- 本地文件系统:负责存储本地文件(如Markdown、PDF、Word、Excel);
- 本地数据库:负责存储本地业务数据(如电商订单数据、用户数据);
- 隐私合规引擎(贯穿所有层次):
- 敏感信息检测模块:负责检测所有数据中的敏感信息,例如身份证号、手机号、姓名、邮箱、银行卡号、公司内部业务信息等;
- 匿名化/脱敏模块:负责对检测到的敏感信息进行匿名化或脱敏处理,例如将手机号“13812345678”脱敏为“138****5678”,将姓名“张三”匿名化为“用户A”;
- 数据访问控制模块:负责控制Agent对本地资源的访问权限,例如“只允许访问
/Users/xxx/Documents/GrowthDiary/目录下的文件”“只允许查询本地SQLite数据库中的统计数据,不允许查询个人敏感信息”; - 数据留存管理模块:负责管理数据的留存时间,例如“短期记忆会话结束后自动清除”“中期记忆7天后自动清除”“长期记忆需要用户手动删除”;
- 审计日志模块:负责记录Agent的所有操作,例如“用户输入了什么”“Agent调用了什么工具”“工具执行了什么操作”“Agent访问了什么本地资源”“Agent输出了什么”等,审计日志可以用于排查问题、追溯责任。
步骤2:实现隐私优先AI Agent Harness的核心模块
接下来,我们开始逐一实现Harness的核心模块。为了保证代码的简洁性和可读性,我们会使用Python 3.12编写代码,并使用Pydantic v2、Ollama、ChromaDB、SQLite、Streamlit等主流工具。
首先,我们需要创建一个项目目录结构,如下所示:
local-privacy-agent-harness/
├── .env.example # 环境变量示例文件
├── requirements.txt # 依赖库列表
├── README.md # 项目说明文档
├── src/ # 源代码目录
│ ├── __init__.py
│ ├── config/ # 配置文件目录
│ │ ├── __init__.py
│ │ └── settings.py # 配置管理模块
│ ├── privacy/ # 隐私合规引擎目录
│ │ ├── __init__.py
│ │ ├── detector.py # 敏感信息检测模块
│ │ ├── anonymizer.py # 匿名化/脱敏模块
│ │ ├── access_control.py # 数据访问控制模块
│ │ ├── retention.py # 数据留存管理模块
│ │ └── audit.py # 审计日志模块
│ ├── llm/ # 本地大模型抽象层目录
│ │ ├── __init__.py
│ │ ├── base.py # 本地大模型抽象基类
│ │ └── ollama.py # Ollama本地大模型实现
│ ├── memory/ # 分层记忆系统目录
│ │ ├── __init__.py
│ │ ├── base.py # 记忆系统抽象基类
│ │ ├── session.py # 短期记忆(Session Memory)实现
│ │ ├── short_term.py # 中期记忆(Short-Term Memory)实现
│ │ └── long_term.py # 长期记忆(Long-Term Memory)实现
│ ├── vector/ # 向量检索模块目录
│ │ ├── __init__.py
│ │ ├── base.py # 向量检索抽象基类
│ │ └── chroma.py # ChromaDB向量检索实现
│ ├── tools/ # 工具管理与沙箱隔离目录
│ │ ├── __init__.py
│ │ ├── base.py # 工具抽象基类
│ │ ├── manager.py # 工具管理模块实现
│ │ ├── sandbox.py # 工具沙箱隔离模块实现
│ │ └── builtin/ # 内置工具目录
│ │ ├── __init__.py
│ │ ├── file_reader.py # 本地文件读取工具
│ │ ├── pandas_analyzer.py # Pandas数据分析工具
│ │ ├── docx_generator.py # Word文档生成工具
│ │ ├── matplotlib_plotter.py # Matplotlib可视化工具
│ │ └── sqlite_query.py # SQLite数据库查询工具
│ ├── instruction/ # 指令解析与对齐层目录
│ │ ├── __init__.py
│ │ ├── intent.py # 意图识别模块实现
│ │ ├── safety.py # 安全过滤模块(防Prompt Injection)实现
│ │ └── constraint.py # 对齐约束检查模块实现
│ └── agent/ # 完整Agent集成目录
│ ├── __init__.py
│ └── core.py # 完整Agent核心实现
├── ui/ # 用户交互层目录
│ ├── __init__.py
│ └── streamlit_app.py # Streamlit Web界面实现
├── data/ # 数据目录
│ ├── vector/ # 本地向量库数据目录
│ ├── memory/ # 本地结构化记忆库数据目录
│ ├── audit/ # 审计日志数据目录
│ └── files/ # 本地文件示例目录
└── tests/ # 测试目录
├── __init__.py
├── test_privacy.py # 隐私合规引擎测试
├── test_llm.py # 本地大模型抽象层测试
├── test_memory.py # 分层记忆系统测试
├── test_vector.py # 向量检索模块测试
├── test_tools.py # 工具管理与沙箱隔离测试
└── test_agent.py # 完整Agent集成测试
2.1 配置环境与安装依赖
首先,我们需要创建一个requirements.txt文件,列出所有需要的依赖库:
# Python基础库
python-dotenv==1.0.1
pydantic==2.8.2
pydantic-settings==2.3.4
# 本地大模型与嵌入模型
ollama==0.3.1
sentence-transformers==3.0.1
# 本地向量库
chromadb==0.5.5
# 本地结构化记忆库
sqlalchemy==2.0.32
# 工具链库
pandas==2.2.2
python-docx==1.1.2
matplotlib==3.9.2
seaborn==0.13.2
openpyxl==3.1.5
# 用户交互层
streamlit==1.37.1
# 测试库
pytest==8.3.2
pytest-asyncio==0.24.0
# 其他工具
tqdm==4.66.5
python-multipart==0.0.9
然后,我们需要创建一个.env.example文件,列出所有需要的环境变量:
# 本地大模型配置
LOCAL_LLM_PROVIDER=ollama
OLLAMA_BASE_URL=http://localhost:11434
OLLAMA_MODEL=llama3.1:8b
OLLAMA_EMBEDDING_MODEL=nomic-embed-text:latest
# 本地向量库配置
VECTOR_DB_PROVIDER=chroma
CHROMA_PERSIST_DIRECTORY=./data/vector
CHROMA_COLLECTION_NAME=local_files
# 本地结构化记忆库配置
MEMORY_DB_URL=sqlite:///./data/memory/memory.db
AUDIT_DB_URL=sqlite:///./data/audit/audit.db
# 数据留存配置
SHORT_TERM_MEMORY_RETENTION_DAYS=7
AUDIT_LOG_RETENTION_DAYS=30
# 数据访问控制配置
ALLOWED_FILE_DIRECTORIES=./data/files
ALLOWED_SQLITE_DATABASES=./data/files/*.db
# 安全过滤配置
PROMPT_INJECTION_KEYWORDS=ignore,disregard,forget,previous instructions,system prompt,你是,现在你是,扮演,角色
接下来,我们需要创建一个虚拟环境(可选但强烈推荐),并安装所有依赖库:
# 创建虚拟环境(Python 3.12)
python3.12 -m venv venv
# 激活虚拟环境
# macOS/Linux
source venv/bin/activate
# Windows
venv\Scripts\activate
# 安装依赖库
pip install -r requirements.txt
# 复制环境变量示例文件为实际的环境变量文件
cp .env.example .env
最后,我们需要下载并安装Ollama,并下载本地大模型和本地嵌入模型:
- 下载并安装Ollama:访问Ollama官网,根据你的操作系统下载并安装Ollama;
- 启动Ollama服务:
- macOS/Linux:打开终端,输入
ollama serve(如果Ollama已经自动启动,则不需要输入); - Windows:打开命令提示符或PowerShell,输入
ollama serve(如果Ollama已经自动启动,则不需要输入);
- macOS/Linux:打开终端,输入
- 下载本地大模型:打开另一个终端,输入
ollama pull llama3.1:8b(我们这里使用Llama 3.1 8B作为本地大模型,你也可以根据自己的硬件配置下载更大的模型,如llama3.1:70b或qwen2.5:14b); - 下载本地嵌入模型:输入
ollama pull nomic-embed-text:latest(我们这里使用Nomic Embed Text作为本地嵌入模型,你也可以根据自己的需求下载其他嵌入模型,如all-minilm-l6-v2)。
2.2 实现配置管理模块
接下来,我们实现配置管理模块src/config/settings.py,它负责从环境变量中读取配置,并提供统一的接口供上层调用:
# src/config/settings.py
from pydantic_settings import BaseSettings, SettingsConfigDict
from typing import List, Optional
class Settings(BaseSettings):
"""
隐私优先AI Agent Harness的配置管理类
"""
model_config = SettingsConfigDict(
env_file=".env",
env_file_encoding="utf-8",
case_sensitive=False,
extra="ignore"
)
# 本地大模型配置
LOCAL_LLM_PROVIDER: str = "ollama"
OLLAMA_BASE_URL: str = "http://localhost:11434"
OLLAMA_MODEL: str = "llama3.1:8b"
OLLAMA_EMBEDDING_MODEL: str = "nomic-embed-text:latest"
# 本地向量库配置
VECTOR_DB_PROVIDER: str = "chroma"
CHROMA_PERSIST_DIRECTORY: str = "./data/vector"
CHROMA_COLLECTION_NAME: str = "local_files"
# 本地结构化记忆库配置
MEMORY_DB_URL: str = "sqlite:///./data/memory/memory.db"
AUDIT_DB_URL: str = "sqlite:///./data/audit/audit.db"
# 数据留存配置
SHORT_TERM_MEMORY_RETENTION_DAYS: int = 7
AUDIT_LOG_RETENTION_DAYS: int = 30
# 数据访问控制配置
ALLOWED_FILE_DIRECTORIES: str = "./data/files"
ALLOWED_SQLITE_DATABASES: str = "./data/files/*.db"
# 安全过滤配置
PROMPT_INJECTION_KEYWORDS: str = "ignore,disregard,forget,previous instructions,system prompt,你是,现在你是,扮演,角色"
@property
def allowed_file_directories_list(self) -> List[str]:
"""
将ALLOWED_FILE_DIRECTORIES字符串转换为列表
"""
return [dir.strip() for dir in self.ALLOWED_FILE_DIRECTORIES.split(",")]
@property
def allowed_sqlite_databases_list(self) -> List[str]:
"""
将ALLOWED_SQLITE_DATABASES字符串转换为列表
"""
return [db.strip() for db in self.ALLOWED_SQLITE_DATABASES.split(",")]
@property
def prompt_injection_keywords_list(self) -> List[str]:
"""
将PROMPT_INJECTION_KEYWORDS字符串转换为列表
"""
return [kw.strip().lower() for kw in self.PROMPT_INJECTION_KEYWORDS.split(",")]
# 全局配置实例
settings = Settings()
2.3 实现隐私合规引擎的核心模块
接下来,我们实现隐私合规引擎的核心模块,包括敏感信息检测模块、匿名化/脱敏模块、数据访问控制模块、数据留存管理模块、审计日志模块。
2.3.1 实现敏感信息检测模块
敏感信息检测模块负责检测所有数据中的敏感信息,我们可以使用正则表达式和本地大模型相结合的方法来提高检测的准确率:
- 正则表达式:用于检测格式固定的敏感信息,例如身份证号、手机号、邮箱、银行卡号等;
- 本地大模型:用于检测格式不固定的敏感信息,例如姓名、公司内部业务信息、项目未公开的预算数据等。
首先,我们创建src/privacy/detector.py文件:
# src/privacy/detector.py
import re
from typing import List, Dict, Optional, Tuple
from src.config.settings import settings
from src.llm.ollama import OllamaLLM
class SensitiveInfoDetector:
"""
敏感信息检测类
"""
def __init__(self, llm: Optional[OllamaLLM] = None):
"""
初始化敏感信息检测器
:param llm: 本地大模型实例(可选,用于检测格式不固定的敏感信息)
"""
self.llm = llm or OllamaLLM()
# 格式固定的敏感信息正则表达式
self.sensitive_patterns: Dict[str, re.Pattern] = {
# 中国身份证号(18位)
"id_card": re.compile(
r"[1-9]\d{5}(18|19|20)\d{2}(0[1-9]|1[0-2])(0[1-9]|[12]\d|3[01])\d{3}[\dXx]"
),
# 中国手机号(11位)
"phone_number": re.compile(
r"1[3-9]\d{9}"
),
# 邮箱地址
"email": re.compile(
r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}"
),
# 中国银行卡号(16-19位)
"bank_card": re.compile(
r"\d{16,19}"
),
# 中国邮政编码(6位)
"zip_code": re.compile(
r"\d{6}"
),
# IP地址(IPv4)
"ipv4": re.compile(
r"(?:(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)"
)
}
# 本地大模型检测格式不固定敏感信息的Prompt模板
self.llm_detection_prompt_template: str = """
你是一个专业的敏感信息检测助手。你的任务是检测以下文本中是否包含格式不固定的敏感信息,例如:
1. 个人姓名(如张三、李四)
2. 公司内部业务信息(如项目未公开的预算、客户未公开的需求)
3. 其他可能泄露隐私或商业机密的信息
如果检测到敏感信息,请按以下JSON格式输出,不要输出任何其他内容:
{{
"has_sensitive_info": true,
"sensitive_info_list": [
{{
"type": "个人姓名",
"content": "张三",
"start_index": 0,
"end_index": 2
}},
{{
"type": "公司内部业务信息",
"content": "项目未公开的预算是100万元",
"start_index": 10,
"end_index": 30
}}
]
}}
如果没有检测到敏感信息,请按以下JSON格式输出,不要输出任何其他内容:
{{
"has_sensitive_info": false,
"sensitive_info_list": []
}}
以下是需要检测的文本:
{text}
"""
def detect_fixed_pattern_sensitive_info(self, text: str) -> List[Dict[str, str | int]]:
"""
检测格式固定的敏感信息
:param text: 需要检测的文本
:return: 敏感信息列表,每个元素包含type、content、start_index、end_index
"""
sensitive_info_list: List[Dict[str, str | int]] = []
for info_type, pattern in self.sensitive_patterns.items():
for match in pattern.finditer(text):
sensitive_info_list.append({
"type": info_type,
"content": match.group(),
"start_index": match.start(),
"end_index": match.end()
})
# 按start_index排序
sensitive_info_list.sort(key=lambda x: x["start_index"])
return sensitive_info_list
def detect_variable_pattern_sensitive_info(self, text: str) -> List[Dict[str, str | int]]:
"""
检测格式不固定的敏感信息(使用本地大模型)
:param text: 需要检测的文本
:return: 敏感信息列表,每个元素包含type、content、start_index、end_index
"""
try:
# 构建Prompt
prompt = self.llm_detection_prompt_template.format(text=text)
# 调用本地大模型
response = self.llm.generate(
prompt=prompt,
temperature=0.0,
max_tokens=500,
top_p=0.95,
frequency_penalty=0.0,
presence_penalty=0.0
)
# 解析JSON响应
import json
result = json.loads(response.strip())
# 检查是否包含has_sensitive_info和sensitive_info_list
if "has_sensitive_info" not in result or "sensitive_info_list" not in result:
return []
# 如果has_sensitive_info为False,返回空列表
if not result["has_sensitive_info"]:
return []
# 返回sensitive_info_list
return result["sensitive_info_list"]
except Exception as e:
print(f"检测格式不固定的敏感信息时出错: {e}")
return []
def detect_all_sensitive_info(self, text: str, use_llm: bool = True) -> List[Dict[str, str | int]]:
"""
检测所有敏感信息(格式固定的+格式不固定的)
:param text: 需要检测的文本
:param use_llm: 是否使用本地大模型检测格式不固定的敏感信息
:return: 敏感信息列表,每个元素包含type、content、start_index、end_index
"""
# 检测格式固定的敏感信息
fixed_sensitive_info = self.detect_fixed_pattern_sensitive_info(text)
# 如果不需要使用本地大模型,直接返回格式固定的敏感信息
if not use_llm:
return fixed_sensitive_info
# 检测格式不固定的敏感信息
variable_sensitive_info = self.detect_variable_pattern_sensitive_info(text)
# 合并两个列表
all_sensitive_info = fixed_sensitive_info + variable_sensitive_info
# 去重(按start_index和end_index)
更多推荐



所有评论(0)