AI Agent 全路径：提示词、工具调用、记忆与多 Agent 协作的四个工程断层

import tiktoken

ENC = tiktoken.get_encoding("cl100k_base")

def build_prompt(system: str, few_shot: str, dynamic: str,
                 model_limit: int = 8192, max_output: int = 1024) -> dict:
    static = system + "\n" + few_shot          # 可缓存前缀
    static_tokens = len(ENC.encode(static))
    budget = model_limit - max_output - static_tokens

    dyn_tokens = len(ENC.encode(dynamic))
    if dyn_tokens > budget:
        return {
            "ok": False,
            "error": "dynamic_exceeds_budget",
            "static_tokens": static_tokens,
            "dynamic_tokens": dyn_tokens,
            "budget": budget,
            "truncated_by": dyn_tokens - budget,
        }
    return {"ok": True, "prompt": static + "\n" + dynamic,
            "cacheable_prefix_tokens": static_tokens}

import hashlib, time

RETRYABLE = {"timeout", "rate_limit", "transient_io"}

def execute(name: str, args: dict, fn, max_retries: int = 2) -> dict:
    key = hashlib.sha256(f"{name}:{sorted(args.items())}".encode()).hexdigest()[:16]
    attempts = 0
    while attempts <= max_retries:
        try:
            result = fn(**args)
            return {"ok": True, "result": result, "attempts": attempts + 1,
                    "idem_key": key}
        except Exception as e:
            err = classify(e)          # 返回 timeout / rate_limit / ...
            attempts += 1
            if err not in RETRYABLE or attempts > max_retries:
                return {"ok": False, "error": err, "detail": str(e),
                        "attempts": attempts, "idem_key": key}
            time.sleep(min(2 ** attempts * 0.2, 2.0))
    return {"ok": False, "error": "exhausted", "idem_key": key}

import time

def retrieve_memory(query_vec, store: list, top_k: int = 5,
                    half_life_days: float = 30) -> list:
    scored = []
    now = time.time()
    for item in store:
        sim = cosine(query_vec, item["vec"])
        age_days = (now - item["ts"]) / 86400
        decay = 0.5 ** (age_days / half_life_days)
        scored.append({
            "content": item["content"],
            "score": sim * decay,
            "age_days": round(age_days, 1),
            "superseded": item.get("superseded_by"),
        })
    fresh = [s for s in scored if not s["superseded"]]
    return sorted(fresh, key=lambda x: x["score"], reverse=True)[:top_k]

from pydantic import BaseModel

class TaskHandoff(BaseModel):
    task_id: str
    objective: str                      # 要达成什么
    constraints: list[str]              # 不能违反的规则
    inputs: dict                        # 必需的输入数据
    acceptance: list[str]               # 验收标准，逐条可检查
    deadline_sec: float
    return_schema: dict                 # 下游必须按此结构返回

def delegate(handoff: TaskHandoff, executor) -> dict:
    missing = [k for k in handoff.inputs if handoff.inputs[k] is None]
    if missing:
        return {"ok": False, "error": "incomplete_handoff",
                "missing_inputs": missing}
    result = executor(handoff)
    failed = [a for a in handoff.acceptance
              if not check_acceptance(a, result)]
    return {"ok": not failed, "failed_criteria": failed,
            "result": result}