从产品级落地到架构内核：Agent 系统的分层解耦与最小内核实现

from dataclasses import dataclass, field
from enum import Enum

class Kind(Enum):
    COLLECT, REASON, ACT, VERIFY = range(4)

@dataclass
class Node:
    id: str
    kind: Kind
    prompt: str
    tools: list[str] = field(default_factory=list)
    edges: dict[str, str | None] = field(default_factory=dict)
    max_iter: int = 5

@dataclass
class State:
    task: str
    data: dict = field(default_factory=dict)
    trace: list = field(default_factory=list)
    error: str | None = None

class Permission:
    def __init__(self, mode="workspace"):
        self.mode = mode
    def allow(self, tool: str, args: dict) -> bool:
        if self.mode == "readonly":
            return tool in ("read", "search", "glob")
        if self.mode == "workspace":
            path = args.get("path", "")
            return not path.startswith(("/etc", "C:\\Windows"))
        return True

class Kernel:
    def __init__(self, nodes, entry, llm, tools, perm, max_steps=20):
        self.nodes, self.entry = nodes, entry
        self.llm, self.tools, self.perm = llm, tools, perm
        self.max_steps = max_steps

    def run(self, task: str) -> State:
        st = State(task=task)
        cur = self.entry
        for step in range(self.max_steps):
            node = self.nodes[cur]
            out = self._step(node, st)
            st.trace.append({"step": step, "node": node.id, "out": out})
            cond = self._route(node, out)
            cur = node.edges.get(cond)
            if cur is None:
                break
        return st

    def _step(self, node, st):
        for _ in range(node.max_iter):
            ctx = {k: st.data.get(k) for k in self._reads(node)}
            act = self.llm.decide(node.prompt, ctx, node.tools)
            if act["kind"] == "done":
                st.data[self._writes(node)] = act["payload"]
                return act
            if not self.perm.allow(act["tool"], act.get("args", {})):
                st.error = "permission denied"
                return {"error": st.error}
            res = self.tools[act["tool"]](**act.get("args", {}))
            st.data.setdefault("_scratch", []).append(res)
        return {"error": "node_not_converged"}

    def _reads(self, node):
        return {
            Kind.COLLECT: [],
            Kind.REASON: ["collected"],
            Kind.ACT: ["hypotheses"],
            Kind.VERIFY: ["action_result"],
        }[node.kind]

    def _writes(self, node):
        return {
            Kind.COLLECT: "collected",
            Kind.REASON: "hypotheses",
            Kind.ACT: "action_result",
            Kind.VERIFY: "verified",
        }[node.kind]

    def _route(self, node, out):
        if out.get("error"):
            return "error"
        if node.kind == Kind.VERIFY:
            return "done" if out.get("ok") else "retry"
        return "ok"

nodes = {
    "collect": Node("collect", Kind.COLLECT, "读取数据源",
                    tools=["read", "query"],
                    edges={"ok": "reason", "error": None}),
    "reason": Node("reason", Kind.REASON, "列出 3 个假设",
                   edges={"ok": "act"}),
    "act": Node("act", Kind.ACT, "验证每个假设",
                tools=["query", "corr"],
                edges={"ok": "verify"}),
    "verify": Node("verify", Kind.VERIFY, "确认结论是否回答问题",
                   edges={"done": None, "retry": "reason"}),
}

kernel = Kernel(nodes, "collect", llm, tools, Permission("workspace"))
state = kernel.run("分析销售异常波动")
print(state.data.get("collected"))
for ev in state.trace:
    print(ev["node"], ev["out"])