Harness Engineering：为什么 90% 的 AI 项目死在从 Demo 到生产的路上

harness_doc_analyzer/├── config/│   ├── prompts.yaml       # Prompt 模板配置│   └── workflow.yaml      # 工作流定义├── src/│   ├── orchestrator.py    # 工作流编排引擎│   ├── context_manager.py # 上下文管理│   ├── quality_checker.py# 质量检查器│   └── security_guard.py  # 安全护栏├── tests/│   ├── test_workflow.py   # 工作流测试│   └── test_security.py  # 安全测试└── monitoring/    └── dashboard.json     # 监控仪表板

# 定义文档分析的工作流name: "智能文档分析"version: "1.0"steps:  - id: "validate_input"    type: "validation"    config:      max_length: 50000      allowed_types: ["pdf", "docx", "txt"]      security_scan: true  - id: "extract_context"    type: "context"    config:      strategy: "sliding_window"      max_tokens: 4000      overlap: 200  - id: "analyze"    type: "llm"    depends_on: ["validate_input", "extract_context"]    config:      model: "gpt-4-turbo"      prompt_template: "doc_analysis_v2"      temperature: 0.3      max_retries: 3  - id: "quality_check"    type: "quality"    depends_on: ["analyze"]    config:      min_confidence: 0.8      check_hallucination: true      fallback_action: "retry_with_stricter_prompt"  - id: "output"    type: "response"    depends_on: ["quality_check"]

class ContextManager:    def __init__(self, max_tokens=4000):        self.max_tokens = max_tokens        self.session_store = {}  # Redis 或数据库    def build_context(self, session_id, new_message):        # 获取历史对话        history = self.session_store.get(session_id, [])        # 滑动窗口策略：保留最近的对话，超出则压缩        context_tokens = self._count_tokens(history + [new_message])        if context_tokens > self.max_tokens:            # 压缩策略：保留关键信息，摘要早期对话            history = self._compress_history(history)        # 添加元数据：用户偏好、任务类型等        metadata = self._get_session_metadata(session_id)        return {            "history": history,            "metadata": metadata,            "current_message": new_message        }    def _compress_history(self, history):        # 使用 LLM 摘要早期对话，保留关键信息        # 实现细节省略...        pass

class QualityChecker:    def evaluate(self, response, context):        scores = {}        # 1. 置信度评分        scores["confidence"] = self._check_confidence(response)        # 2. 幻觉检测：要求 AI 标注不确定内容        scores["hallucination"] = self._detect_hallucination(response, context)        # 3. 一致性检查：与已知事实对比        scores["consistency"] = self._check_consistency(response, context.facts)        # 4. 完整性检查：是否回答了所有问题        scores["completeness"] = self._check_completeness(response, context.query)        # 综合评分        overall_score = sum(scores.values()) / len(scores)        if overall_score < 0.8:            return {                "pass": False,                "action": "retry",                "reason": f"质量评分 {overall_score:.2f} < 0.8",                "scores": scores            }        return {            "pass": True,            "scores": scores        }

class SecurityGuard:    def validate_input(self, user_input):        # 1. Prompt 注入检测        injection_patterns = [            "忽略之前的指令",            "system:",            "developer mode",            "绕过所有限制"        ]        for pattern in injection_patterns:            if pattern.lower() in user_input.lower():                raise SecurityException("检测到 Prompt 注入攻击")        # 2. 敏感信息检测        if self._contains_sensitive_info(user_input):            raise SecurityException("输入包含敏感信息")        return True    def filter_output(self, ai_response):        # 过滤可能泄露系统提示的内容        # 过滤可能的有害建议        pass

📂 your-ai-project/├── 📄 Agent.md <-- 核心：定义工作流、上下文、质量、安全、观测├── 📂 config/├── 📂 src/└── 📂 tests/

# Agent.md — Harness Engineering 落地规范 v1.0## 1. 项目 AI 能力定位- 核心任务：[文档摘要/客服问答/代码生成]- 预期 SLA：响应时间 < 2s，可用性 99.9%- 风险等级：[低/中/高]（涉及 PII 或金融建议则强制安全审计）## 2. 工作流编排规范- 所有 LLM 调用必须通过 `workflow.yaml` 定义步骤（validate → context → llm → quality → output）- 每个步骤必须声明 `depends_on` 和重试策略（max_retries=3）- 禁止在业务代码中直接调用 LLM SDK，必须经过 Orchestrator## 3. 上下文管理策略- 会话存储：Redis (TTL=30min) / 数据库持久化- 窗口大小：max_tokens=4000，超出采用摘要压缩（调用 GPT-3.5-turbo 生成上轮摘要）- 必须注入会话元数据：user_id, session_type, 上一次意图## 4. 质量控制门禁- 输出必须通过 `QualityChecker` 评估，综合分 ≥ 0.8 才可返回用户- 强制检测项：置信度、幻觉率（使用 SelfCheckGPT 风格）、完整性- 低质量输出的 fallback：返回"我暂时无法确定，请转人工" + 记录到 slow queue## 5. 安全护栏 (必须实现)- 输入层：正则 + 模型检测 prompt 注入模式（如"忽略指令"、"system prompt"）- 输出层：过滤身份证、银行卡、API key 等敏感信息- 每周运行一次 red-team 测试（用 Garak 或内部注入脚本）## 6. 可观测性与成本控制- 埋点指标：llm_duration_ms, step_error_rate, input_tokens, output_tokens, 质量评分分布- 仪表盘：Grafana / Datadog，告警阈值：错误率 >5% 或单日成本超预算 20%- 缓存策略：对相同或相似 query 启用语义缓存（Redis + 向量相似度 >0.95）## 7. 测试要求- 单元测试：每个 Harness 组件（context_manager, quality_checker）- 集成测试：完整工作流 + 注入攻击测试用例- 回归测试：每次 prompt 变更必须跑 golden dataset（准确率不低于基线）## 8. 变更与 review 流程- 任何 prompt 修改、工作流调整必须更新 Agent.md 并经过另一位工程师 + AI 安全负责人审核- CI 流水线中增加 `lint-agent` 步骤：校验 Agent.md 与代码实现是否一致

from your_project import QualityChecker, SecurityGuard@app.post("/chat")async def chat(request: Request):    # 1. 安全护栏    SecurityGuard().validate_input(request.message)    # 2. 调用 LLM (通过已有工作流)    raw_reply = await llm_chain.run(request.message)    # 3. 质量检查 (新加一行，立刻生效)    quality = QualityChecker().evaluate(raw_reply, context)    if not quality["pass"]:        return {            "reply": "系统正在升级，请稍后重试",            "fallback": True        }    return {        "reply": raw_reply    }