Agent Harness 架构设计：从大 Prompt 到可测试的模块化运行时

1
Task / Goal
2
   ↓
3
Harness
4
   ├── Context
5
   ├── State
6
   ├── Planner / Router
7
   ├── Model Gateway
8
   ├── Tool Runtime
9
   ├── Guardrails / Policy
10
   ├── Checkpoint / Recovery
11
   ├── HITL
12
   └── Trace / Eval
13
   ↓
14
Verified Outcome

1
System Prompt
2
+ 所有历史消息
3
+ 所有工具描述
4
+ 所有文档
5
+ 所有业务规则
6
= Agent

1
Harness 先根据身份、状态和任务选择最小能力集合，
2
再构建最小充分上下文，
3
让模型只在被授权的候选动作内做局部决策，
4
最后由确定性系统执行、验证和记录结果。

1
Control Plane
2
  Agent/Prompt/Tool/Policy 版本、模型路由、权限、灰度、预算
3

4
Data & Execution Plane
5
  Run 调度、节点执行、模型调用、工具调用、Checkpoint、队列
6

7
Evidence Plane
8
  Trace、Audit、Evaluation、Failure Sample、Replay、Cost

1
{
2
  "run_id": "run_01J...",
3
  "agent_version": "orderflow@3.2.0",
4
  "workflow_version": "refund@7",
5
  "prompt_bundle_version": "2026-07-15.2",
6
  "tool_registry_version": "42",
7
  "policy_bundle_version": "refund-cn@18",
8
  "model_route_version": "support-default@9"
9
}

1
任务闭环：理解 -> 决策 -> 行动 -> 验证 -> 交付
2
状态闭环：读取 -> 条件迁移 -> 持久化 -> 恢复
3
安全闭环：识别主体 -> 授权 -> 策略 -> 审计
4
改进闭环：观察 -> 归因 -> 数据集 -> 评估 -> 发布

1
Run
2
└── Step: query_order
3
    ├── Attempt 1
4
    │   └── Tool Call: timeout
5
    └── Attempt 2
6
        └── Tool Call: success

1
created
2
  -> queued
3
  -> running
4
     -> awaiting_input
5
     -> awaiting_confirmation
6
     -> awaiting_human
7
     -> retry_scheduled
8
     -> succeeded
9
     -> failed
10
     -> cancelled
11
     -> budget_exhausted

1
from datetime import datetime
2
from typing import Literal
3
from pydantic import BaseModel, Field
4

5

6
class BudgetState(BaseModel):
7
    steps_used: int = 0
8
    model_calls_used: int = 0
9
    tool_calls_used: int = 0
10
    input_tokens_used: int = 0
11
    output_tokens_used: int = 0
12
    cost_usd: float = 0.0
13

14

15
class AgentState(BaseModel):
16
    schema_version: Literal["2"] = "2"
17
    run_id: str
18
    thread_id: str
19
    tenant_id: str
20
    user_id: str
21
    status: Literal[
22
        "created",
23
        "queued",
24
        "running",
25
        "awaiting_input",
26
        "awaiting_confirmation",
27
        "awaiting_human",
28
        "succeeded",
29
        "failed",
30
        "cancelled",
31
        "budget_exhausted",
32
    ]
33
    current_node: str
34
    intent: str | None = None
35
    slots: dict[str, str] = Field(default_factory=dict)
36
    fact_refs: list[str] = Field(default_factory=list)
37
    evidence_refs: list[str] = Field(default_factory=list)
38
    pending_action_ref: str | None = None
39
    completed_steps: list[str] = Field(default_factory=list)
40
    budget: BudgetState = Field(default_factory=BudgetState)
41
    state_version: int = 0
42
    updated_at: datetime

1
进程重启后恢复下一步是否必需？
2
  是 -> State 或稳定引用
3
  否 -> Trace/临时缓存
4

5
它是否是业务真相？
6
  是 -> 业务系统保存，State 只放 ID + version/hash

1
class StateTransition(BaseModel):
2
    transition_id: str
3
    run_id: str
4
    from_version: int
5
    from_status: str
6
    to_status: str
7
    node: str
8
    event_ref: str
9
    patch: list[dict]
10
    occurred_at: datetime

1
UPDATE agent_runs
2
SET state = :new_state,
3
    state_version = state_version + 1
4
WHERE run_id = :run_id
5
  AND state_version = :expected_version;

1
class ContextItem(BaseModel):
2
    item_id: str
3
    kind: Literal[
4
        "instruction",
5
        "task_state",
6
        "conversation",
7
        "tool_fact",
8
        "retrieved_evidence",
9
        "memory",
10
    ]
11
    trust: Literal["system", "verified", "untrusted"]
12
    source_ref: str
13
    content: str
14
    token_estimate: int
15
    effective_at: datetime | None = None
16
    expires_at: datetime | None = None
17

18

19
class ContextPackage(BaseModel):
20
    package_id: str
21
    policy_version: str
22
    items: list[ContextItem]
23
    total_token_estimate: int
24
    omitted_item_refs: list[str]

1
1. 固定系统与业务边界
2
2. 当前目标和结构化状态
3
3. 已验证的最新工具事实
4
4. 与当前节点相关的规则证据
5
5. 最少必要对话片段/摘要
6
6. 经过写入策略治理的记忆
7
7. 保留输出与工具调用预算

1
Context Precision
2
Required Fact Recall
3
Stale Context Rate
4
Unauthorized Context Rate
5
Instruction/Data Separation Pass Rate
6
Tokens per Successful Step

1
Router：选择进入哪个能力域/子图
2
Planner：提出完成目标的步骤或下一动作
3
Executor：在策略、权限和状态约束下执行动作

1
{
2
  "route": "refund_workflow",
3
  "candidate_capabilities": [
4
    "order.read",
5
    "logistics.read",
6
    "policy.search.refund"
7
  ],
8
  "excluded_capabilities": [
9
    "refund.create",
10
    "compensation.create"
11
  ],
12
  "reason_code": "facts_not_verified"
13
}

1
class PlanStep(BaseModel):
2
    step_id: str
3
    objective: str
4
    capability: str
5
    dependencies: list[str]
6
    success_condition: str
7
    failure_route: str
8
    max_attempts: int = 1
9

10

11
class Plan(BaseModel):
12
    plan_id: str
13
    goal: str
14
    assumptions: list[str]
15
    steps: list[PlanStep]
16
    replan_conditions: list[str]

1
class ModelRequest(BaseModel):
2
    model_call_id: str
3
    run_id: str
4
    step_id: str
5
    purpose: Literal[
6
        "route",
7
        "extract",
8
        "plan",
9
        "decide",
10
        "verify",
11
        "respond",
12
    ]
13
    context_package_ref: str
14
    tool_registry_snapshot_ref: str | None
15
    output_schema: dict | None
16
    timeout_seconds: float
17
    model_route: str

1
model.started
2
model.text.delta
3
model.tool_call.delta
4
model.output.completed
5
model.usage.completed
6
model.blocked
7
model.failed

1
原始事件是否丢失
2
工具参数增量能否正确组装
3
终止原因是否归一化
4
token/cost 是否只计一次
5
取消是否真正传播
6
未知事件是否 fail-closed 或可观测

1
LLM output: ToolProposal
2
    ↓ parse + schema
3
Harness: ActionRequest
4
    ↓ authz + policy + confirmation
5
Executor: ToolCommand
6
    ↓ adapter
7
Remote System
8
    ↓ normalize + verify
9
Harness: Observation

1
class Observation(BaseModel):
2
    tool_call_id: str
3
    tool_name: str
4
    tool_version: str
5
    status: Literal[
6
        "succeeded",
7
        "rejected",
8
        "failed_retryable",
9
        "failed_terminal",
10
        "outcome_unknown",
11
    ]
12
    result_ref: str | None
13
    error_code: str | None
14
    fact_version: str | None
15
    started_at: datetime
16
    finished_at: datetime

1
max_steps
2
max_tool_calls
3
per_tool_limit
4
duplicate_call_detector
5
no_progress_detector
6
wall_clock_deadline
7
cost_budget
8
terminal_state_checker

1
Input Guardrail
2
  检测输入风险，但不承担全部授权
3

4
Context Guardrail
5
  ACL、来源、敏感数据、指令/数据隔离
6

7
Action Guardrail
8
  主体、资源、权限、金额、频率、确认
9

10
Output Guardrail
11
  忠实性、隐私、格式、承诺边界
12

13
Runtime Guardrail
14
  步数、成本、超时、循环、沙箱

1
class PolicyDecision(BaseModel):
2
    decision: Literal["allow", "deny", "require_confirmation", "require_human"]
3
    policy_version: str
4
    reason_codes: list[str]
5
    obligations: list[str]
6
    expires_at: datetime | None

1
mask.phone
2
limit.refund_amount=5000
3
log.audit.high_risk
4
verify.order_owner
5
confirm.user.v2

1
{
2
  "run_id": "run_01J...",
3
  "decision_type": "refund_over_threshold",
4
  "request_summary": "订单超过72小时未发货，用户申请全额退款",
5
  "verified_facts": ["fact://order/123@v8", "fact://logistics/123@v3"],
6
  "policy_evidence": ["policy://refund/v18#R001"],
7
  "proposed_action": "action://run_01J/refund-1",
8
  "risk_reasons": ["amount_over_auto_limit"],
9
  "allowed_decisions": ["approve", "reject", "request_more_info"],
10
  "expires_at": "2026-07-15T12:30:00Z"
11
}

1
Pinned Versions
2
Current Typed State
3
Completed Step IDs
4
Pending External Operation
5
Retry Schedule
6
Human/Confirmation Token
7
Budget Usage
8
Last Durable Event Sequence

1
Run 创建后
2
每个有副作用动作前
3
每个工具结果归一化后
4
状态迁移后
5
暂停/人工接管前
6
最终响应前

1
at-least-once delivery
2
+ idempotent consumer
3
+ idempotency key
4
+ conditional state transition
5
+ outbox/inbox
6
+ final-state reconciliation
7
= effectively-once business outcome

1
class RecoveryDecision(BaseModel):
2
    action: Literal[
3
        "resume_next_step",
4
        "retry_attempt",
5
        "reconcile_external_state",
6
        "await_human",
7
        "fail_terminal",
8
    ]
9
    reason: str
10
    checkpoint_version: int

1
from typing import Any
2

3

4
class AgentEvent(BaseModel):
5
    schema_version: Literal["1"] = "1"
6
    event_id: str
7
    trace_id: str
8
    run_id: str
9
    thread_id: str
10
    step_id: str | None
11
    attempt_id: str | None
12
    sequence: int
13
    type: str
14
    occurred_at: datetime
15
    producer: str
16
    data: dict[str, Any]
17
    data_classification: Literal["public", "internal", "confidential", "restricted"]

1
run.created / run.started / run.paused / run.completed / run.failed
2
step.started / step.completed / step.failed
3
model.started / model.completed / model.failed
4
tool.proposed / tool.authorized / tool.completed / tool.failed
5
state.transitioned / checkpoint.saved
6
human.requested / human.decided
7
budget.warning / budget.exhausted
8
eval.queued / eval.completed

1
Client
2
  ↓
3
API Gateway / Identity / Rate Limit
4
  ↓
5
Run Service ───────────────→ Event Stream
6
  ↓                              ↑
7
StateGraph Runtime ────────→ Checkpoint Store
8
  ├── Input + Intent
9
  ├── Context Builder ─────→ Memory/RAG Ports
10
  ├── Capability Router
11
  ├── Planner / Decision
12
  ├── Policy Decision Point
13
  ├── Tool Executor ───────→ Order/Logistics/Refund Services
14
  ├── Verifier
15
  ├── HITL Gateway
16
  └── Response Composer
17
          ↓
18
Trace + Audit + Eval + Failure Collector

1
normalize_input
2
  -> identify_intent
3
  -> collect_required_slots
4
  -> query_order
5
  -> verify_order_ownership
6
  -> query_logistics
7
  -> retrieve_effective_policy
8
  -> decide_eligibility
9
  -> verify_decision
10
  -> request_confirmation?
11
  -> create_refund
12
  -> reconcile_refund_state
13
  -> compose_response

1
from typing import Protocol
2

3

4
class NodeContext(BaseModel):
5
    run_id: str
6
    step_id: str
7
    attempt_id: str
8
    deadline_at: datetime
9
    cancellation_requested: bool
10

11

12
class NodeResult(BaseModel):
13
    outcome: Literal[
14
        "completed",
15
        "retry",
16
        "pause_for_input",
17
        "pause_for_confirmation",
18
        "pause_for_human",
19
        "failed",
20
    ]
21
    state_patch: list[dict] = Field(default_factory=list)
22
    emitted_event_refs: list[str] = Field(default_factory=list)
23
    next_node: str | None = None
24
    retry_after_seconds: float | None = None
25

26

27
class AgentNode(Protocol):
28
    async def run(
29
        self,
30
        state: AgentState,
31
        context: NodeContext,
32
    ) -> NodeResult: ...

1
每次动作有结构化记录
2
每个任务有外部成功条件
3
最终状态由环境/业务系统验证
4
失败可以定位到具体 Step

1
Input normalization
2
State transition table
3
Budget calculation
4
Context selection
5
Capability filtering
6
Policy decisions
7
State patch validation
8
Event projection/redaction

1
Provider event -> ModelEvent
2
Tool adapter raw result -> Observation
3
Checkpoint old schema -> migration/new schema
4
Internal event -> SSE public event
5
Policy bundle -> PolicyDecision

1
def test_write_tool_is_unreachable_before_confirmation(graph) -> None:
2
    reachable = graph.reachable_nodes(
3
        start="identify_intent",
4
        blocked_conditions={"confirmation_granted"},
5
    )
6

7
    assert "create_refund" not in reachable

1
模型返回一半进程死亡
2
工具请求发出后进程死亡
3
工具成功但事件发布失败
4
Checkpoint 成功但 Worker 丢 lease
5
人工批准同时 Run 被取消
6
旧版本 Run 被新版本 Worker 恢复

1
{
2
  "case_id": "H-RF-017",
3
  "initial_state": {
4
    "order_status": "paid",
5
    "logistics_status": "not_shipped",
6
    "hours_since_payment": 72
7
  },
8
  "user_message": "还没发货，给我退款",
9
  "expected_required_facts": ["order", "logistics", "policy:R001"],
10
  "expected_nodes": [
11
    "identify_intent",
12
    "query_order",
13
    "verify_order_ownership",
14
    "query_logistics",
15
    "retrieve_policy",
16
    "decide_eligibility",
17
    "request_confirmation"
18
  ],
19
  "forbidden_nodes": ["create_compensation"],
20
  "expected_terminal_status": "awaiting_confirmation"
21
}

1
Task Success Rate
2
Verified Outcome Rate
3
Correct Route Rate
4
Required Fact Recall
5
Tool Selection / Argument Accuracy
6
Policy Violation Rate
7
Human Handoff Precision / Recall
8
Response Faithfulness

1
Steps per Successful Run
2
Model Calls per Successful Run
3
Tool Calls per Successful Run
4
Tokens / Cost per Successful Run
5
No-progress Loop Rate
6
Context Utilization

1
Run Success / Failure / Cancellation Rate
2
Checkpoint Recovery Success Rate
3
Duplicate Side-effect Rate
4
Unknown Outcome Reconciliation Rate
5
Retry Amplification
6
p50 / p95 / p99 Run Duration
7
Queue Wait Time

1
Trace Completeness
2
Version Attribution Coverage
3
Failure Reproduction Rate
4
State Transition Integrity
5
Audit Evidence Coverage