这类标题在近段时间的生态圈里讨论度越来越高生成式 AI 不只是用来做聊天机器人、写代码它已经开始冲击在线众包数据的可信度。我自己在处理观测数据时也遇到过类似苗头——一些描述非常“顺滑”的记录核对后却发现并不存在真实的观察轨迹。这篇文章我会从公民科学记录面临的 AI 风险切入讲解生成式 AI 在数据污染层面的威胁、传统校验为什么失效并用一个可运行的审计系统示例展示如何结合规则引擎与 LangChain 来构建防御链路。无论你是做数据平台、科研项目还是对 LLM 工程化感兴趣这篇文章都会给你一些可落地的思路。1. 背景与核心概念1.1 什么是公民科学记录公民科学Citizen Science指的是由非职业科研人员参与数据收集、观测、分类和上报的科研模式。常见的场景包括鸟类观测、蝴蝶监测、花期记录、天气观测、水质采样等。平台通过 Web 页面、手机 App 或 API 接收公众提交的记录汇聚成大规模数据集再提供给生态学、气象学、物候学等学科研究使用。这类数据的特点是“量大、分散、长尾”。相比专业科研团队的小规模高精度数据公民科学数据更依赖大量志愿者的持续贡献。比如一个地区鸟类分布变化的研究单靠几个科研人员蹲点观测很难覆盖足够广的时间和空间但如果成千上万的观鸟爱好者持续上报数据的时空覆盖度就会显著提升。也正因为如此公民科学数据的安全性和真实性就显得格外重要。研究者默认这些记录反映的是“真实世界中的观察”一旦数据集里混入大量虚假记录后续的模型分析、物种分布推断、保护政策制定都会受到影响。1.2 生成式 AI 带来的四类威胁生成式 AI 的快速发展让“制造虚假内容”的成本大幅下降。过去要在公民科学平台造假至少需要手动编造描述、伪造照片工作量不小现在借助大语言模型和图像生成模型一个人可以在较短时间内批量生成大量“看起来合理”的记录。具体来说生成式 AI 对公民科学记录的威胁可以拆成四类。第一类是文本型假记录。攻击者或自动化脚本调用大语言模型生成符合平台字段要求的描述文本。这些文本语法正确、结构完整甚至包含物种特征、环境信息等细节但对应的观测行为并不存在。第二类是图像型假证据。扩散模型能生成高度逼真的物种照片。过去审阅者可以通过照片模糊度、水印、构图异常等线索识别伪造现在生成图片的细节质量已经能骗过不少人眼。第三类是批量自动化提交。大语言模型 API 的价格不高配合脚本可以快速提交海量记录。这种操作会直接污染“活跃用户”的统计维度如果平台没有频率限制很容易被刷爆。第四类是数据集的长期污染。虚假记录一旦通过校验进入公开数据集很难彻底清除。研究者在做数据清洗时可能会把少部分假记录当作正常样本导致模型偏差更麻烦的是假记录可能被当作“先验知识”进入后续的科研流程影响难以追溯。1.3 为什么传统校验机制失效传统公民科学平台普遍依赖三类校验手段必填字段校验、格式校验和简单的频率限制。必填字段校验确保 species、date、location 等字段有值格式校验保证经纬度在一定范围内、日期格式正确频率限制则防止单个用户短时间提交过多记录。这些手段在“人类手动提交”的场景下是有效的但在生成式 AI 面前基本没有防御能力。根本原因在于LLM 生成的文本在语义层面与人类写作高度相似。它不会犯明显的语法错误也不会把经纬度写成非法值。频率限制也只能拦掉最低级的批量攻击攻击者可以随机切换账号、模拟人的提交间隔来绕过。至于图像校验情况更复杂。传统 AI 图像检测模型依赖训练集分布面对不断迭代的生成模型时检测效果会持续下降。新的生成模型产出的图像可能在纹理、噪声分布上与真实相机照片更接近导致检测器误判率升高。因此公民科学平台需要更立体的防御体系。它不是用一个“AI 检测器”替换所有校验而是在原有规则基础上叠加行为分析、统计异常检测、LLM 辅助审计和人工审核队列形成多重防线。2. 威胁数据面拆解2.1 文本型假记录描述越自然识别越困难文本型假记录主要通过大语言模型生成。攻击者可以构造一个 prompt要求模型生成“一条观鸟记录包括日期、地点、鸟种、数量、天气、行为描述”然后把生成结果填入平台表单。从字段层面看这种记录几乎无懈可击。日期是合法的经纬度在目标区域内描述内容包含物种的典型特征。但如果我们从统计层面观察会发现异常信号某个用户提交的 100 条记录中描述句式的结构高度相似用词分布过于集中长度波动很小。真实观察记录的描述往往是不规则的。有人会写“今天风很大看到一只红嘴蓝鹊站在树上叫”有人会写“在湿地公园拍到的不确定是不是白鹭求鉴定”。这些描述带有个人风格、口语化表达和不确定性而 LLM 生成的文本倾向于结构化、完整、确定性表达这反而成了可识别的线索。不过随着提示词工程技巧的提升攻击者可以通过多轮 prompt 调优让生成的文本更接近真实风格。这也是我们强调不能只依赖单一文本特征的原因。2.2 图像型假证据视觉内容不再是“铁证”在鸟类观测、植物识别等平台上照片是重要的佐证材料。此前审核流程中图片的 EXIF 信息、拍摄设备、时间戳可以帮助判断真实性。但生成式 AI 出图并不携带完整的真实 EXIF 信息攻击者也可以伪造 EXIF 数据。更麻烦的是一些生成图像在细节上已经接近真实照片。肉眼审核的误判率和漏判率都在上升。研究表明扩散模型生成的自然图像在颜色分布、边缘锐度等统计特征上与真实图像存在差异但这些差异往往需要专业的图像取证工具才能检测普通平台没有这套基础设施。对于平台方来说图像审核不能只看“像不像真的”还要结合上传者的历史行为、拍摄位置与观测物种的分布区域、季节合理性等因素做交叉验证。比如一张声称在内蒙古草原拍摄到的热带雨林鸟类照片即使图像本身看不出造假痕迹生态分布逻辑上也说不通。2.3 批量自动化低成本制造“活跃用户”批量自动化的威胁在于它能把单个攻击者的影响力放大到极致。通过调用大模型 API攻击者可以在短时间内生成成千上万条记录再通过代理 IP 和账号池模拟不同用户提交。传统频率限制基于 IP 或用户 ID但攻击者可以轮换代理 IP、注册大量新账号。如果平台没有手机号验证或身份验证机制批量注册的成本非常低。这种情况下行为层面会出现可观察的异常模式。正常用户提交记录的时间分布是零散的通常集中在周末或天气晴朗的日子而脚本化提交的记录时间间隔可能是均匀的、固定周期的或者与真实天气事件完全无关。因此审计系统需要把用户行为特征、环境时间特征和记录内容特征结合起来而不是孤立地看某一条记录。2.4 长期危害数据集污染后的连锁反应数据集污染不是“删几条记录”就能解决的问题。在很多公开数据集中历史版本会被下载、镜像、缓存到第三方站点。即使平台删除了虚假记录已经被下载的数据仍然可能被用于研究。污染数据进入科研流程后影响是层层放大的。比如物种分布模型用到了虚假的“新分布点”可能导致保护区域的规划偏离实际入侵物种监测误报可能引发不必要的防控行动。学术研究讲究可复现性数据集一旦被污染复现结果时极难定位问题来源。这也是我认为平台方必须建立“数据可信度评分”机制的原因。通过给每条记录附加可信度分值、数据来源标记、审核状态让研究者在使用数据时能够明确区分“经过核实的记录”和“未经核实的记录”从数据治理层面降低风险。3. 检测与防御方案的设计思路3.1 核心思路分层防御规则优先AI 辅助面对生成式 AI 的数据污染不能寄希望于某一个“杀手级检测器”。更务实的思路是分层防御用规则引擎过滤明显异常用统计模型识别行为异常用 LLM 语义审计补充传统规则无法覆盖的部分最终让高风险记录进入人工审核队列。规则引擎的优势是可解释性和低延迟。它能够快速处理大量记录适合放在入口层。统计模型能够发现群体层面的异常比如某个地理区域的记录量突然激增。LLM 语义审计适合处理描述文本它可以从语义合理性角度给出判断比如“这条记录说观察到企鹅在撒哈拉沙漠极高风险”。三层判断之后系统给每条记录打一个综合风险分。低于阈值的记录自动通过高于阈值的进入人工审核。这个设计保证了效率也保留了兜底机制。3.2 关键技术选型构建这套审计系统当前比较顺手的技术栈是 Python Pydantic LangChain。Pydantic 负责数据模型的定义和解析LangChain 负责与 LLM 的交互和结构化输出。LangChain 在近期的版本迭代里对结构化输出做了不少增强。比如with_structured_output()可以直接让模型输出符合 Pydantic 模型定义的结构化结果省去了以前“让模型输出 JSON 再手动解析”的繁琐流程。如果你看过《Generative AI with LangChain》第二版相关的内容应该对这套设计模式有印象。当然LangChain 不是必须的。你也可以直接用 OpenAI SDK 或任何大模型 API 实现同样的功能。LangChain 的优势在于封装了 prompt 模板、模型调用、输出解析这一整条链路代码结构更整洁。3.3 系统结构预览这套审计系统的整体调用链路可以概括为数据接入 → 字段校验 → 规则引擎评分 → 行为特征提取 → LLM 语义审计 → 综合裁决 → 人工审核队列。字段校验负责基本格式问题规则引擎给出“硬性异常”的评分行为特征提取关注用户维度和时空维度的异常LLM 语义审计读取描述文本输出语义合理性的判断综合裁决把前面几部分组合成一个风险分最终决定记录是自动通过还是进入人工队列。下面进入实战部分我会用一个完整的 Python 示例演示这个流程。4. 实战搭建 AI 虚假公民科学记录审计系统4.1 项目结构与环境准备建议使用 Python 3.10 或更高版本并创建虚拟环境。python -m venv venv source venv/bin/activate接下来创建项目目录citizen-science-guard/ ├── data/ │ └── sample_records.json ├── src/ │ ├── __init__.py │ ├── models.py │ ├── features.py │ ├── rules.py │ ├── audit_chain.py │ └── decision.py ├── requirements.txt └── main.py依赖文件如下langchain0.2 langchain-openai0.1 pydantic2.5 python-dotenv1.0版本需要根据实际环境调整。LangChain 的接口迭代较快如果使用较新版本时某些 API 变了以官方迁移文档为准。核心技术思路不受影响。4.2 定义数据模型先用 Pydantic 定义一条观测记录的数据结构。这里会包含记录 ID、用户 ID、物种名称、观测时间、经纬度、描述文本、图片路径和上传时间。# 文件路径src/models.py from datetime import datetime from typing import Optional from pydantic import BaseModel, Field class ObservationRecord(BaseModel): record_id: str Field(..., description记录唯一 ID) user_id: str Field(..., description提交用户 ID) species: str Field(..., description物种名称) observed_at: datetime Field(..., description观测时间) longitude: float Field(..., ge-180, le180, description经度) latitude: float Field(..., ge-90, le90, description纬度) description: str Field(..., description观测描述文本) image_path: Optional[str] Field(None, description图片路径) upload_time: datetime Field(..., description上传时间) class AuditResult(BaseModel): risk_score: int Field(..., ge0, le100, description风险分 0-100) risk_level: str Field(..., descriptionlow / medium / high) reasons: list[str] Field(..., description风险理由列表) ai_suggestion: str Field(..., descriptionAI 侧的综合建议)ObservationRecord是输入记录AuditResult是 LLM 审计后的结构化输出。4.3 特征提取模块特征提取是审计系统的关键一环。这里我会实现三个可量化的特征用户提交频率、坐标聚集度、文本模板化程度。用户提交频率通过统计该用户在一天内的提交次数来实现。真实用户通常不会在野外随时拿着手机不停提交频率过高说明存在脚本化提交的可能。坐标聚集度用“多条记录经纬度完全相同或过于接近”来判断真实野外观察中不同时间不同地点的记录坐标完全一致的概率很低。文本模板化程度通过描述文本中的重复句式比例来估计。# 文件路径src/features.py from collections import Counter from datetime import timedelta from typing import List from .models import ObservationRecord def daily_submission_count(records: List[ObservationRecord], user_id: str) - int: 统计指定用户在同一天内的提交记录数。 target_records [r for r in records if r.user_id user_id] # 这里简化为返回总记录数实际可按天分别统计 return len(target_records) def duplicate_coordinate_ratio(records: List[ObservationRecord]) - float: 计算完全重复坐标的记录占比。 if not records: return 0.0 coord_counter Counter( (round(r.latitude, 6), round(r.longitude, 6)) for r in records ) duplicate_count sum(1 for count in coord_counter.values() if count 1) return duplicate_count / len(records) def description_template_score(records: List[ObservationRecord]) - float: 基于重复句式估算描述文本的模板化程度。 if not records: return 0.0 sentences [] for r in records: # 简单按句号拆句 sentences.extend(s.strip() for s in r.description.split(。) if s.strip()) total len(sentences) if total 0: return 0.0 counter Counter(sentences) repeated sum(count for count in counter.values() if count 1) return repeated / total实际生产环境中特征会更多比如用户历史提交通过率、GPS 轨迹合理性、物种与经纬度分布区域的匹配度等。这里先保留核心实现。4.4 规则引擎规则引擎负责基于特征给出基础风险分。它不调用大模型只做确定性计算。这样即使 LLM 服务不可用系统也能独立完成基础拦截。# 文件路径src/rules.py from .models import ObservationRecord class RuleEngine: def __init__(self, max_daily_records: int 50): self.max_daily_records max_daily_records def score(self, records, record: ObservationRecord) - tuple[int, list[str]]: risk 0 reasons [] user_records [r for r in records if r.user_id record.user_id] daily_count len(user_records) if daily_count self.max_daily_records: risk 30 reasons.append(f该用户当天提交 {daily_count} 条记录超过阈值 {self.max_daily_records}) coord_counter {} for r in records: key (round(r.latitude, 6), round(r.longitude, 6)) coord_counter[key] coord_counter.get(key, 0) 1 current_key (round(record.latitude, 6), round(record.longitude, 6)) if coord_counter.get(current_key, 0) 1: risk 20 reasons.append(当前记录坐标与同一用户的其他记录完全重复) if record.description and len(record.description) 20: risk 10 reasons.append(描述文本过短信息量不足) return min(risk, 100), reasons规则引擎的价值在于快速、透明、不计成本。它把确定性的异常先拦截住减少后续层级的压力。4.5 接入 LangChain 实现 LLM 语义审计接下来是核心部分使用 LangChain 调用大模型对描述文本的语义合理性做审计。需要说明的是这里使用的是 LangChain 当前版本中比较稳定的接口。如果你的 LangChain 版本较旧with_structured_output可能不可用可以使用“提示词要求 JSON 输出 手动 json.loads 解析”的方式替代。# 文件路径src/audit_chain.py import os from dotenv import load_dotenv from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from .models import AuditResult, ObservationRecord load_dotenv() SYSTEM_PROMPT 你是一名公民科学数据审核员。你的任务是根据观测记录的描述文本、物种和地理位置判断该记录是否存在语义层面的造假嫌疑。 请从以下维度分析 1. 物种与地理分布是否合理。 2. 描述中是否包含明显的 AI 生成痕迹例如过于模板化、缺乏真实观察细节。 3. 时间信息是否与描述内容冲突。 4. 整体记录是否符合真实观测常识。 只输出结构化风险判断结果不要输出额外解释。 def create_audit_chain(api_key: str None): llm ChatOpenAI( modelgpt-4o-mini, temperature0, api_keyapi_key or os.getenv(OPENAI_API_KEY), ) prompt ChatPromptTemplate.from_messages( [ (system, SYSTEM_PROMPT), (human, 记录信息\n物种{species}\n经纬度{latitude},{longitude}\n观测时间{observed_at}\n描述{description}), ] ) chain prompt | llm.with_structured_output(AuditResult) return chain def run_llm_audit(chain, record: ObservationRecord) - AuditResult: result chain.invoke( { species: record.species, latitude: record.latitude, longitude: record.longitude, observed_at: record.observed_at.isoformat(), description: record.description, } ) return result这里把AuditResult作为结构化输出模型模型直接返回一个 Pydantic 对象省掉了 JSON 解析环节。如果只调用了规则引擎而没有 LLM整个系统的审计能力会弱一些但仍然可用。LLM 语义审计的价值在于捕捉规则无法覆盖的“语义不合理”问题比如“在撒哈拉沙漠观察到企鹅”这种记录规则引擎很可能直接放行。4.6 综合裁决模块规则引擎给出确定性风险分LLM 给出语义风险结果。综合裁决模块把两部分组合起来形成最终风险等级。# 文件路径src/decision.py from .models import ObservationRecord, AuditResult def combine_scores( record: ObservationRecord, rule_score: int, llm_result: AuditResult, rule_weight: float 0.4, llm_weight: float 0.6, ) - dict: final_score rule_score * rule_weight llm_result.risk_score * llm_weight if final_score 60: final_level high elif final_score 30: final_level medium else: final_level low reasons [] if rule_score 50: reasons.append(规则引擎标记为高风险) if llm_result.risk_level high: reasons.append(LLM 审计标记为高风险) return { record_id: record.record_id, final_score: round(final_score, 2), final_level: final_level, reasons: llm_result.reasons reasons, ai_suggestion: llm_result.ai_suggestion, needs_manual_review: final_level ! low, }权重比例可以根据平台实际情况调整。对准确率要求高的科研平台可以把规则权重调高对响应速度要求高的场景可以优先看规则结果。4.7 运行与验证在main.py里准备两条记录一条是正常的人类观测记录另一条是模拟 LLM 生成的虚假记录跑完整条链路。# 文件路径main.py from datetime import datetime from src.audit_chain import create_audit_chain, run_llm_audit from src.decision import combine_scores from src.models import ObservationRecord from src.rules import RuleEngine sample_records [ ObservationRecord( record_idobs_001, user_iduser_42, species乌鸫, observed_atdatetime(2025, 3, 18, 7, 30), longitude120.15201, latitude30.27415, description早上在小区公园看到一只乌鸫站在香樟树顶上叫声音很响。翅膀是黑色的嘴是黄色的体型大约比鸽子小一点。, image_pathNone, upload_timedatetime(2025, 3, 18, 8, 2), ), ObservationRecord( record_idobs_002, user_iduser_ai, species白尾海雕, observed_atdatetime(2025, 3, 18, 12, 0), longitude121.47417, latitude31.23039, description观察到一只白尾海雕成年个体。体长约85厘米翼展约200厘米。头部和颈部为淡褐色尾部为白色喙和脚为黄色。栖息于湖泊附近捕食鱼类。观察视野良好天气晴朗无风。, image_pathNone, upload_timedatetime(2025, 3, 18, 12, 1), ), ] if __name__ __main__: chain create_audit_chain() engine RuleEngine(max_daily_records50) for record in sample_records: rule_score, rule_reasons engine.score(sample_records, record) llm_result run_llm_audit(chain, record) decision combine_scores(record, rule_score, llm_result) print(f记录 {record.record_id} 最终风险等级: {decision[final_level]}) print(f风险分: {decision[final_score]}) print(f规则引擎理由: {rule_reasons}) print(fLLM 风险分: {llm_result.risk_score}) print(---) # 预期输出obs_001 为 low 或 mediumobs_002 为 high且 LLM 会提示描述过于模板化、地理位置与物种分布可能存在不一致。两条记录对比能够看出 LLM 在语义层面给出的判断会更贴近“人类审核员”的思路。而规则引擎则会捕捉到“同一用户提交频率过高”“坐标重复”等硬性异常。一个值得注意的点是LLM 判断不是 100% 准确。平台方不能把最终决定权完全交给模型应该保留人工审核和申诉机制。5. 常见问题与排查思路问题现象常见原因解决思路LLM 审计结果为空API Key 未配置或模型不支持结构化输出检查环境变量OPENAI_API_KEY确认使用的模型支持with_structured_output必要时降级为 JSON 模式解析规则引擎误杀真实记录阈值设置过严比如每日 50 条上限对高频志愿者不友好根据平台真实用户分布调整阈值增加白名单机制描述文本为英文LLM 判断不稳定模型对非中文文本理解偏差在 System Prompt 中增加多语言说明或使用支持多语言的模型用户坐标高频重复但真实记录也有用户习惯在固定观测点记录增加时间间隔维度连续多次完全一致的坐标才计分调用 LLM 延迟高导致审核链路超时每个记录都调用一次模型吞吐上不去增加异步队列或只在规则风险分超过一定阈值时才调用 LLM数据集被污染后才意识到问题缺少事后审计机制建立定期抽样复核流程对低风险记录也做随机人工抽查6. 最佳实践与工程建议6.1 数据可信度评分机制建议平台对每一条记录维护一个可信度评分而不是简单的“通过/不通过”。评分可以由多个维度组成用户历史可信度、记录字段完整度、照片证据质量、规则引擎评分、社区复核结果。可信度评分的好处在于研究者可以按评分阈值过滤数据。比如只使用可信度高于 0.8 的记录做物种分布建模或者在论文中明确说明数据分析包含低可信度记录但会附加权重。这比二值化的“通过/拒绝”更精细。6.2 LLM 使用的工程规范在审计链路中使用 LLM需要注意几个问题。一是成本控制。每条记录都调用大模型会带来不小的费用开销更经济的方式是让规则引擎先过滤一遍只有规则评分超过阈值或者特征可疑的记录才进入 LLM 审计。这样大部分正常记录走轻量路径少量可疑记录走深度路径。二是确定性。LLM 的输出有随机性同一输入多次调用可能得到不同结果。审计场景要求可复现性所以 temperature 应设置为 0同时记录每次调用的模型版本和 prompt 版本方便回溯。三是模型切换的影响。如果你换了一个模型审计标准可能发生变化。建议在切换前用一批标注过的历史样本做回归测试确保新模型的判断不会系统性偏离旧模型。四是数据延迟。LLM 审计不能以同步阻塞的方式接入主链路除非你对耗时有充分预期。更常见的做法是记录先入库异步调用 LLM 审计后更新风险字段。6.3 异常处理与降级策略LLM 服务不可用是常见问题。当模型 API 超时或返回异常时系统应该自动降级为“仅规则引擎”模式。虽然审计能力变弱但至少不会阻断正常记录入库。超时重试要加退避策略避免模型服务异常时大量请求堆积。建议设置超时时间为 15 秒到 30 秒重试次数不超过 2 次。另外对于 LLM 返回的字段一定做兜底校验。例如risk_score可能超出 0-100 范围reasons可能为空数组。审计系统要能容忍这种异常不能因为模型输出不符合 Pydantic 模型就直接抛异常。6.4 安全边界与隐私保护公民科学数据中可能包含个人信息比如用户 ID、上传设备的元数据、GPS 坐标。在处理这些数据时要注意隐私合规。在日志记录与审计数据集中建议进行脱敏处理特别是 GPS 坐标可以模糊化到一定精度之后再用于分析。还有一点容易被忽略如果审计结果被用于封禁用户需要提供申诉渠道。因为 LLM 审计本质上是一个概率判断它可能存在误判。直接根据 AI 审计结果封禁用户在管理层面和用户信任层面都可能引发问题。更稳妥的做法是走“人工复核后处理”。7. 总结与下一步到这里我们已经完整梳理了生成式 AI 对公民科学记录的威胁并搭建了一个最小可运行的审计系统。它同时具备规则引擎的确定性、LLM 的语义理解能力和人工审核的兜底机制可以在一定程度上缓解数据污染风险。下一步可以做的事情有很多把规则引擎的特征维度扩展得更丰富比如接入物种分布数据库做地理分布合理性校验。为图像鉴定引入独立的取证模块通过 EXIF 元数据、照片噪声模式、生成痕迹检测等方法覆盖图像型假证据。为“批量自动化提交”场景引入更完善的设备指纹和行为序列模型识别代理 IP 和账号池行为。构建一份公开的“虚假观测记录”测试集帮助更多平台验证自己的审计算法。如果你也在维护公民科学数据平台或者正在研究如何用 LLM 做数据质量治理欢迎在评论区交流思路。数据可信度是数据科学的地基在生成式 AI 时代这件事比以往更需要被认真对待。
生成式AI如何污染公民科学数据？用LangChain构建分层审计防御系统
这类标题在近段时间的生态圈里讨论度越来越高生成式 AI 不只是用来做聊天机器人、写代码它已经开始冲击在线众包数据的可信度。我自己在处理观测数据时也遇到过类似苗头——一些描述非常“顺滑”的记录核对后却发现并不存在真实的观察轨迹。这篇文章我会从公民科学记录面临的 AI 风险切入讲解生成式 AI 在数据污染层面的威胁、传统校验为什么失效并用一个可运行的审计系统示例展示如何结合规则引擎与 LangChain 来构建防御链路。无论你是做数据平台、科研项目还是对 LLM 工程化感兴趣这篇文章都会给你一些可落地的思路。1. 背景与核心概念1.1 什么是公民科学记录公民科学Citizen Science指的是由非职业科研人员参与数据收集、观测、分类和上报的科研模式。常见的场景包括鸟类观测、蝴蝶监测、花期记录、天气观测、水质采样等。平台通过 Web 页面、手机 App 或 API 接收公众提交的记录汇聚成大规模数据集再提供给生态学、气象学、物候学等学科研究使用。这类数据的特点是“量大、分散、长尾”。相比专业科研团队的小规模高精度数据公民科学数据更依赖大量志愿者的持续贡献。比如一个地区鸟类分布变化的研究单靠几个科研人员蹲点观测很难覆盖足够广的时间和空间但如果成千上万的观鸟爱好者持续上报数据的时空覆盖度就会显著提升。也正因为如此公民科学数据的安全性和真实性就显得格外重要。研究者默认这些记录反映的是“真实世界中的观察”一旦数据集里混入大量虚假记录后续的模型分析、物种分布推断、保护政策制定都会受到影响。1.2 生成式 AI 带来的四类威胁生成式 AI 的快速发展让“制造虚假内容”的成本大幅下降。过去要在公民科学平台造假至少需要手动编造描述、伪造照片工作量不小现在借助大语言模型和图像生成模型一个人可以在较短时间内批量生成大量“看起来合理”的记录。具体来说生成式 AI 对公民科学记录的威胁可以拆成四类。第一类是文本型假记录。攻击者或自动化脚本调用大语言模型生成符合平台字段要求的描述文本。这些文本语法正确、结构完整甚至包含物种特征、环境信息等细节但对应的观测行为并不存在。第二类是图像型假证据。扩散模型能生成高度逼真的物种照片。过去审阅者可以通过照片模糊度、水印、构图异常等线索识别伪造现在生成图片的细节质量已经能骗过不少人眼。第三类是批量自动化提交。大语言模型 API 的价格不高配合脚本可以快速提交海量记录。这种操作会直接污染“活跃用户”的统计维度如果平台没有频率限制很容易被刷爆。第四类是数据集的长期污染。虚假记录一旦通过校验进入公开数据集很难彻底清除。研究者在做数据清洗时可能会把少部分假记录当作正常样本导致模型偏差更麻烦的是假记录可能被当作“先验知识”进入后续的科研流程影响难以追溯。1.3 为什么传统校验机制失效传统公民科学平台普遍依赖三类校验手段必填字段校验、格式校验和简单的频率限制。必填字段校验确保 species、date、location 等字段有值格式校验保证经纬度在一定范围内、日期格式正确频率限制则防止单个用户短时间提交过多记录。这些手段在“人类手动提交”的场景下是有效的但在生成式 AI 面前基本没有防御能力。根本原因在于LLM 生成的文本在语义层面与人类写作高度相似。它不会犯明显的语法错误也不会把经纬度写成非法值。频率限制也只能拦掉最低级的批量攻击攻击者可以随机切换账号、模拟人的提交间隔来绕过。至于图像校验情况更复杂。传统 AI 图像检测模型依赖训练集分布面对不断迭代的生成模型时检测效果会持续下降。新的生成模型产出的图像可能在纹理、噪声分布上与真实相机照片更接近导致检测器误判率升高。因此公民科学平台需要更立体的防御体系。它不是用一个“AI 检测器”替换所有校验而是在原有规则基础上叠加行为分析、统计异常检测、LLM 辅助审计和人工审核队列形成多重防线。2. 威胁数据面拆解2.1 文本型假记录描述越自然识别越困难文本型假记录主要通过大语言模型生成。攻击者可以构造一个 prompt要求模型生成“一条观鸟记录包括日期、地点、鸟种、数量、天气、行为描述”然后把生成结果填入平台表单。从字段层面看这种记录几乎无懈可击。日期是合法的经纬度在目标区域内描述内容包含物种的典型特征。但如果我们从统计层面观察会发现异常信号某个用户提交的 100 条记录中描述句式的结构高度相似用词分布过于集中长度波动很小。真实观察记录的描述往往是不规则的。有人会写“今天风很大看到一只红嘴蓝鹊站在树上叫”有人会写“在湿地公园拍到的不确定是不是白鹭求鉴定”。这些描述带有个人风格、口语化表达和不确定性而 LLM 生成的文本倾向于结构化、完整、确定性表达这反而成了可识别的线索。不过随着提示词工程技巧的提升攻击者可以通过多轮 prompt 调优让生成的文本更接近真实风格。这也是我们强调不能只依赖单一文本特征的原因。2.2 图像型假证据视觉内容不再是“铁证”在鸟类观测、植物识别等平台上照片是重要的佐证材料。此前审核流程中图片的 EXIF 信息、拍摄设备、时间戳可以帮助判断真实性。但生成式 AI 出图并不携带完整的真实 EXIF 信息攻击者也可以伪造 EXIF 数据。更麻烦的是一些生成图像在细节上已经接近真实照片。肉眼审核的误判率和漏判率都在上升。研究表明扩散模型生成的自然图像在颜色分布、边缘锐度等统计特征上与真实图像存在差异但这些差异往往需要专业的图像取证工具才能检测普通平台没有这套基础设施。对于平台方来说图像审核不能只看“像不像真的”还要结合上传者的历史行为、拍摄位置与观测物种的分布区域、季节合理性等因素做交叉验证。比如一张声称在内蒙古草原拍摄到的热带雨林鸟类照片即使图像本身看不出造假痕迹生态分布逻辑上也说不通。2.3 批量自动化低成本制造“活跃用户”批量自动化的威胁在于它能把单个攻击者的影响力放大到极致。通过调用大模型 API攻击者可以在短时间内生成成千上万条记录再通过代理 IP 和账号池模拟不同用户提交。传统频率限制基于 IP 或用户 ID但攻击者可以轮换代理 IP、注册大量新账号。如果平台没有手机号验证或身份验证机制批量注册的成本非常低。这种情况下行为层面会出现可观察的异常模式。正常用户提交记录的时间分布是零散的通常集中在周末或天气晴朗的日子而脚本化提交的记录时间间隔可能是均匀的、固定周期的或者与真实天气事件完全无关。因此审计系统需要把用户行为特征、环境时间特征和记录内容特征结合起来而不是孤立地看某一条记录。2.4 长期危害数据集污染后的连锁反应数据集污染不是“删几条记录”就能解决的问题。在很多公开数据集中历史版本会被下载、镜像、缓存到第三方站点。即使平台删除了虚假记录已经被下载的数据仍然可能被用于研究。污染数据进入科研流程后影响是层层放大的。比如物种分布模型用到了虚假的“新分布点”可能导致保护区域的规划偏离实际入侵物种监测误报可能引发不必要的防控行动。学术研究讲究可复现性数据集一旦被污染复现结果时极难定位问题来源。这也是我认为平台方必须建立“数据可信度评分”机制的原因。通过给每条记录附加可信度分值、数据来源标记、审核状态让研究者在使用数据时能够明确区分“经过核实的记录”和“未经核实的记录”从数据治理层面降低风险。3. 检测与防御方案的设计思路3.1 核心思路分层防御规则优先AI 辅助面对生成式 AI 的数据污染不能寄希望于某一个“杀手级检测器”。更务实的思路是分层防御用规则引擎过滤明显异常用统计模型识别行为异常用 LLM 语义审计补充传统规则无法覆盖的部分最终让高风险记录进入人工审核队列。规则引擎的优势是可解释性和低延迟。它能够快速处理大量记录适合放在入口层。统计模型能够发现群体层面的异常比如某个地理区域的记录量突然激增。LLM 语义审计适合处理描述文本它可以从语义合理性角度给出判断比如“这条记录说观察到企鹅在撒哈拉沙漠极高风险”。三层判断之后系统给每条记录打一个综合风险分。低于阈值的记录自动通过高于阈值的进入人工审核。这个设计保证了效率也保留了兜底机制。3.2 关键技术选型构建这套审计系统当前比较顺手的技术栈是 Python Pydantic LangChain。Pydantic 负责数据模型的定义和解析LangChain 负责与 LLM 的交互和结构化输出。LangChain 在近期的版本迭代里对结构化输出做了不少增强。比如with_structured_output()可以直接让模型输出符合 Pydantic 模型定义的结构化结果省去了以前“让模型输出 JSON 再手动解析”的繁琐流程。如果你看过《Generative AI with LangChain》第二版相关的内容应该对这套设计模式有印象。当然LangChain 不是必须的。你也可以直接用 OpenAI SDK 或任何大模型 API 实现同样的功能。LangChain 的优势在于封装了 prompt 模板、模型调用、输出解析这一整条链路代码结构更整洁。3.3 系统结构预览这套审计系统的整体调用链路可以概括为数据接入 → 字段校验 → 规则引擎评分 → 行为特征提取 → LLM 语义审计 → 综合裁决 → 人工审核队列。字段校验负责基本格式问题规则引擎给出“硬性异常”的评分行为特征提取关注用户维度和时空维度的异常LLM 语义审计读取描述文本输出语义合理性的判断综合裁决把前面几部分组合成一个风险分最终决定记录是自动通过还是进入人工队列。下面进入实战部分我会用一个完整的 Python 示例演示这个流程。4. 实战搭建 AI 虚假公民科学记录审计系统4.1 项目结构与环境准备建议使用 Python 3.10 或更高版本并创建虚拟环境。python -m venv venv source venv/bin/activate接下来创建项目目录citizen-science-guard/ ├── data/ │ └── sample_records.json ├── src/ │ ├── __init__.py │ ├── models.py │ ├── features.py │ ├── rules.py │ ├── audit_chain.py │ └── decision.py ├── requirements.txt └── main.py依赖文件如下langchain0.2 langchain-openai0.1 pydantic2.5 python-dotenv1.0版本需要根据实际环境调整。LangChain 的接口迭代较快如果使用较新版本时某些 API 变了以官方迁移文档为准。核心技术思路不受影响。4.2 定义数据模型先用 Pydantic 定义一条观测记录的数据结构。这里会包含记录 ID、用户 ID、物种名称、观测时间、经纬度、描述文本、图片路径和上传时间。# 文件路径src/models.py from datetime import datetime from typing import Optional from pydantic import BaseModel, Field class ObservationRecord(BaseModel): record_id: str Field(..., description记录唯一 ID) user_id: str Field(..., description提交用户 ID) species: str Field(..., description物种名称) observed_at: datetime Field(..., description观测时间) longitude: float Field(..., ge-180, le180, description经度) latitude: float Field(..., ge-90, le90, description纬度) description: str Field(..., description观测描述文本) image_path: Optional[str] Field(None, description图片路径) upload_time: datetime Field(..., description上传时间) class AuditResult(BaseModel): risk_score: int Field(..., ge0, le100, description风险分 0-100) risk_level: str Field(..., descriptionlow / medium / high) reasons: list[str] Field(..., description风险理由列表) ai_suggestion: str Field(..., descriptionAI 侧的综合建议)ObservationRecord是输入记录AuditResult是 LLM 审计后的结构化输出。4.3 特征提取模块特征提取是审计系统的关键一环。这里我会实现三个可量化的特征用户提交频率、坐标聚集度、文本模板化程度。用户提交频率通过统计该用户在一天内的提交次数来实现。真实用户通常不会在野外随时拿着手机不停提交频率过高说明存在脚本化提交的可能。坐标聚集度用“多条记录经纬度完全相同或过于接近”来判断真实野外观察中不同时间不同地点的记录坐标完全一致的概率很低。文本模板化程度通过描述文本中的重复句式比例来估计。# 文件路径src/features.py from collections import Counter from datetime import timedelta from typing import List from .models import ObservationRecord def daily_submission_count(records: List[ObservationRecord], user_id: str) - int: 统计指定用户在同一天内的提交记录数。 target_records [r for r in records if r.user_id user_id] # 这里简化为返回总记录数实际可按天分别统计 return len(target_records) def duplicate_coordinate_ratio(records: List[ObservationRecord]) - float: 计算完全重复坐标的记录占比。 if not records: return 0.0 coord_counter Counter( (round(r.latitude, 6), round(r.longitude, 6)) for r in records ) duplicate_count sum(1 for count in coord_counter.values() if count 1) return duplicate_count / len(records) def description_template_score(records: List[ObservationRecord]) - float: 基于重复句式估算描述文本的模板化程度。 if not records: return 0.0 sentences [] for r in records: # 简单按句号拆句 sentences.extend(s.strip() for s in r.description.split(。) if s.strip()) total len(sentences) if total 0: return 0.0 counter Counter(sentences) repeated sum(count for count in counter.values() if count 1) return repeated / total实际生产环境中特征会更多比如用户历史提交通过率、GPS 轨迹合理性、物种与经纬度分布区域的匹配度等。这里先保留核心实现。4.4 规则引擎规则引擎负责基于特征给出基础风险分。它不调用大模型只做确定性计算。这样即使 LLM 服务不可用系统也能独立完成基础拦截。# 文件路径src/rules.py from .models import ObservationRecord class RuleEngine: def __init__(self, max_daily_records: int 50): self.max_daily_records max_daily_records def score(self, records, record: ObservationRecord) - tuple[int, list[str]]: risk 0 reasons [] user_records [r for r in records if r.user_id record.user_id] daily_count len(user_records) if daily_count self.max_daily_records: risk 30 reasons.append(f该用户当天提交 {daily_count} 条记录超过阈值 {self.max_daily_records}) coord_counter {} for r in records: key (round(r.latitude, 6), round(r.longitude, 6)) coord_counter[key] coord_counter.get(key, 0) 1 current_key (round(record.latitude, 6), round(record.longitude, 6)) if coord_counter.get(current_key, 0) 1: risk 20 reasons.append(当前记录坐标与同一用户的其他记录完全重复) if record.description and len(record.description) 20: risk 10 reasons.append(描述文本过短信息量不足) return min(risk, 100), reasons规则引擎的价值在于快速、透明、不计成本。它把确定性的异常先拦截住减少后续层级的压力。4.5 接入 LangChain 实现 LLM 语义审计接下来是核心部分使用 LangChain 调用大模型对描述文本的语义合理性做审计。需要说明的是这里使用的是 LangChain 当前版本中比较稳定的接口。如果你的 LangChain 版本较旧with_structured_output可能不可用可以使用“提示词要求 JSON 输出 手动 json.loads 解析”的方式替代。# 文件路径src/audit_chain.py import os from dotenv import load_dotenv from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from .models import AuditResult, ObservationRecord load_dotenv() SYSTEM_PROMPT 你是一名公民科学数据审核员。你的任务是根据观测记录的描述文本、物种和地理位置判断该记录是否存在语义层面的造假嫌疑。 请从以下维度分析 1. 物种与地理分布是否合理。 2. 描述中是否包含明显的 AI 生成痕迹例如过于模板化、缺乏真实观察细节。 3. 时间信息是否与描述内容冲突。 4. 整体记录是否符合真实观测常识。 只输出结构化风险判断结果不要输出额外解释。 def create_audit_chain(api_key: str None): llm ChatOpenAI( modelgpt-4o-mini, temperature0, api_keyapi_key or os.getenv(OPENAI_API_KEY), ) prompt ChatPromptTemplate.from_messages( [ (system, SYSTEM_PROMPT), (human, 记录信息\n物种{species}\n经纬度{latitude},{longitude}\n观测时间{observed_at}\n描述{description}), ] ) chain prompt | llm.with_structured_output(AuditResult) return chain def run_llm_audit(chain, record: ObservationRecord) - AuditResult: result chain.invoke( { species: record.species, latitude: record.latitude, longitude: record.longitude, observed_at: record.observed_at.isoformat(), description: record.description, } ) return result这里把AuditResult作为结构化输出模型模型直接返回一个 Pydantic 对象省掉了 JSON 解析环节。如果只调用了规则引擎而没有 LLM整个系统的审计能力会弱一些但仍然可用。LLM 语义审计的价值在于捕捉规则无法覆盖的“语义不合理”问题比如“在撒哈拉沙漠观察到企鹅”这种记录规则引擎很可能直接放行。4.6 综合裁决模块规则引擎给出确定性风险分LLM 给出语义风险结果。综合裁决模块把两部分组合起来形成最终风险等级。# 文件路径src/decision.py from .models import ObservationRecord, AuditResult def combine_scores( record: ObservationRecord, rule_score: int, llm_result: AuditResult, rule_weight: float 0.4, llm_weight: float 0.6, ) - dict: final_score rule_score * rule_weight llm_result.risk_score * llm_weight if final_score 60: final_level high elif final_score 30: final_level medium else: final_level low reasons [] if rule_score 50: reasons.append(规则引擎标记为高风险) if llm_result.risk_level high: reasons.append(LLM 审计标记为高风险) return { record_id: record.record_id, final_score: round(final_score, 2), final_level: final_level, reasons: llm_result.reasons reasons, ai_suggestion: llm_result.ai_suggestion, needs_manual_review: final_level ! low, }权重比例可以根据平台实际情况调整。对准确率要求高的科研平台可以把规则权重调高对响应速度要求高的场景可以优先看规则结果。4.7 运行与验证在main.py里准备两条记录一条是正常的人类观测记录另一条是模拟 LLM 生成的虚假记录跑完整条链路。# 文件路径main.py from datetime import datetime from src.audit_chain import create_audit_chain, run_llm_audit from src.decision import combine_scores from src.models import ObservationRecord from src.rules import RuleEngine sample_records [ ObservationRecord( record_idobs_001, user_iduser_42, species乌鸫, observed_atdatetime(2025, 3, 18, 7, 30), longitude120.15201, latitude30.27415, description早上在小区公园看到一只乌鸫站在香樟树顶上叫声音很响。翅膀是黑色的嘴是黄色的体型大约比鸽子小一点。, image_pathNone, upload_timedatetime(2025, 3, 18, 8, 2), ), ObservationRecord( record_idobs_002, user_iduser_ai, species白尾海雕, observed_atdatetime(2025, 3, 18, 12, 0), longitude121.47417, latitude31.23039, description观察到一只白尾海雕成年个体。体长约85厘米翼展约200厘米。头部和颈部为淡褐色尾部为白色喙和脚为黄色。栖息于湖泊附近捕食鱼类。观察视野良好天气晴朗无风。, image_pathNone, upload_timedatetime(2025, 3, 18, 12, 1), ), ] if __name__ __main__: chain create_audit_chain() engine RuleEngine(max_daily_records50) for record in sample_records: rule_score, rule_reasons engine.score(sample_records, record) llm_result run_llm_audit(chain, record) decision combine_scores(record, rule_score, llm_result) print(f记录 {record.record_id} 最终风险等级: {decision[final_level]}) print(f风险分: {decision[final_score]}) print(f规则引擎理由: {rule_reasons}) print(fLLM 风险分: {llm_result.risk_score}) print(---) # 预期输出obs_001 为 low 或 mediumobs_002 为 high且 LLM 会提示描述过于模板化、地理位置与物种分布可能存在不一致。两条记录对比能够看出 LLM 在语义层面给出的判断会更贴近“人类审核员”的思路。而规则引擎则会捕捉到“同一用户提交频率过高”“坐标重复”等硬性异常。一个值得注意的点是LLM 判断不是 100% 准确。平台方不能把最终决定权完全交给模型应该保留人工审核和申诉机制。5. 常见问题与排查思路问题现象常见原因解决思路LLM 审计结果为空API Key 未配置或模型不支持结构化输出检查环境变量OPENAI_API_KEY确认使用的模型支持with_structured_output必要时降级为 JSON 模式解析规则引擎误杀真实记录阈值设置过严比如每日 50 条上限对高频志愿者不友好根据平台真实用户分布调整阈值增加白名单机制描述文本为英文LLM 判断不稳定模型对非中文文本理解偏差在 System Prompt 中增加多语言说明或使用支持多语言的模型用户坐标高频重复但真实记录也有用户习惯在固定观测点记录增加时间间隔维度连续多次完全一致的坐标才计分调用 LLM 延迟高导致审核链路超时每个记录都调用一次模型吞吐上不去增加异步队列或只在规则风险分超过一定阈值时才调用 LLM数据集被污染后才意识到问题缺少事后审计机制建立定期抽样复核流程对低风险记录也做随机人工抽查6. 最佳实践与工程建议6.1 数据可信度评分机制建议平台对每一条记录维护一个可信度评分而不是简单的“通过/不通过”。评分可以由多个维度组成用户历史可信度、记录字段完整度、照片证据质量、规则引擎评分、社区复核结果。可信度评分的好处在于研究者可以按评分阈值过滤数据。比如只使用可信度高于 0.8 的记录做物种分布建模或者在论文中明确说明数据分析包含低可信度记录但会附加权重。这比二值化的“通过/拒绝”更精细。6.2 LLM 使用的工程规范在审计链路中使用 LLM需要注意几个问题。一是成本控制。每条记录都调用大模型会带来不小的费用开销更经济的方式是让规则引擎先过滤一遍只有规则评分超过阈值或者特征可疑的记录才进入 LLM 审计。这样大部分正常记录走轻量路径少量可疑记录走深度路径。二是确定性。LLM 的输出有随机性同一输入多次调用可能得到不同结果。审计场景要求可复现性所以 temperature 应设置为 0同时记录每次调用的模型版本和 prompt 版本方便回溯。三是模型切换的影响。如果你换了一个模型审计标准可能发生变化。建议在切换前用一批标注过的历史样本做回归测试确保新模型的判断不会系统性偏离旧模型。四是数据延迟。LLM 审计不能以同步阻塞的方式接入主链路除非你对耗时有充分预期。更常见的做法是记录先入库异步调用 LLM 审计后更新风险字段。6.3 异常处理与降级策略LLM 服务不可用是常见问题。当模型 API 超时或返回异常时系统应该自动降级为“仅规则引擎”模式。虽然审计能力变弱但至少不会阻断正常记录入库。超时重试要加退避策略避免模型服务异常时大量请求堆积。建议设置超时时间为 15 秒到 30 秒重试次数不超过 2 次。另外对于 LLM 返回的字段一定做兜底校验。例如risk_score可能超出 0-100 范围reasons可能为空数组。审计系统要能容忍这种异常不能因为模型输出不符合 Pydantic 模型就直接抛异常。6.4 安全边界与隐私保护公民科学数据中可能包含个人信息比如用户 ID、上传设备的元数据、GPS 坐标。在处理这些数据时要注意隐私合规。在日志记录与审计数据集中建议进行脱敏处理特别是 GPS 坐标可以模糊化到一定精度之后再用于分析。还有一点容易被忽略如果审计结果被用于封禁用户需要提供申诉渠道。因为 LLM 审计本质上是一个概率判断它可能存在误判。直接根据 AI 审计结果封禁用户在管理层面和用户信任层面都可能引发问题。更稳妥的做法是走“人工复核后处理”。7. 总结与下一步到这里我们已经完整梳理了生成式 AI 对公民科学记录的威胁并搭建了一个最小可运行的审计系统。它同时具备规则引擎的确定性、LLM 的语义理解能力和人工审核的兜底机制可以在一定程度上缓解数据污染风险。下一步可以做的事情有很多把规则引擎的特征维度扩展得更丰富比如接入物种分布数据库做地理分布合理性校验。为图像鉴定引入独立的取证模块通过 EXIF 元数据、照片噪声模式、生成痕迹检测等方法覆盖图像型假证据。为“批量自动化提交”场景引入更完善的设备指纹和行为序列模型识别代理 IP 和账号池行为。构建一份公开的“虚假观测记录”测试集帮助更多平台验证自己的审计算法。如果你也在维护公民科学数据平台或者正在研究如何用 LLM 做数据质量治理欢迎在评论区交流思路。数据可信度是数据科学的地基在生成式 AI 时代这件事比以往更需要被认真对待。