一个 agent 项目变乱的第一个征兆，是"改个重试逻辑要动三个文件"，第二个是"换供应商要改业务代码"。根因几乎总是同一件事：把协议适配、调用治理、任务调度塞进了同一个类。这篇按三层切开——LLMInterface 管协议，LLMClient 管治理，LLMEngine 管调度——再谈进程内与进程外两种运行时，以及能力（skills / MCP）从哪来。

1. 一次调用穿过哪几层

分层的判据很朴素：改某一层时，另一层的测试不该变。做不到就说明边界画错了。

2. LLMInterface：最小契约

from typing import Protocol, Iterator, Sequence

class Message(dict): ...          # {"role": ..., "content": ..., "tool_calls": [...]}

class LLMInterface(Protocol):
    def complete(self, msgs: Sequence[Message], **kw) -> Message: ...
    def stream(self, msgs: Sequence[Message], **kw) -> Iterator[str]: ...
    def count_tokens(self, msgs: Sequence[Message]) -> int: ...

count_tokens 被低估得最厉害。没有它，"超长输入"只能靠异常兜底，表现为偶发的 400；有了它，可以在拼上下文之前就做预算分配。

class OpenAIAdapter:
    def __init__(self, client): self._c = client
    def complete(self, msgs, **kw):
        r = self._c.chat.completions.create(model=kw["model"], messages=list(msgs))
        return Message(role="assistant", content=r.choices[0].message.content)
    def stream(self, msgs, **kw):
        for ch in self._c.chat.completions.create(..., stream=True):
            if ch.choices[0].delta.content:
                yield ch.choices[0].delta.content

3. LLMClient：治理逻辑只在这里长

import random, time

class LLMClient:
    def __init__(self, impl, max_retry=4, base=0.8, budget_tokens=None):
        self.impl, self.max_retry, self.base = impl, max_retry, base
        self.used = 0
        self.budget_tokens = budget_tokens

    def call(self, msgs, **kw):
        for i in range(self.max_retry):
            try:
                if self.budget_tokens and self.used >= self.budget_tokens:
                    raise BudgetExceeded(self.used)
                r = self.impl.complete(msgs, **kw)
                self.used += self.impl.count_tokens(msgs)
                return r
            except (RateLimit, Timeout) as e:
                wait = self.base * (2 ** i) + random.uniform(0, 0.3)   # 抖动
                time.sleep(min(wait, 20))
        raise RetryExhausted(msgs[-1].get("content", "")[:80])

退避必须带抖动。固定间隔重试会把所有失败请求对齐到同一时刻，形成周期性尖峰——这是"重试把故障放大"的标准剧本。

4. LLMEngine：并发与批处理的边界

import asyncio

class LLMEngine:
    def __init__(self, client, max_concurrency=8):
        self.client = client
        self.sem = asyncio.Semaphore(max_concurrency)

    async def submit(self, msgs, **kw):
        async with self.sem:                     # 并发闸门在这一层
            loop = asyncio.get_running_loop()
            return await loop.run_in_executor(None, lambda: self.client.call(msgs, **kw))

    async def batch(self, items, **kw):
        return await asyncio.gather(*[self.submit(m, **kw) for m in items],
                                    return_exceptions=True)

return_exceptions=True 是必须的：批处理里有 1 条失败就整体抛异常，等于让一条坏数据毁掉整批。

5. 进程内 vs 进程外：CLIAgent 与 Google-ADK 的分野

判据：需要与业务事务同生共死就进程内，需要隔离与独立伸缩就进程外。把 CLI 型代理嵌进 Web 请求线程里，是最常见的"看起来跑通、一上量就崩"的写法。

6. Agent-Skills 与 MCPClient：能力从哪来

class MCPClient:
    def __init__(self, transport): self.t = transport; self.tools = {}

    def discover(self):
        self.tools = {t["name"]: t for t in self.t.call("tools/list")["tools"]}
        return list(self.tools)

    def invoke(self, name, args):
        if name not in self.tools:
            raise UnknownTool(name)           # 名字漂移要显式失败
        return self.t.call("tools/call", {"name": name, "arguments": args})

按需加载 Skill 而不是全量注入，是控制上下文的关键：一个 20 个 skill 的项目，全量注入能吃满半个窗口。

7. UIToolCall：把"要人确认"的动作显式化

def dispatch(call):
    if call.name.startswith("ui."):
        return {"type": "ui_action", "action": call.name[3:],
                "args": call.args, "require_confirm": call.args.get("destructive", False)}
    return {"type": "tool", "result": tools[call.name](**call.args)}

关键点：UI 类动作的回执是不可信输入。前端说"用户已确认"，服务端仍需按自己的权限模型校验一次。

8. E2BBackend：把执行放到远端沙箱

class E2BBackend:
    def __init__(self, sandbox): self.sb = sandbox
    def run(self, code, timeout=30):
        return self.sb.run_code(code, timeout=timeout,
                                envs={"PYTHONUNBUFFERED": "1"})

选它的判据不是"更安全"，而是你的代码来源是否可信 + 是否需要多租户隔离。可信代码上远端沙箱只是把延迟从毫秒推到秒。

9. 排错：六种接口层现场

10. 边界：这些情况不需要三层

11. 自测

1. LLMInterface 里为什么必须有 count_tokens？缺了它会以什么形式暴露？
2. 指数退避为什么必须带抖动？不带会出现什么周期性现象？
3. 并发闸门为什么放在 LLMEngine 而不是 LLMClient？
4. 进程内 SDK 与 CLIAgent 式进程外代理，各自适合什么部署约束？
5. Agent-Skills 按需加载解决的是哪一类预算问题？全量注入会怎样？
6. UIToolCall 的回执为什么必须当不可信输入处理？如何在服务端兜住？