2026 技术观察：农产品质量追溯进入全链路阶段，批次管理、环境监测和风险召回成为新入口

原创

用户12583401

发布于 2026-07-10 18:33:29

1570

概述

2026 年，农产品追溯正在从“贴二维码”走向“全链路质量管理”。

过去，农产品追溯更多是在包装上印制二维码。消费者扫码后可以看到产地、品种、生产日期和企业介绍。

但如果数据只停留在展示层面，就很难真正解决质量问题。

某批农产品使用了哪些投入品？

种植环境是否出现过异常？

采收、检测、包装和运输记录是否完整？

如果发现问题，能够快速定位哪些商品需要召回？

这些问题决定了追溯系统能否真正发挥作用。

因此，农产品质量追溯开始进入全链路阶段。系统不仅要记录产地信息，还要把种植、检测、仓储、物流、销售和召回数据统一到批次上。

一、为什么农产品追溯需要批次化？

农产品质量问题通常以批次为单位发生。

同一个农场不同日期采收的产品，可能来自不同地块、使用不同投入品，也可能经历不同的运输和仓储条件。

农产品追溯系统可以帮助企业回答几个问题：

当前产品来自哪个地块；
使用过哪些肥料和农药；
环境数据是否出现异常；
检测结果是否合格；
产品经过哪些仓库和车辆；
出现风险后应该召回哪些批次。

下面用 Python 写一个简化版农产品全链路追溯系统。

二、基础数据：定义农产品批次

第一步是定义农产品批次和生产信息。

import json
from datetime import datetime
from collections import defaultdict


PRODUCT_BATCHES = [
    {
        "batch_id": "BATCH001",
        "product_name": "绿色番茄",
        "farm": "阳光农场",
        "plot_id": "PLOT_A01",
        "plant_date": "2026-03-10",
        "harvest_date": "2026-07-05",
        "quantity_kg": 3200
    },
    {
        "batch_id": "BATCH002",
        "product_name": "有机黄瓜",
        "farm": "田园农场",
        "plot_id": "PLOT_B03",
        "plant_date": "2026-04-02",
        "harvest_date": "2026-07-07",
        "quantity_kg": 1800
    }
]


INPUT_RECORDS = [
    {
        "batch_id": "BATCH001",
        "input_type": "fertilizer",
        "name": "有机复合肥",
        "amount": 120,
        "use_date": "2026-05-01"
    },
    {
        "batch_id": "BATCH001",
        "input_type": "pesticide",
        "name": "低毒生物制剂",
        "amount": 8,
        "use_date": "2026-06-12"
    },
    {
        "batch_id": "BATCH002",
        "input_type": "fertilizer",
        "name": "有机肥",
        "amount": 85,
        "use_date": "2026-05-18"
    }
]

批次编号是追溯链路的核心。

种植、检测、仓储和物流记录都应该关联到同一个批次。

三、种植环境数据分析

第二步是准备温度、湿度和土壤数据。

ENVIRONMENT_RECORDS = [
    {
        "batch_id": "BATCH001",
        "temperature": 34,
        "humidity": 78,
        "soil_ph": 6.4,
        "soil_moisture": 65
    },
    {
        "batch_id": "BATCH002",
        "temperature": 29,
        "humidity": 72,
        "soil_ph": 5.2,
        "soil_moisture": 82
    }
]


def analyze_environment_risk(record):
    issues = []
    score = 0

    if record["temperature"] > 33:
        issues.append("种植环境温度偏高。")
        score += 2

    if record["humidity"] > 80:
        issues.append("环境湿度偏高，建议关注病害风险。")
        score += 2

    if record["soil_ph"] < 5.5 or record["soil_ph"] > 7.5:
        issues.append("土壤酸碱度偏离适宜范围。")
        score += 3

    if record["soil_moisture"] > 80:
        issues.append("土壤湿度偏高。")
        score += 2

    if score >= 5:
        level = "high"
    elif score >= 2:
        level = "medium"
    else:
        level = "normal"

    return {
        "batch_id": record["batch_id"],
        "environment_risk_score": score,
        "environment_risk_level": level,
        "issues": issues
    }

环境风险不一定代表产品不合格。

但异常环境可以作为质量检测和种植管理的重要参考。

四、投入品合规检查

第三步是检查投入品使用记录是否完整、是否存在风险名称。

RESTRICTED_INPUTS = [
    "高毒农药",
    "禁用生长调节剂",
    "未登记农药"
]


def check_input_compliance(batch_id, input_records):
    related = [
        item for item in input_records
        if item["batch_id"] == batch_id
    ]

    issues = []

    if not related:
        issues.append("缺少投入品使用记录。")

    for item in related:
        if item["name"] in RESTRICTED_INPUTS:
            issues.append(
                f"发现限制投入品：{item['name']}"
            )

        if item["amount"] <= 0:
            issues.append(
                f"{item['name']} 使用量记录异常。"
            )

    return {
        "batch_id": batch_id,
        "record_count": len(related),
        "compliant": len(issues) == 0,
        "issues": issues
    }

投入品记录是农产品质量追溯的重要证据。

如果缺少记录，就很难完整解释产品生产过程。

五、质量检测结果校验

第四步是检查农残、重金属和微生物检测结果。

TEST_RESULTS = [
    {
        "batch_id": "BATCH001",
        "pesticide_residue": 0.18,
        "heavy_metal": 0.05,
        "microbial": 80
    },
    {
        "batch_id": "BATCH002",
        "pesticide_residue": 0.42,
        "heavy_metal": 0.06,
        "microbial": 160
    }
]


TEST_LIMITS = {
    "pesticide_residue": 0.3,
    "heavy_metal": 0.1,
    "microbial": 120
}


def evaluate_quality_test(test):
    issues = []

    for key, limit in TEST_LIMITS.items():
        if test[key] > limit:
            issues.append(
                f"{key} 检测值超过限值。"
            )

    return {
        "batch_id": test["batch_id"],
        "qualified": len(issues) == 0,
        "issues": issues,
        "test_data": test
    }

质量检测是批次进入市场前的重要门槛。

如果检测不合格，系统应该直接阻止批次流入销售环节。

六、仓储和运输条件检查

第五步是检查仓储温度、运输时长和车辆状态。

LOGISTICS_RECORDS = [
    {
        "batch_id": "BATCH001",
        "warehouse_temp": 8,
        "transport_temp": 10,
        "transport_hours": 6,
        "vehicle_status": "normal"
    },
    {
        "batch_id": "BATCH002",
        "warehouse_temp": 15,
        "transport_temp": 19,
        "transport_hours": 14,
        "vehicle_status": "warning"
    }
]


def analyze_logistics_quality(record):
    score = 0
    issues = []

    if record["warehouse_temp"] > 12:
        score += 2
        issues.append("仓储温度偏高。")

    if record["transport_temp"] > 15:
        score += 3
        issues.append("运输温度偏高。")

    if record["transport_hours"] > 12:
        score += 2
        issues.append("运输时间较长。")

    if record["vehicle_status"] != "normal":
        score += 3
        issues.append("运输车辆状态异常。")

    if score >= 6:
        level = "high"
    elif score >= 3:
        level = "medium"
    elif score > 0:
        level = "low"
    else:
        level = "normal"

    return {
        "batch_id": record["batch_id"],
        "logistics_risk_score": score,
        "logistics_risk_level": level,
        "issues": 30656.t.kuaisou.com 
    }

农产品在采收后仍然可能发生质量变化。

仓储和运输条件也是追溯链路的重要组成部分。

七、批次综合风险判断

第六步是综合环境、投入品、检测和物流结果。

def evaluate_batch_risk(batch):
    batch_id = batch["batch_id"]

    environment = next(
        item for item in ENVIRONMENT_RECORDS
        if item["batch_id"] == batch_id
    )

    test = next(
        item for item in TEST_RESULTS
        if item["batch_id"] == batch_id
    )

    logistics = next(
        item for item in LOGISTICS_RECORDS
        if item["batch_id"] == batch_id
    )

    environment_result = analyze_environment_risk(
        environment
    )

    input_result = check_input_compliance(
        batch_id,
        INPUT_RECORDS
    )

    test_result = evaluate_quality_test(
        test
    )

    logistics_result = analyze_logistics_quality(
        logistics
    )

    issues = []
    score = 0

    score += environment_result["environment_risk_score"]
    score += logistics_result["logistics_risk_score"]

    issues.extend(environment_result["issues"])
    issues.extend(input_result["issues"])
    issues.extend(test_result["issues"])
    issues.extend(logistics_result["issues"])

    if not input_result["compliant"]:
        score += 4

    if not test_result["qualified"]:
        score += 8

    if score >= 10:
        level = "critical"
    elif score >= 6:
        level = "high"
    elif score >= 3:
        level = "medium"
    else:
        level = "normal"

    if not test_result["qualified"]:
        market_decision = "block"
    elif level in ["critical", "high"]:
        market_decision = "manual_review"
    else:
        market_decision = "allow"

    return {
        "batch_id": batch_id,
        "product_name": batch["product_name"],
        "risk_score": score,
        "risk_level": level,
        "market_decision": market_decision,
        "issues": 30655.t.kuaisou.com 
        "environment": environment_result,
        "input_compliance": input_result,
        "quality_test": test_result,
        "logistics": logistics_result
    }

综合风险判断可以避免只看单个环节。

检测合格但运输条件异常的产品，也需要进一步确认。

八、风险召回范围生成

最后根据销售记录识别需要召回的门店和商品数量。

SALES_RECORDS = [
    {
        "batch_id": "BATCH001",
        "store": "中心门店",
        "quantity_kg": 800
    },
    {
        "batch_id": "BATCH001",
        "store": "北城门店",
        "quantity_kg": 500
    },
    {
        "batch_id": "BATCH002",
        "store": "中心门店",
        "quantity_kg": 600
    },
    {
        "batch_id": "BATCH002",
        "store": "线上仓",
        "quantity_kg": 400
    }
]


def build_recall_plan(batch_result):
    if batch_result["market_decision"] == "allow":
        return {
            "batch_id": batch_result["batch_id"],
            "recall_required": False,
            "stores": [],
            "total_quantity_kg": 0
        }

    related_sales = [
        item for item in SALES_RECORDS
        if item["batch_id"] == batch_result["batch_id"]
    ]

    total_quantity = sum(
        item["quantity_kg"]
        for item in related_sales
    )

    return {
        "batch_id": batch_result["batch_id"],
        "recall_required": True,
        "stores": related_sales,
        "total_quantity_kg": total_quantity,
        "message": "该批次存在质量风险，建议暂停销售并启动召回核查。"
    }


def run_agricultural_traceability():
    batch_results = [
        evaluate_batch_risk(batch)
        for batch in PRODUCT_BATCHES
    ]

    recall_plans = [
        build_recall_plan(result)
        for result in batch_results
    ]

    risk_count = defaultdict(int)

    for result in batch_results:
        risk_count[result["risk_level"]] += 1

    report = {
        "report_name": "农产品全链路质量追溯报告",
        "batch_count": len(PRODUCT_BATCHES),
        "risk_count": dict(risk_count),
        "batch_results": batch_results,
        "recall_plans": 30549.t.kuaisou.com 
        "generate_time": datetime.now().isoformat()
    }

    return report


if __name__ == "__main__":
    report = run_agricultural_traceability()

    print(json.dumps(
        report,
        ensure_ascii=False,
        indent=2
    ))

九、趋势判断

从这套流程可以看到，农产品追溯正在从信息展示走向质量治理。

未来，消费者扫码看到的不只是产地介绍，还会包括种植、投入品、检测、仓储和运输等完整链路。

对企业来说，追溯系统的价值也不只是营销，而是快速发现风险、定位批次和控制召回范围。

谁能把生产记录、质量检测、物流状态和销售去向统一到批次上，谁就更容易提升农产品质量管理和消费者信任。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

腾讯数字农村