Збери свого першого AI агента (безпечно, з кодом)

class TapeTools(SafeTools):
  def __init__(self, allow: set[str], impl: dict, tape: list[dict] | None = None):
      super().__init__(allow=allow, impl=impl)
      self.tape = tape if tape is not None else []

  def call(self, name: str, *, args: dict[str, Any], request_id: str) -> Any:
      try:
          out = super().call(name, args=args, request_id=request_id)
          self.tape.append({"tool": name, "args": args, "ok": True, "out": out})
          return out
      except Exception as e:
          self.tape.append({"tool": name, "args": args, "ok": False, "err": str(e)})
          raise


def replay(tape: list[dict]):
  # Replace real tools with deterministic playback.
  i = 0
  def impl(name: str, **args):
      nonlocal i
      item = tape[i]
      i += 1
      assert item["tool"] == name
      if not item["ok"]:
          raise RuntimeError(item["err"])
      return item["out"]
  return impl

export class TapeTools extends SafeTools {
constructor({ allow, impl, tape }) {
  super({ allow, impl });
  this.tape = tape || [];
}

async call(name, { args, requestId }) {
  try {
    const out = await super.call(name, { args, requestId });
    this.tape.push({ tool: name, args, ok: true, out });
    return out;
  } catch (e) {
    this.tape.push({
      tool: name,
      args,
      ok: false,
      err: String(e && e.message ? e.message : e),
    });
    throw e;
  }
}
}

export function replay(tape) {
let i = 0;
return async function call(name, args) {
  const item = tape[i];
  i += 1;
  if (!item || item.tool !== name) throw new Error("tape mismatch");
  if (!item.ok) throw new Error(item.err);
  return item.out;
};
}

from dataclasses import dataclass
import time
from typing import Any


@dataclass(frozen=True)
class Budget:
  max_steps: int = 12
  max_seconds: int = 20


class Tools:
  def __init__(self):
      self.allow = {"web.search"}

  def call(self, name: str, *, args: dict[str, Any]) -> Any:
      if name not in self.allow:
          raise RuntimeError(f"tool not allowed: {name}")
      return tool_impl(name, args=args)  # (pseudo)


def first_agent(question: str, *, tools: Tools, budget: Budget) -> str:
  started = time.time()
  last = None

  for step in range(budget.max_steps):
      if time.time() - started > budget.max_seconds:
          return "stopped: time budget exceeded"

      action = llm_pick_action(question)  # (pseudo): either {"tool":..., "args":...} or {"finish":...}

      if action.get("finish"):
          return action["finish"]

      key = f"{action['tool']}:{action['args']}"
      if key == last:
          return "stopped: loop guard (same call twice)"
      last = key

      obs = tools.call(action["tool"], args=action["args"])
      question = update_state(question, action, obs)  # (pseudo)

  return "stopped: step budget exceeded"

export class Tools {
constructor() {
  this.allow = new Set(["web.search"]);
}

async call(name, { args }) {
  if (!this.allow.has(name)) throw new Error("tool not allowed: " + name);
  return toolImpl(name, { args }); // (pseudo)
}
}

export async function firstAgent(question, { tools, budget }) {
const started = Date.now();
let last = null;

for (let step = 0; step < budget.max_steps; step++) {
  if ((Date.now() - started) / 1000 > budget.max_seconds) {
    return "stopped: time budget exceeded";
  }

  const action = await llmPickAction(question); // (pseudo): { tool, args } or { finish }
  if (action.finish) return action.finish;

  const key = String(action.tool) + ":" + JSON.stringify(action.args);
  if (key === last) return "stopped: loop guard (same call twice)";
  last = key;

  const obs = await tools.call(action.tool, { args: action.args });
  question = updateState(question, action, obs); // (pseudo)
}

return "stopped: step budget exceeded";
}

from dataclasses import dataclass
import hashlib
import json
import time
import uuid
from typing import Any, Callable


@dataclass(frozen=True)
class Budget:
  max_steps: int = 12
  max_seconds: int = 20


def args_hash(args: dict[str, Any]) -> str:
  raw = json.dumps(args, sort_keys=True, default=str).encode("utf-8")
  return hashlib.sha256(raw).hexdigest()[:12]


class LoopGuard(RuntimeError):
  pass


class Monitor:
  def __init__(self, *, max_repeat: int = 2):
      self.max_repeat = max_repeat
      self.counts: dict[str, int] = {}

  def mark(self, tool: str, args: dict[str, Any]) -> None:
      key = f"{tool}:{args_hash(args)}"
      self.counts[key] = self.counts.get(key, 0) + 1
      if self.counts[key] >= self.max_repeat:
          raise LoopGuard(f"loop guard: {key} repeated {self.counts[key]}x")


class SafeTools:
  def __init__(self, allow: set[str], impl: dict[str, Callable[..., Any]]):
      self.allow = allow
      self.impl = impl

  def call(self, name: str, *, args: dict[str, Any], request_id: str) -> Any:
      if name not in self.allow:
          raise RuntimeError(f"[{request_id}] tool not allowed: {name}")
      started = time.time()
      try:
          return self.impl[name](**args)
      finally:
          ms = int((time.time() - started) * 1000)
          print(f"[{request_id}] tool={name} ms={ms} args_hash={args_hash(args)}")


def run_first_agent(question: str, *, tools: SafeTools, budget: Budget) -> str:
  request_id = uuid.uuid4().hex
  started = time.time()
  monitor = Monitor(max_repeat=2)
  state = {"question": question, "notes": []}

  for step in range(budget.max_steps):
      if time.time() - started > budget.max_seconds:
          return f"[{request_id}] stopped: time budget"

      action = llm_pick_action(state)  # (pseudo)
      if action.get("finish"):
          return action["finish"]

      tool = action["tool"]
      args = action["args"]
      monitor.mark(tool, args)

      obs = tools.call(tool, args=args, request_id=request_id)
      state = update_state(state, action, obs)  # (pseudo)

  return f"[{request_id}] stopped: step budget"

import crypto from "node:crypto";

function stableStringify(obj) {
if (obj === null || typeof obj !== "object") return JSON.stringify(obj);
if (Array.isArray(obj)) return "[" + obj.map(stableStringify).join(",") + "]";
const keys = Object.keys(obj).sort();
return "{" + keys.map((k) => JSON.stringify(k) + ":" + stableStringify(obj[k])).join(",") + "}";
}

export function argsHash(args) {
return crypto.createHash("sha256").update(stableStringify(args)).digest("hex").slice(0, 12);
}

export class LoopGuard extends Error {}

export class Monitor {
constructor({ maxRepeat = 2 } = {}) {
  this.maxRepeat = maxRepeat;
  this.counts = new Map();
}

mark(tool, args) {
  const key = tool + ":" + argsHash(args);
  const next = (this.counts.get(key) || 0) + 1;
  this.counts.set(key, next);
  if (next >= this.maxRepeat) throw new LoopGuard("loop guard: " + key + " repeated " + next + "x");
}
}

export class SafeTools {
constructor({ allow, impl }) {
  this.allow = allow;
  this.impl = impl;
}

async call(name, { args, requestId }) {
  if (!this.allow.has(name)) throw new Error("[" + requestId + "] tool not allowed: " + name);
  const started = Date.now();
  try {
    return await this.impl[name](args);
  } finally {
    const ms = Date.now() - started;
    console.log("[" + requestId + "] tool=" + name + " ms=" + ms + " args_hash=" + argsHash(args));
  }
}
}

export async function runFirstAgent(question, { tools, budget }) {
const requestId = crypto.randomUUID().replace(/-/g, "");
const started = Date.now();
const monitor = new Monitor({ maxRepeat: 2 });
let state = { question, notes: [] };

for (let step = 0; step < budget.max_steps; step++) {
  if ((Date.now() - started) / 1000 > budget.max_seconds) return "[" + requestId + "] stopped: time budget";

  const action = await llmPickAction(state); // (pseudo)
  if (action.finish) return action.finish;

  const tool = action.tool;
  const args = action.args;
  monitor.mark(tool, args);

  const obs = await tools.call(tool, { args, requestId });
  state = updateState(state, action, obs); // (pseudo)
}

return "[" + requestId + "] stopped: step budget";
}

def validate_web_search(args: dict[str, Any]) -> dict[str, Any]:
  q = str(args.get("q", "")).strip()
  if not q or len(q) > 200:
      raise ValueError("invalid q")

  k = int(args.get("k", 5))
  if k < 1 or k > 8:
      raise ValueError("invalid k")

  return {"q": q, "k": k}

export function validateWebSearch(args) {
const q = String((args && args.q) || "").trim();
if (!q || q.length > 200) throw new Error("invalid q");

const k = Number((args && args.k) ?? 5);
if (!Number.isFinite(k) || k < 1 || k > 8) throw new Error("invalid k");

return { q, k };
}