Agentic AI：从被动响应到自主目标达成的工程化路径

import os, json
from dataclasses import dataclass, field, asdict
from openai import OpenAI

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

@dataclass
class State:
    goal: str
    plan: list = field(default_factory=list)
    history: list = field(default_factory=list)
    step: int = 0

    def save(self, path="state.json"):
        with open(path, "w", encoding="utf-8") as f:
            json.dump(asdict(self), f, ensure_ascii=False, indent=2)

    @classmethod
    def load(cls, path="state.json"):
        with open(path, encoding="utf-8") as f:
            return cls(**json.load(f))

import subprocess, shlex

TOOLS = {}

def tool(name, desc):
    def deco(fn):
        TOOLS[name] = {"fn": fn, "desc": desc}
        return fn
    return deco

@tool("calc", "计算数学表达式，输入 expression")
def calc(expression: str) -> str:
    allowed = set("0123456789+-*/(). ")
    if not set(expression) <= allowed:
        return "错误：仅允许数字与四则运算"
    return str(eval(expression, {"__builtins__": {}}, {}))

@tool("run_test", "运行指定测试文件，输入 path")
def run_test(path: str) -> str:
    if not path.startswith("tests/"):
        return "拒绝：只允许运行 tests/ 下文件"
    p = subprocess.run(["pytest", "-q", path],
                       capture_output=True, text=True, timeout=60)
    return p.stdout[-1500:] or p.stderr[-1500:]

SYS = """你是任务型 Agent。每轮输出 JSON：
{"thought":"...","action":"工具名或final","input":"...","done":bool}
规则：优先复用历史结果；连续两次相同动作必须反思换策略；不确定就调用工具验证。"""

def agent(state: State, max_steps=8):
    while state.step < max_steps:
        ctx = {
            "goal": state.goal,
            "plan": state.plan,
            "history": state.history[-10:],
        }
        r = client.chat.completions.create(
            model="gpt-4o", temperature=0,
            response_format={"type": "json_object"},
            messages=[{"role": "system", "content": SYS},
                      {"role": "user", "content": json.dumps(ctx, ensure_ascii=False)}],
        )
        d = json.loads(r.choices[0].message.content)
        state.step += 1

        if d["action"] == "final" or d["done"]:
            state.save()
            return d["input"]

        fn = TOOLS.get(d["action"], {}).get("fn")
        try:
            obs = fn(d["input"]) if fn else f"未知工具 {d['action']}"
        except Exception as e:
            obs = f"工具异常：{e}"

        state.history.append({"thought": d["thought"],
                              "action": d["action"],
                              "input": d["input"],
                              "obs": obs})
        state.save()

    return "达到步数上限，未完成"

# 使用
s = State(goal="计算 (23*17+5)/4 并验证结果是否为整数")
print(agent(s))

class Bus:
    def __init__(self):
        self.queues = {}
    def register(self, name):
        self.queues[name] = []
    def send(self, to, msg):
        self.queues[to].append(msg)
    def recv(self, name):
        return self.queues[name].pop(0) if self.queues[name] else None

bus = Bus()
for role in ["planner", "coder", "reviewer"]:
    bus.register(role)

bus.send("coder", {"task": "实现 calculate_discount"})
bus.send("reviewer", {"task": "审查 coder 的产出"})