第 18 章 · Agent 架构设计：从 200 行循环到可维护的系统

def run_agent(client, max_steps=25):
    messages = [{"role": "system", "content": SYSTEM_PROMPT}]
    for i in range(max_steps):
        reply = client.chat(messages, TOOLS)      # ← 状态只活在这个函数里
        ...
    return report, messages

# clinic/agent_core.py（节选）
@dataclass
class AgentState:
    """Agent 的全部可变状态。刻意做成可 JSON 序列化的普通数据。"""
    goal: str
    step_index: int = 0
    status: str = "running"          # running / waiting_confirm / done / failed
    steps: list[Step] = field(default_factory=list)
    messages: list[Message] = field(default_factory=list)
    artifacts: dict[str, Any] = field(default_factory=dict)   # 中间产物
    budget: Budget = field(default_factory=Budget)

    def to_json(self) -> str:
        return json.dumps(asdict(self), ensure_ascii=False, default=str)

    @classmethod
    def from_json(cls, s: str) -> "AgentState":
        ...

# 1) 断点恢复
state = AgentState.from_json(open("checkpoint.json").read())
agent.run(goal, state=state)          # 从第 8 步继续

# 2) 事后复盘（监管最关心这个）
for s in state.steps:
    print(s.index, s.kind, s.tool, s.args, "→", s.result)

# 3) 回归测试：把线上真实跑过的状态当固定用例
def test_replay():
    state = AgentState.from_json(FIXTURE.read_text())
    assert replay(state)["结论"] == EXPECTED

@dataclass
class Step:
    index: int
    kind: str                        # think / tool / answer / error
    tool: str | None = None
    args: dict | None = None
    result: str | None = None
    error: str | None = None
    elapsed: float = 0.0
    tokens: int = 0

@dataclass
class ToolSpec:
    name: str
    description: str
    parameters: dict                    # JSON Schema
    func: Callable[[dict], ToolResult]
    side_effect: str = "read"           # read / write / irreversible
    timeout: float = 30.0
    tags: tuple[str, ...] = ()          # 便于做策略匹配，如 ("clinical", "qc")

def execute(self, name: str, args: dict, policy: Policy, state: AgentState) -> ToolResult:
    spec = self.registry.get(name)
    if spec is None:
        return ToolResult.fail(f"未知工具 {name}", error_type="validation")
    if spec.side_effect != "read" and policy.confirm_side_effects:
        state.status = "waiting_confirm"          # ← 交还控制权给人类
        return ToolResult.pending(f"{name} 需要人工确认：{args}")
    ...

@dataclass
class Policy:
    max_steps: int = 25
    max_tokens: int = 120_000
    deadline_sec: float = 300.0
    tool_timeout: float = 30.0
    confirm_side_effects: bool = True
    allowed_datasets: frozenset[str] = frozenset({"dm", "adsl", "adae", "ae", "vs", "adlbc"})

@dataclass
class Budget:
    max_steps: int = 25
    max_tokens: int = 120_000
    deadline_sec: float = 300.0
    steps_used: int = 0
    tokens_used: int = 0
    started_at: float = field(default_factory=time.time)

    def exceeded(self) -> str | None:
        """返回超限原因，None 表示还正常。"""
        if self.steps_used >= self.max_steps:
            return f"步数超限（{self.steps_used}/{self.max_steps}）"
        if self.tokens_used >= self.max_tokens:
            return f"token 超限（{self.tokens_used}/{self.max_tokens}）"
        if time.time() - self.started_at > self.deadline_sec:
            return f"时间超限（{time.time() - self.started_at:.0f}s）"
        return None

# clinic/agent_core.py 中 Agent.run() 的骨架
def run(self, goal: str, state: AgentState | None = None) -> AgentState:
    state = state or AgentState(goal=goal, budget=Budget(**asdict(self.policy.budget)))
    state.messages.insert(0, Message.system(self.system_prompt))

    while True:
        if (why := state.budget.exceeded()):
            return self._finish(state, f"因预算限制提前结束：{why}")

        reply = self.llm.chat(state.messages, self.registry.schemas())
        state.budget.tokens_used += reply.tokens
        state.messages.append(reply)

        if not reply.tool_calls:                       # 模型给出最终答案
            return self._finish(state, reply.content)

        for call in reply.tool_calls:
            result = self.registry.execute(call.name, call.arguments,
                                           self.policy, state)
            if state.status == "waiting_confirm":       # 交还控制权
                return state
            state.messages.append(Message.tool(call.id, result.to_llm_text()))
            state.steps.append(Step(...))
            state.budget.steps_used += 1

python cases/case08_分层架构QC_Agent.py                 # 离线 Mock，直接跑通
python cases/case08_分层架构QC_Agent.py --resume out.json   # 演示断点恢复