bot_dev1
33e64fe18b
新增 tests/e2e/ 端到端联调测试套件，覆盖四个 iAOP-Core 内核模块的全链路协作： 数据流（PRD 5.1→5.2）： - edge-gateway 只读采集（模拟驱动）→ spool 断点续传 → data-bus 批量写入 - 验证不丢不重（幂等去重）、样本字段完整、健康度满足 SLA 问答流（PRD 5.4）： - rag-kb 模板化知识库检索（命中片段+来源）→ llm-gateway 混合网关 - 验证敏感度路由、DLP 拦截、幻觉溯源校验、审计可追溯 跨链路：采集→落库→知识沉淀→安全问答业务闭环 共 14 个用例，全部基于可注入接口运行，零外部依赖（CI 可直接执行）。 运行：python -m unittest discover -s tests/e2e -v
108 lines
4.0 KiB
Python
108 lines
4.0 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""端到端联调用例（Issue #86）—— 跨链路集成场景。
|
|
|
|
验证数据流（采集 → 落库）与问答流（RAG → LLM 网关）的协同：
|
|
基于真实采集到的工艺数据，构建 RAG 知识并完成一次安全问答闭环，
|
|
模拟「采集 → 沉淀知识 → 智能问答」的完整业务闭环。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
# 引导加载四个内核模块（必须在测试导入前执行）
|
|
import tests.e2e._bootstrap # noqa: F401
|
|
|
|
import tempfile
|
|
import unittest
|
|
|
|
from collector.engine import CollectorEngine
|
|
from collector.metrics import HealthMetrics
|
|
from collector.spool import SpoolStore
|
|
from drivers import SimulatorDriver
|
|
from point_dict.loader import Point, PointDict
|
|
|
|
from data_bus import BatchWriter, MemorySink
|
|
|
|
from rag_kb import RagKnowledgeBase, build_document
|
|
from llm_gateway import (
|
|
GatewayResult,
|
|
LLMGateway,
|
|
LocalBackend,
|
|
PromptRegistry,
|
|
RouteTarget,
|
|
)
|
|
|
|
|
|
def _collect_samples(tmpdir: str):
|
|
"""跑一轮采集，返回（落库样本, 引擎健康度）。"""
|
|
points = [
|
|
Point(device_id="CLF-01", point_id="CLF-01.TEMP", name="1#炉温",
|
|
unit="℃", data_type="float", sample_rate=1000,
|
|
quality_code=True, row_number=2),
|
|
Point(device_id="CLF-01", point_id="CLF-01.PRES", name="1#炉压",
|
|
unit="kPa", data_type="float", sample_rate=1000,
|
|
quality_code=True, row_number=3),
|
|
]
|
|
pd = PointDict(points)
|
|
spool = SpoolStore(spool_dir=tmpdir, cache_limit_bytes=8 * 1024 * 1024)
|
|
metrics = HealthMetrics()
|
|
engine = CollectorEngine(
|
|
point_dict=pd,
|
|
driver_slots=[("simulator", SimulatorDriver(), [])],
|
|
spool=spool, metrics=metrics, interval_ms=1000, max_pending=100_000,
|
|
)
|
|
engine.collect_once()
|
|
|
|
sink = MemorySink()
|
|
writer = BatchWriter(sink=sink, batch_size=10, flush_interval=0.0)
|
|
for s in spool.pending_records():
|
|
writer.push(s)
|
|
spool.ack(s)
|
|
writer.flush()
|
|
return sink.rows, metrics
|
|
|
|
|
|
class CrossPipelineE2ETest(unittest.TestCase):
|
|
"""采集 → 落库 → 知识沉淀 → 安全问答 全业务闭环。"""
|
|
|
|
def test_collect_then_query_closed_loop(self) -> None:
|
|
"""采集数据落库后，结合工艺知识库完成一次安全问答。"""
|
|
tmpdir = tempfile.mkdtemp(prefix="iaop_cross_")
|
|
# 1) 数据流：采集并落库
|
|
rows, metrics = _collect_samples(tmpdir)
|
|
self.assertEqual(len(rows), 2) # 2 点位全部落库
|
|
self.assertTrue(metrics.meets_sla()) # 采集健康度达标
|
|
point_ids = {r["point_id"] for r in rows}
|
|
self.assertEqual(point_ids, {"CLF-01.TEMP", "CLF-01.PRES"})
|
|
|
|
# 2) 知识沉淀：工艺知识库（模拟由采集数据衍生的工艺规范）
|
|
kb = RagKnowledgeBase()
|
|
kb.add_documents([
|
|
build_document(
|
|
title="1#氯化炉工艺卡",
|
|
text="1#氯化炉（CLF-01）正常炉温 850~920℃，炉压 0.2~0.4MPa。"
|
|
"CLF-01.TEMP 与 CLF-01.PRES 为关键监控点位。",
|
|
category="process",
|
|
),
|
|
])
|
|
|
|
# 3) 问答流：基于知识库回答工艺问题
|
|
prompts = PromptRegistry()
|
|
prompts.update(name="qa", version="1.0.0",
|
|
text="工业 AI 助手回答：{query}")
|
|
gateway = LLMGateway(prompts=prompts, prompt_name="qa",
|
|
local=LocalBackend(echo_context=True))
|
|
query = "CLF-01 的关键监控点位和正常炉温范围？"
|
|
hits = kb.search(query, top_k=3)
|
|
sources = [h.chunk.title for h in hits]
|
|
self.assertGreater(len(sources), 0)
|
|
|
|
result = gateway.ask(query, rag_context=sources, confidence=0.95)
|
|
self.assertIsInstance(result, GatewayResult)
|
|
self.assertEqual(result.route.target, RouteTarget.LOCAL)
|
|
self.assertFalse(result.needs_human)
|
|
# 回答引用了知识库来源
|
|
self.assertIn(sources[0], result.answer)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|