Harness工程六层栈：Agent量产基础设施全拆解

Agent = Model + Harness

graph TB
    subgraph "Harness六层架构"
        L6["L6 部署与运维<br/>Canary/Shadow/Rollback"]
        L5["L5 评估与测试<br/>Trajectory/Eval Gate"]
        L4["L4 沙盒与隔离<br/>E2B/Daytona/Firecracker"]
        L3["L3 编排与流程<br/>DAG/Loop/Retry"]
        L2["L2 工具与记忆<br/>Function/VectorStore"]
        L1["L1 提示与模型<br/>SystemPrompt/Model"]
    end
    L1 --> L2 --> L3 --> L4 --> L5 --> L6
    style L6 fill:#e74c3c,color:#fff
    style L5 fill:#e67e22,color:#fff
    style L4 fill:#f1c40f,color:#333
    style L3 fill:#2ecc71,color:#fff
    style L2 fill:#3498db,color:#fff
    style L1 fill:#9b59b6,color:#fff

from anthropic import Anthropic

client = Anthropic()

# L1层：通过Harness约束模型输出格式
response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[{"role": "user", "content": "分析这段代码的安全性"}],
    # Harness约束：强制JSON输出
    response_format={
        "type": "json_schema",
        "json_schema": {
            "name": "code_analysis",
            "schema": {
                "type": "object",
                "properties": {
                    "risk_level": {"type": "string", "enum": ["low", "medium", "high", "critical"]},
                    "vulnerabilities": {"type": "array", "items": {"type": "string"}},
                    "recommendation": {"type": "string"}
                },
                "required": ["risk_level", "vulnerabilities", "recommendation"]
            }
        }
    }
)

# L2层：工具定义 + 记忆管理
from langchain_core.tools import tool
from langchain_community.vectorstores import Chroma

@tool
def search_knowledge_base(query: str, top_k: int = 3) -> list[dict]:
    """搜索知识库，返回最相关的top_k条结果"""
    # 参数校验是Harness的责任，不是模型的
    if top_k > 10:
        top_k = 10  # 防止模型请求过多结果
    results = vectorstore.similarity_search(query, k=top_k)
    return [{"content": r.page_content, "score": r.metadata.get("score", 0)} for r in results]

# 记忆压缩策略
class CompressedMemory:
    def __init__(self, max_tokens=4000):
        self.max_tokens = max_tokens

    def compress(self, messages: list) -> list:
        """当上下文超过阈值时，自动压缩旧消息"""
        total = sum(len(m["content"]) for m in messages)
        if total > self.max_tokens:
            # 保留最近5条 + 压缩历史为摘要
            recent = messages[-5:]
            history = messages[:-5]
            summary = self._summarize(history)
            return [{"role": "system", "content": f"历史摘要: {summary}"}] + recent
        return messages

graph LR
    subgraph "Agent沙盒选型决策树"
        A[Agent需要执行代码?] -->|是| B{需要GPU?}
        A -->|否| C[无需沙盒<br/>纯文本交互]
        B -->|是| D[Modal<br/>GPU直通]
        B -->|否| E{需要文件系统?}
        E -->|是| F{企业合规要求?}
        E -->|否| G[Cloudflare V8<br/>最轻量]
        F -->|是| H[Daytona自建<br/>完全控制]
        F -->|否| I{延迟敏感?}
        I -->|是| J[Fly.io Sprites<br/>边缘部署]
        I -->|否| K[E2B<br/>生态最完善]
    end

# 第一层：结构化断言
def test_code_analysis_prompt():
    result = agent.analyze("def foo(): pass")
    assert result.risk_level in ["low", "medium", "high", "critical"]
    assert isinstance(result.vulnerabilities, list)
    assert len(result.recommendation) > 10

# 第二层：轨迹断言
def test_agent_trajectory():
    session = agent.run("帮我重构这个函数并写测试")

    # 检查Agent是否按正确顺序执行了正确的工具
    assert session.tool_calls[0].name == "read_file"
    assert session.tool_calls[1].name == "write_file"
    assert session.tool_calls[2].name == "run_test"

    # 检查步骤数在合理范围
    assert 3 <= len(session.tool_calls) <= 10

    # 检查没有循环（同一工具+同一参数调用>3次）
    call_signatures = [(c.name, c.args) for c in session.tool_calls]
    for sig in call_signatures:
        assert call_signatures.count(sig) <= 3

# 第三层：语义断言
async def test_refactor_workflow():
    result = await agent.run("重构src/auth.py，添加类型注解")

    # 用评判LLM检查质量
    judge_prompt = f"""
    评估以下重构结果的质量（1-5分）：
    原始代码：{result.original_code}
    重构后代码：{result.refactored_code}
    测试结果：{result.test_output}

    评估维度：类型注解完整性、功能保持、代码风格
    """
    score = await judge_llm.evaluate(judge_prompt)
    assert score >= 4.0  # 至少4分才算通过

# 第四层：回归测试
def test_regression_against_baseline():
    baseline = load_baseline("v2.3.1")  # 加载基线行为
    current = agent.run_batch(regression_test_suite)

    for test_name, (base_result, curr_result) in zip(baseline, current):
        # 语义相似度而非文本匹配
        similarity = semantic_similarity(base_result.output, curr_result.output)
        assert similarity > 0.85, f"{test_name} 回归: 相似度 {similarity:.2f}"

        # 工具调用序列一致性
        base_tools = [c.name for c in base_result.tool_calls]
        curr_tools = [c.name for c in curr_result.tool_calls]
        tool_overlap = len(set(base_tools) & set(curr_tools)) / max(len(base_tools), len(curr_tools))
        assert tool_overlap > 0.7, f"{test_name} 工具调用变化过大"

graph LR
    User[用户请求] --> Router[流量路由器]
    Router -->|"100%流量"| Old[旧版Agent v2.3]
    Router -->|"100%镜像"| New[新版Agent v2.4]
    Old --> Response[返回用户]
    New -->|静默记录| Comparator[行为比较器]
    Comparator --> Alert[差异告警]

# 金丝雀发布配置示例
canary_config = {
    "agent_v2.4": {
        "traffic_percentage": 5,  # 初始5%
        "step_up_interval": "1h",  # 每小时评估一次
        "step_up_percentage": 10,  # 每次增加10%
        "rollback_thresholds": {
            "error_rate": 0.05,       # 错误率>5%回滚
            "latency_p99_ms": 5000,   # P99延迟>5s回滚
            "tool_failure_rate": 0.1,  # 工具失败率>10%回滚
            "cost_per_request": 0.15,  # 单次成本>$0.15回滚
            "semantic_score": 3.5,     # 语义评分<3.5/5回滚
        },
        "evaluation_window": "30m",    # 评估窗口30分钟
    }
}

# eval-gate.yaml
eval_gates:
  - name: "trajectory_correctness"
    type: trajectory
    threshold: 0.85
    test_cases: 50

  - name: "tool_selection_accuracy" 
    type: tool_usage
    threshold: 0.90
    test_cases: 30

  - name: "output_quality"
    type: semantic
    judge_model: "claude-haiku-4-5-20251001"
    threshold: 4.0  # 5分制
    test_cases: 100

  - name: "cost_regression"
    type: cost
    max_increase_percentage: 20
    test_cases: 200

deployment:
  requires_all_gates: true
  auto_rollback_on_regression: true

# 功能标志控制Agent能力范围
from flags import get_flag

class AgentWithFlags:
    def __init__(self, user_id: str):
        self.user_id = user_id

    def get_tools(self):
        tools = [read_file, write_file]  # 基础工具
        if get_flag("agent.code_execution", self.user_id, default=False):
            tools.append(execute_code)  # 渐进开放
        if get_flag("agent.internet_access", self.user_id, default=False):
            tools.append(web_search)  # 高风险能力
        if get_flag("agent.database_write", self.user_id, default=False):
            tools.append(db_query)  # 最高风险
        return tools

    def get_step_budget(self):
        if get_flag("agent.extended_thinking", self.user_id, default=False):
            return 50  # 高信任用户给更多步骤
        return 20

graph LR
    subgraph "Azure AgentOps五阶段"
        D[开发<br/>Dev] --> T[测试<br/>Test] --> S[预演<br/>Staging] --> P[生产<br/>Prod] --> M[运维<br/>Operate]
    end
    D -->|代码+提示词| T
    T -->|评估套件| S
    S -->|影子流量| P
    P -->|金丝雀+回滚| M
    M -->|反馈数据| D