Agent 工程基础底座：从 Python 服务到可演进的 Runtime 骨架

1
Control Plane
2
  配置、版本、策略、权限、模型路由、发布
3

4
Execution Plane
5
  Run、节点、模型调用、工具执行、状态迁移
6

7
Evidence Plane
8
  Trace、Audit、Eval、Failure Sample、成本与延迟

1
L0 脚本：一次模型调用，结果打印到终端
2
L1 服务：HTTP API + Schema + 错误处理
3
L2 Runtime：状态、工具注册、超时、幂等、追踪
4
L3 可恢复：Checkpoint、队列、重放、人工接管
5
L4 可治理：版本、评估门禁、灰度、审计、成本预算

1
[project]
2
name = "orderflow-agent"
3
version = "0.1.0"
4
requires-python = ">=3.12"
5
dependencies = [
6
  "fastapi>=0.115,<1",
7
  "pydantic>=2.10,<3",
8
  "pydantic-settings>=2.7,<3",
9
  "uvicorn[standard]>=0.34,<1",
10
  "httpx>=0.28,<1",
11
  "sqlalchemy[asyncio]>=2.0,<3",
12
  "asyncpg>=0.30,<1",
13
  "redis>=5,<7",
14
  "opentelemetry-api>=1.30,<2",
15
]
16

17
[dependency-groups]
18
dev = [
19
  "pytest>=8,<9",
20
  "pytest-asyncio>=0.25,<1",
21
  "mypy>=1.14,<2",
22
  "ruff>=0.9,<1",
23
]

1
from pydantic import SecretStr
2
from pydantic_settings import BaseSettings, SettingsConfigDict
3

4

5
class Settings(BaseSettings):
6
    model_config = SettingsConfigDict(
7
        env_file=".env",
8
        env_nested_delimiter="__",
9
        extra="ignore",
10
    )
11

12
    environment: str = "local"
13
    database_url: SecretStr
14
    redis_url: SecretStr
15
    model_provider: str = "openai"
16
    model_name: str
17
    model_api_key: SecretStr
18
    request_timeout_seconds: float = 30.0
19
    max_agent_steps: int = 12
20
    max_run_cost_usd: float = 0.25

1
from typing import Annotated, Literal
2
from pydantic import BaseModel, Field, StringConstraints
3

4
OrderId = Annotated[
5
    str,
6
    StringConstraints(pattern=r"^ORD-[A-Z0-9]{8}$"),
7
]
8

9

10
class AgentRequest(BaseModel):
11
    request_id: str
12
    user_id: str
13
    message: Annotated[str, Field(min_length=1, max_length=8000)]
14
    order_id: OrderId | None = None
15
    response_mode: Literal["sync", "stream"] = "stream"

1
class RefundIntent(BaseModel):
2
    intent: Literal["refund_request"]
3
    order_id: OrderId | None
4
    reason: str | None
5
    confidence: Annotated[float, Field(ge=0, le=1)]
6
    needs_clarification: bool
7

8

9
class OtherIntent(BaseModel):
10
    intent: Literal["logistics_query", "complaint", "unknown"]
11
    confidence: Annotated[float, Field(ge=0, le=1)]
12

13

14
IntentResult = Annotated[
15
    RefundIntent | OtherIntent,
16
    Field(discriminator="intent"),
17
]

1
class CreateRefundArgs(BaseModel):
2
    order_id: OrderId
3
    amount_cents: Annotated[int, Field(gt=0)]
4
    reason_code: Literal[
5
        "merchant_not_shipped",
6
        "logistics_lost",
7
        "quality_issue",
8
    ]
9
    idempotency_key: str
10

11

12
class RefundCreated(BaseModel):
13
    status: Literal["created"]
14
    refund_id: str
15
    order_id: OrderId
16

17

18
class RefundRejected(BaseModel):
19
    status: Literal["rejected"]
20
    code: Literal[
21
        "permission_denied",
22
        "state_conflict",
23
        "amount_exceeded",
24
        "duplicate_request",
25
    ]
26
    retryable: bool = False

1
from collections.abc import AsyncIterator
2
from dataclasses import dataclass
3
from typing import Any, Protocol
4

5

6
@dataclass(frozen=True)
7
class ModelRequest:
8
    run_id: str
9
    messages: list[dict[str, Any]]
10
    tools: list[dict[str, Any]]
11
    output_schema: dict[str, Any] | None
12
    timeout_seconds: float
13

14

15
@dataclass(frozen=True)
16
class ModelEvent:
17
    kind: str
18
    payload: dict[str, Any]
19

20

21
class ModelGateway(Protocol):
22
    async def stream(
23
        self,
24
        request: ModelRequest,
25
    ) -> AsyncIterator[ModelEvent]: ...

1
response.started
2
text.delta
3
tool_call.started
4
tool_call.arguments.delta
5
tool_call.completed
6
usage.completed
7
response.failed

1
class ModelError(Exception):
2
    retryable: bool = False
3

4

5
class ModelRateLimited(ModelError):
6
    retryable = True
7

8

9
class ModelTimeout(ModelError):
10
    retryable = True
11

12

13
class ModelInvalidOutput(ModelError):
14
    retryable = False
15

16

17
class ModelSafetyBlocked(ModelError):
18
    retryable = False
19

20

21
class ModelContextExceeded(ModelError):
22
    retryable = False

1
任务类型
2
+ 数据敏感级别
3
+ 结构化输出能力
4
+ 工具调用能力
5
+ 延迟预算
6
+ 单 Run 成本预算
7
+ 区域与合规要求
8
+ 当前错误率/限流状态
9
= provider + model + fallback chain

1
from fastapi import APIRouter, Depends, status
2

3
router = APIRouter(prefix="/v1/agent-runs")
4

5

6
@router.post("", status_code=status.HTTP_202_ACCEPTED)
7
async def create_run(
8
    body: AgentRequest,
9
    service: "AgentRunService" = Depends(),
10
) -> dict[str, str]:
11
    run = await service.create(body)
12
    return {
13
        "run_id": run.id,
14
        "status": run.status,
15
        "events_url": f"/v1/agent-runs/{run.id}/events",
16
    }

1
{
2
  "event_id": "evt_01J...",
3
  "run_id": "run_01J...",
4
  "sequence": 17,
5
  "type": "tool.completed",
6
  "occurred_at": "2026-03-21T08:12:03.421Z",
7
  "data": {
8
    "tool_call_id": "tc_01J...",
9
    "tool_name": "query_order",
10
    "result_ref": "result://tc_01J..."
11
  }
12
}

1
Model Tool Proposal
2
  -> Parse
3
  -> Schema Validate
4
  -> Resolve Identity
5
  -> Authorize
6
  -> Policy Check
7
  -> Risk / Confirmation Check
8
  -> Idempotency Reservation
9
  -> Execute with Timeout
10
  -> Normalize Result
11
  -> Verify Business State
12
  -> Audit
13
  -> Return Observation

1
from dataclasses import dataclass
2
from enum import StrEnum
3
from typing import type
4
from pydantic import BaseModel
5

6

7
class SideEffect(StrEnum):
8
    NONE = "none"
9
    READ = "read"
10
    WRITE = "write"
11
    EXTERNAL = "external"
12

13

14
@dataclass(frozen=True)
15
class ToolSpec:
16
    name: str
17
    version: str
18
    input_model: type[BaseModel]
19
    output_model: type[BaseModel]
20
    side_effect: SideEffect
21
    permission: str
22
    timeout_seconds: float
23
    confirmation_required: bool

1
idempotency_key = hash(
2
  tenant_id,
3
  user_id,
4
  operation,
5
  business_resource_id,
6
  normalized_arguments,
7
  confirmation_version
8
)

1
reserved -> executing -> succeeded | failed_unknown | rejected

1
agent_runs 1---n run_steps
2
agent_runs 1---n tool_calls
3
agent_runs 1---n model_calls
4
agent_runs 1---n state_transitions
5
agent_runs 1---n eval_results
6
tool_calls 1---0..1 idempotency_records

1
UPDATE agent_runs
2
SET status = :next_status,
3
    version = version + 1,
4
    updated_at = now()
5
WHERE id = :run_id
6
  AND status = :expected_status
7
  AND version = :expected_version;

1
{
2
  "document_id": "policy-refund-v7",
3
  "chunk_id": "R001-C03",
4
  "tenant_id": "tenant-a",
5
  "acl": ["support", "risk"],
6
  "effective_from": "2026-01-01",
7
  "effective_to": null,
8
  "source_uri": "policy://refund/v7#R001",
9
  "content_hash": "sha256:..."
10
}

1
from pydantic import BaseModel, Field
2

3

4
class RunBudget(BaseModel):
5
    max_steps: int = Field(default=12, ge=1, le=50)
6
    max_model_calls: int = Field(default=8, ge=1)
7
    max_tool_calls: int = Field(default=12, ge=0)
8
    max_elapsed_seconds: float = Field(default=90, gt=0)
9
    max_cost_usd: float = Field(default=0.25, gt=0)

1
{
2
  "status": "needs_human",
3
  "reason": "run_budget_exhausted",
4
  "completed_steps": ["intent", "order_lookup"],
5
  "pending_steps": ["policy_check", "refund_decision"]
6
}

1
HTTP request timeout
2
  >= Agent Run sync budget
3
     >= node timeout
4
        >= model/tool attempt timeout

1
trace_id / run_id / tenant_id
2
operation_name / node_name / attempt
3
model_provider / model_name / model_config_version
4
prompt_version / tool_name / tool_version
5
input_size / output_size / token_usage
6
latency / error_type / retryable
7
state_before_hash / state_after_hash

1
agent_runs_total{status}
2
agent_run_duration_seconds
3
model_calls_total{provider,model,status}
4
model_tokens_total{direction}
5
tool_calls_total{tool,status}
6
tool_call_duration_seconds{tool}
7
agent_retries_total{reason}
8
agent_budget_exhausted_total{budget_type}
9
human_handoffs_total{reason}

1
End User Identity
2
  -> API Authentication
3
  -> Agent Principal
4
  -> Delegated Tool Scope
5
  -> Resource-level Authorization

1
query_order: order:read:self
2
search_policy: policy:read:published
3
create_refund: refund:create:eligible_order
4
create_compensation: compensation:create:approved_limit

1
def test_refund_cannot_exceed_paid_amount() -> None:
2
    decision = decide_refund(
3
        paid_amount_cents=10_000,
4
        requested_amount_cents=12_000,
5
        policy_limit_cents=10_000,
6
    )
7

8
    assert decision.allowed is False
9
    assert decision.reason == "amount_exceeded"

1
provider event
2
  -> internal ModelEvent
3
  -> tool arguments
4
  -> validated domain command

1
{
2
  "case_id": "foundation-refund-001",
3
  "input": {
4
    "message": "订单三天没发货，我要退款",
5
    "order_id": "ORD-A1B2C3D4"
6
  },
7
  "fixtures": {
8
    "order_status": "paid",
9
    "paid_amount_cents": 12900,
10
    "logistics_status": "not_shipped"
11
  },
12
  "expected": {
13
    "tool_sequence": [
14
      "query_order",
15
      "query_logistics",
16
      "search_policy"
17
    ],
18
    "forbidden_tools": ["create_compensation"],
19
    "terminal_status": "awaiting_confirmation"
20
  }
21
}

1
Schema Validity
2
Tool Selection Accuracy
3
Argument Accuracy
4
Task Success
5
Unsafe Action Rate
6
Latency p50/p95/p99
7
Tokens / Cost per Successful Run

1
POST /v1/agent-runs
2
  -> authenticate user
3
  -> create run
4
  -> classify intent
5
  -> query order
6
  -> verify ownership
7
  -> query logistics
8
  -> retrieve effective policy
9
  -> produce structured decision
10
  -> policy engine validates decision
11
  -> request confirmation when required
12
  -> execute refund with idempotency key
13
  -> query refund state
14
  -> persist evidence and audit
15
  -> emit final response

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

4

5
class AgentState(BaseModel):
6
    schema_version: Literal["1"] = "1"
7
    run_id: str
8
    user_id: str
9
    tenant_id: str
10
    status: Literal[
11
        "created",
12
        "running",
13
        "awaiting_input",
14
        "awaiting_confirmation",
15
        "succeeded",
16
        "failed",
17
        "needs_human",
18
    ]
19
    current_node: str
20
    completed_nodes: list[str] = Field(default_factory=list)
21
    order_id: OrderId | None = None
22
    evidence_refs: list[str] = Field(default_factory=list)
23
    pending_action: dict | None = None
24
    version: int = 0

1
succeeded
2
failed_retryable
3
failed_terminal
4
awaiting_input
5
awaiting_confirmation
6
needs_human
7
cancelled
8
budget_exhausted

1
工作目录与模块路径
2
依赖是否锁定
3
环境变量是否完整
4
容器内 DNS 是否使用 service name
5
数据库迁移是否执行
6
健康检查是否把启动慢误判为死亡

1
参数是否通过业务校验
2
用户是否拥有订单
3
订单状态是否在调用前已变化
4
幂等键是否冲突
5
远端是否成功但响应丢失
6
是否执行了 final-state verification

1
Redis 是否被误当真相源
2
TTL 是否短于 Run 生命周期
3
缓存键是否缺 tenant/user/version
4
写数据库与删缓存是否存在竞态
5
消费是否具备幂等与重放能力

1
Prompt / model / tool / policy 版本是否记录
2
随机性参数是否记录
3
外部工具结果是否有不可变引用或 hash
4
状态迁移前后是否可定位
5
敏感原文被脱敏后是否保留受控证据引用

1
Run 数量
2
每 Run 模型调用数
3
每次输入 token
4
重试率
5
检索 chunk 数
6
模型路由分布
7
未终止循环
8
缓存命中率
9
成功 Run 的单位成本

1
1. 建 API Schema
2
2. 建 AgentState
3
3. 用 FakeModel 返回固定结构
4
4. 用 FakeTool 模拟订单与物流
5
5. 完成状态迁移单元测试
6
6. 输出结构化事件

1
1. PostgreSQL 持久化 Run 与 Tool Call
2
2. Redis 做 TTL 缓存和限流
3
3. Docker Compose 启动依赖
4
4. 增加迁移、Readiness 和集成测试

1
1. 实现一个 Provider Adapter
2
2. 结构化输出解析与错误分类
3
3. 接入只读工具
4
4. 记录 token、延迟和成本
5
5. 跑固定 Scenario Dataset

1
1. 身份和资源授权
2
2. Policy Check
3
3. Human Confirmation
4
4. Idempotency Record
5
5. Final-state Verification
6
6. Audit Log 与故障演练