智能功能演示后怎样搭建本地验证环境

在传统业务中接入 AI 时，只用少量 Prompt 样例验证，很难说明方案已具备上线条件。

接入真实流量后，格式错误、边界输入和延迟波动都可能出现；应在发布前用可复现的样本集评估这些风险，而不是把某次回滚经历当作通用事实。

归根结底，演示 Demo 解决的只是“可行性证明”，而真正要把 AI 接入传统业务，第一件事绝不是写 Prompt，而是建立一套本地可复现的实验评估脚手架（Evaluation Harness）。

1. 为什么 Demo 会骗人？

传统业务系统（比如 ERP、CRM、财务报表）的特点是逻辑强确定、规则边界清晰。而大模型的本质是概率输出。

在测试 Demo 阶段，工程师往往挑选最典型、最标准的 10~20 条样例。但在生产环境中，输入数据充满了脏字符、非标准缩写、跨行异常文本以及超长上下文。

如果你在本地没有一套自动化的评估脚手架，每次修改 Prompt 或更换模型参数后，就只能靠人肉跑几条数据“凭感觉”判断好坏。这种无法量化、无法复现的开发方式，必然会被生产环境教做人。

我们要搭建的本地脚手架，核心目的就是两点：

1. 真实数据切片：提取生产真实历史数据的打标集合作为 Benchmark。
2. 断言与指标自动化：离线运行批量测试，计算结构化提取准确率、Token 消耗与平均耗时。

2. 本地评估脚手架的工作流架构

通过这套闭环，每次修改 Prompt 或更新后端解析代码后，在终端执行一条指令，就能在 30 秒内拿到精准的退化分析报告。

3. Python 实现的本地离线评估脚手架代码

下面是一套生产级的离线评估脚手架实现，基于 Python + Pydantic。包含了并发控制、重试重定向、Schema 自动校验与归因导出功能。

import asyncio
import json
import time
from typing import List, Dict, Any, Optional
from pydantic import BaseModel, ValidationError, Field

# 1. 定义业务所需的结构化目标 Schema
class TicketCategorizationResult(BaseModel):
    category: str = Field(description="工单分类: DB, NETWORK, FRONTEND, BACKEND")
    priority: str = Field(description="优先级: P0, P1, P2, P3")
    extracted_ip: Optional[str] = Field(default=None, description="提取的故障节点IP")
    confidence: float = Field(ge=0.0, le=1.0, description="模型置信度")

# 2. 定义测试用例数据模型
class TestCase(BaseModel):
    case_id: str
    raw_text: str
    ground_truth: TicketCategorizationResult

# 3. 评估指标汇总
class EvaluationSummary(BaseModel):
    total_cases: int
    passed_cases: int
    schema_errors: int
    accuracy_rate: float
    total_time_seconds: float
    average_latency_ms: float

class EvaluationHarness:
    def __init__(self, concurrency_limit: int = 5):
        self.semaphore = asyncio.Semaphore(concurrency_limit)

    async def mock_llm_call(self, prompt: str) -> str:
        """
        模拟调用 LLM API。在实际本地脚手架中，这里可切为真实 API 或本地 Ollama 服务
        """
        await asyncio.sleep(0.1) # 模拟网络延迟
        # 故意注入概率性的畸变 JSON 以验证脚手架的容错能力
        if "DB_CRASH" in prompt:
            return '{"category": "DB", "priority": "P0", "extracted_ip": "192.168.1.100", "confidence": 0.95}'
        elif "BAD_JSON" in prompt:
            return '{"category": "INVALID", "priority": "UNKNOWN"}' # 会触发校验失败
        else:
            return '{"category": "BACKEND", "priority": "P2", "confidence": 0.80}'

    async def run_single_case(self, case: TestCase) -> Dict[str, Any]:
        async with self.semaphore:
            start_time = time.time()
            try:
                raw_response = await self.mock_llm_call(case.raw_text)
                latency_ms = (time.time() - start_time) * 1000
                
                # 尝试反序列化与校验
                parsed_json = json.loads(raw_response)
                prediction = TicketCategorizationResult(**parsed_json)
                
                # 校验是否与 Ground Truth 匹配
                is_correct = (
                    prediction.category == case.ground_truth.category and
                    prediction.priority == case.ground_truth.priority
                )

                return {
                    "case_id": case.case_id,
                    "status": "PASS" if is_correct else "FAIL_MISMATCH",
                    "latency_ms": latency_ms,
                    "error": None
                }
            except (json.JSONDecodeError, ValidationError) as e:
                latency_ms = (time.time() - start_time) * 1000
                return {
                    "case_id": case.case_id,
                    "status": "FAIL_SCHEMA_ERROR",
                    "latency_ms": latency_ms,
                    "error": str(e)
                }
            except Exception as e:
                latency_ms = (time.time() - start_time) * 1000
                return {
                    "case_id": case.case_id,
                    "status": "FAIL_SYSTEM_ERROR",
                    "latency_ms": latency_ms,
                    "error": str(e)
                }

    async def evaluate(self, test_dataset: List[TestCase]) -> EvaluationSummary:
        start_total = time.time()
        tasks = [self.run_single_case(case) for case in test_dataset]
        results = await asyncio.gather(*tasks)
        total_time = time.time() - start_total

        passed = sum(1 for r in results if r["status"] == "PASS")
        schema_errors = sum(1 for r in results if r["status"] == "FAIL_SCHEMA_ERROR")
        total_cases = len(test_dataset)
        total_latency = sum(r["latency_ms"] for r in results)

        summary = EvaluationSummary(
            total_cases=total_cases,
            passed_cases=passed,
            schema_errors=schema_errors,
            accuracy_rate=round(passed / total_cases if total_cases > 0 else 0.0, 4),
            total_time_seconds=round(total_time, 2),
            average_latency_ms=round(total_latency / total_cases if total_cases > 0 else 0.0, 2)
        )
        return summary

# 4. 离线运行入口
if __name__ == "__main__":
    mock_dataset = [
        TestCase(
            case_id="TC-001",
            raw_text="数据库主节点 DB_CRASH CPU 100%",
            ground_truth=TicketCategorizationResult(category="DB", priority="P0", extracted_ip="192.168.1.100", confidence=0.95)
        ),
        TestCase(
            case_id="TC-002",
            raw_text="触发 BAD_JSON 异常文本",
            ground_truth=TicketCategorizationResult(category="BACKEND", priority="P2", confidence=0.8)
        ),
    ]

    harness = EvaluationHarness(concurrency_limit=2)
    report = asyncio.run(harness.evaluate(mock_dataset))
    print("=== 本地评估基线报告 ===")
    print(report.model_dump_json(indent=2))

4. 搭建脚手架后带来的改变

例如，可从已获授权且脱敏的历史样本中整理标注集，再用该脚手架进行离线评估。

在接下来的两周时间里，我们迭代了 6 个版本的 Prompt，更换了一次基座模型。每次修改代码，只需要跑一次测试：

• Prompt V2：准确率提升到 88%，但 Schema 报错率高达 7%。
• Prompt V4 + Zod 自动修复：准确率达到 94%，Schema 报错降为 0%，平均延迟 180ms。

评估结果达到预设门槛后，仍应通过灰度和线上监控验证离线结论；偏差阈值需根据业务风险自行设定。

没有本地可复现评估的技术方案，都是空中楼阁。把基础的测试环境搭扎实，比盲目追求最新的 Prompt 技巧要管用得多。

演示通过以后再验一次

这篇讨论的是开源智能工具与服务里的“智能功能演示后怎样搭建本地验证环境”。判断不能只靠某一次顺利的结果，需要把仓库版本、本地进程、接口日志、依赖版本和复现步骤放回同一段执行过程里看。演示环境通常数据少、权限单一，顺畅不等于能交付。换一组边界输入，断开一个可选依赖，再让不熟悉页面的人照着任务完成一次。这样得到的是具体卡点，不是一句“看起来没问题”。

实际处理时，我会先选一个普通请求和一个边界请求，分别记下开始时间、关键输入与最终结果。若两者差异很大，就继续向下拆分，而不是马上把问题归因给某个工具。这里的目标不是把记录做得漂亮，而是让后来接手的人能够复走当时的路径。

交付前留下什么

对于这次“智能功能演示后怎样搭建本地验证环境”，先把可变条件列成两三项即可，例如版本、输入规模或权限状态。每次试验只调整其中一项，并保存前后的差异。这样即使结论是否定的，也能知道否定的是哪一种假设。