超越代码：AI-DLC的新构件
AI给软件工程带来的最有趣的变化可能不是代理编写更多的代码。可能是开发系统本身能够从它已经构建的软件中学习。
梯形图转SCL | AI模型价格对比 | AI工具导航 | ONNX模型库 | Vibe Coding教程 | PLC在线仿真器 | Tripo 3D | Meshy AI | ElevenLabs | KlingAI | ArtSpace | Phot.AI | InVideo
在软件工程历史的大部分时间里，我们的开发生命周期一直以一组熟悉的构件为中心。需求描述我们想要什么。设计解释我们打算如何构建它。代码实现行为。测试告诉我们它是否有效。配置和基础设施使其可运行。我们通过回顾、标准、指导和经验来改进围绕这些构件的流程，但学习本身主要存在于人、文档和组织习惯中。
AI原生开发创造了一种添加新东西的可能性：从工程经验中生成的持久过程知识，像软件一样被评估、版本化，然后反馈到下一项工作中。与其依靠强化学习来更新模型权重——大多数企业不太可能自己做这件事——这种学习可以存在于模型之外，保持可审计性，并可能跨不同模型转移。
过去几天我进行的一个小实验让这种可能性变得更加具体。
1、从追踪到技能
实验始于WikiSkill（https://arxiv.org/html/2608.27454v1），这是一篇描述代理学习架构的最新论文，包含三个层次：执行追踪、从这些追踪中提炼知识的持久wiki，以及编码代理应遵循程序的可重用技能。
循环是直接的。推理代理执行任务。Wiki维护者研究成功和失败的轨迹并积累经验。技能提议者使用这些经验来提议对SKILL.md的更改。然后验证集决定候选技能是否真正提高了性能。如果没有，技能将被回滚，而积累的知识将保留。
WikiSkill论文报告称，进化的技能在多个基准测试和模型家族中提高了性能。移除持久wiki层将平均性能从63.7%降低到48.7%。有趣的是，允许推理代理在执行期间直接读取wiki会使情况略微恶化，从63.7%到60.9%。价值似乎不是来自将更多历史塞入上下文窗口，而是来自将经验提炼成简洁的过程。
我想理解机制而不仅仅是阅读基准测试，所以我使用免费API层在Google Colab中构建了一个小版本。
自然，我选择了咖啡店餐饮账单。
代理收到涉及kopi-o、teh tarik、鸡肉饭、咖喱角和其他熟悉项目的订单。GST、配送和舍入规则被明确说明。两个内部规则被隐藏：十件或更多物品的任何行享受5%折扣，以及3%的忠诚度折扣。Python生成了精确的基本事实，这给了我一个确定性的评分器。
代理尝试了账单，测试工具捕获了其追踪和结果，维护者将模式提炼成持久的wiki，提议者生成了候选技能以及解释为何建议每个更改的出处。然后未触及的验证集决定每个提案是否值得保留。
有趣的部分不是算术。而是循环的形状。
将一个WikiSkill迭代转换为普通工程语言，它看起来惊人地熟悉。代理推出是正在执行的工作。Wiki维护是回顾。提出的技能更改是拉取请求。验证集是CI。接受是合并；拒绝是回滚；技能影响日志是PR历史。
熟悉的工程周期 工作 → 反思 → 改进 → 测试 已经成为系统本身的一部分。
不需要更改任何模型权重。该架构允许模型保持冻结，而其周围的行为构件进化。
这就是我认为AI原生开发的影响变得更加有趣的地方。
2、实验也表明了为什么门控很重要
我的干净运行没有发现隐藏的规则。基线验证准确率为12.5%，在保留测试集上为20%。三次迭代后，WikiSkill分支和无wiki分支都保持在20%。没有候选技能通过验证门控。
优化器模型完全有能力产生完善的程序。它们无法可靠地做的是从证据中推断实际的隐藏规则。一位分析师甚至发明了"通常10%"的忠诚度折扣，几乎完全重复了另一个模型几天前产生的错误假设。
八个已知答案的验证账单拒绝了它。
那个结果给了我一种思考机器生成技能的有用方式。SKILL.md不会仅仅因为LLM在其中写了一些合理的内容就成为知识。提出的技能是关于更好行为的假设。独立证据决定该假设是否值得成为持久的。
这种区别很重要，因为一旦不正确的指令被提升为可重用技能，失败的性质就会改变。聊天中的幻觉答案会随着对话消失。提升为行为构件的幻觉可以被每个使用它的未来代理继承。
这就是为什么验证层最终可能比技能作者更重要。
3、功能可能留下的不仅仅是代码
考虑一个更现实的例子：为应用程序添加登录功能。AI编码代理可能使用头脑风暴技能来探索需求，规划技能来分解工作，调试技能来调查问题，以及验证技能来检查完成的实现。这些技能目前大多是静态的。有人已经知道一种有用的工作方式，所以有人为代理写下来。现在想象一下，在实现过程中，代理引入了一个受保护的路由，但忘记在应用程序的route-manifest.ts中注册它。安全测试发现了这个遗漏。代理进行调查，更新清单，重新生成授权覆盖，最终通过验证。通常功能就到此结束。代码修复保留下来，但工程经验可能只存在于碰巧看到它的人的头脑中。
学习型开发生命周期可以保留其他东西：
每当添加或修改受保护路由时，更新route-manifest.ts，重新生成授权覆盖，运行路由覆盖测试，并验证路由出现在授权矩阵中。这不是登录功能本身的一部分。它是关于如何在此特定系统中可靠地构建此类功能的知识。下一个认证功能可以继承该经验。因此，该功能产生的不仅仅是代码和测试。它可能产生了开发下一个功能的更好方法。软件开发开始不仅产生软件，还产生关于下一个软件应如何开发的知识。
4、这就是为什么技能不仅仅是另一种形式的RAG
当今许多企业代理都附带了大量知识。例如，财务代理可能可以访问包含财务政策、标准操作程序、批准矩阵、用户指南和常见问题的SharePoint。当有人要求它创建采购订单时，我们希望模型检索正确的文档，理解相关策略，重建过程，然后正确执行任务。这给执行时的检索和推理带来了很大责任。
我现在认为将几个不同的构件类别分开是有用的。
知识描述环境：我们使用哪个系统，当前财务指南在哪里，哪个文档是权威的。
策略描述必须和必须不发生的事情。
技能描述如何在该环境和这些约束中可靠地执行一类工作。
因此，采购订单技能不应复制可能下个月更改的财务批准阈值。相反，它应该描述过程：检索当前批准矩阵，确定适用哪个批准，验证供应商和预算，收集任何缺失的信息，通过授权工具创建采购订单，然后在返回结果之前验证交易。
知识保持权威性和可更改性。技能解释如何使用它。当技能可以被机器管理时，这种区别变得更加重要。代理可能合理地学习到尽早检查供应商状态可以减少被拒绝的采购订单提交，并提议将其作为程序的改进。它不应从历史行为中学习到某个特定批准"通常被跳过"并悄悄地将其编码为新的工作方式。技能可以改进策略的执行方式。不应允许它们重新定义策略。
5、更难的问题是将经验与结果联系起来
当我们想要改进的技能不是特定任务的技能（如认证或采购订单），而是开发技能（如头脑风暴、规划和验证）时，这变得特别有趣。调试技能有相对快速的反馈。代理提出假设，应用修复，运行测试，并快速了解假设是否有帮助。头脑风暴则不同。
假设头脑风暴会议提议创建一个新的认证服务。在会议结束时，这个想法可能看起来完全合理。两天后，在实现过程中，团队发现现有服务已经提供了大部分所需功能，提议设计的很大一部分被丢弃。只有这样我们才知道早期头脑风暴的质量。规划也有同样的延迟反馈。计划可能看起来很出色，直到实现反复发现缺失的迁移、生成的客户端、安全工作或下游依赖。如果我们希望AI原生开发改进这些技能，单个代理会话是不够的。为同一工作产品做出贡献的会话必须连接起来。
我会为每个有意义的功能提供一个持久的相关性ID，并在整个开发生命周期中携带它：
FEATURE-142
需求
↓
头脑风暴
↓
规划
↓
实现
↓
调试/修正
↓
PR + CI
↓
安全验证
↓
UAT
↓
生产结果
一旦存在该链，计划学习过程就可以回顾整个事件。初始架构决策是否导致了下游返工？哪些假设后来被纠正了？原始计划未能预见什么？哪些调试路径反复浪费时间？什么逃过了验证，被UAT或生产捕获？这与普通的代理跟踪不同。
代理可观察性告诉我们代理做了什么。生命周期可观察性告诉我们它所做的是否有用。
因此，AI原生开发中的学习单元可能是工作产品的生命周期，而不是单个代理会话。
# WikiSkill — minimal Colab replication. FIRST LINE of output must say v49. Same as v48 (startup health check, fallbacks, limiter, report) with WORKERS=4 — no throughput loss, fewer in-flight connections.
# Secrets: NVIDIA_API_KEY + GEMINI_API_KEY. Restart session once before running.
# ======================================================================
# WikiSkill — minimal Colab replication
# ======================================================================
%pip -q install openai
import os, re, io, json, time, random, shutil, difflib, contextlib, statistics
from decimal import Decimal, ROUND_HALF_UP
from concurrent.futures import ThreadPoolExecutor
from openai import OpenAI
from google.colab import userdata
import ast as _ast
# ---- pick a provider (one flag). Keys go in Colab Secrets (key icon, left sidebar).
PROVIDERS = {
"anthropic": dict(base_url="https://api.anthropic.com/v1/", secret="ANTHROPIC_API_KEY"),
"deepseek": dict(base_url="https://api.deepseek.com", secret="DEEPSEEK_API_KEY"),
"gemini": dict(base_url="https://generativelanguage.googleapis.com/v1beta/openai/", secret="GEMINI_API_KEY"),
"openai": dict(base_url=None, secret="OPENAI_API_KEY"),
"nvidia": dict(base_url="https://integrate.api.nvidia.com/v1", secret="NVIDIA_API_KEY"),
"cerebras": dict(base_url="https://api.cerebras.ai/v1", secret="CEREBRAS_API_KEY"),
"groq": dict(base_url="https://api.groq.com/openai/v1", secret="GROQ_API_KEY"),
"mistral": dict(base_url="https://api.mistral.ai/v1", secret="MISTRAL_API_KEY"),
"sambanova": dict(base_url="https://api.sambanova.ai/v1", secret="SAMBANOVA_API_KEY"),
}
# roles can live on different providers so free-tier rate caps add up
# NVIDIA free tier caps REQUESTS (~40/min) not tokens — the right shape for token-heavy agent loops.
# Groq free tier caps tokens (8k TPM on gpt-oss-20b): too small for this workload, kept here as reference.
HARNESS_VERSION = "v49"
print(f"=== WikiSkill harness {HARNESS_VERSION} ===")
import threading as _th
_zombies = [t for t in _th.enumerate() if t.name.startswith("ThreadPoolExecutor")]
if _zombies:
raise SystemExit(f"REFUSING TO START: {len(_zombies)} worker threads from an interrupted run are still alive and calling the API. "
"Runtime -> Restart session (or Disconnect and delete runtime), then run this cell again.")
INFER_PROVIDER = "nvidia"
INFER_MODEL = "google/gemma-4-31b-it" # rollouts: plain non-reasoning instruction-follower — short outputs, no thinking channel
INFER_FALLBACKS = ["nvidia:deepseek-ai/deepseek-v4-flash-0731", "nvidia:nvidia/nemotron-3.5-lightning-30b-a3b",
"nvidia:google/gemma-3-12b-it", "nvidia:mistralai/mistral-nemotron", "nvidia:microsoft/phi-3.5-moe-instruct",
"gemini:gemini-flash-lite-latest"] # probed at startup (first live one wins) and used per call if the chosen one dies
# reasoning alternative: "nvidia/nemotron-3.5-lightning-30b-a3b" (thinking-on; slower, answers can land in reasoning_content)
OPT_PROVIDER = "gemini" # analyst on separate infrastructure: ~20 calls/run sits well inside Gemini's free tier
OPT_MODEL = "gemini-3.5-flash" # an older Flash generation: separate free-tier quota bucket, far less demand than the newest
OPT_FALLBACKS = ["gemini-flash-lite-latest", "gemini-3.1-flash-lite", "gemini-flash-latest",
"nvidia:google/gemma-4-31b-it"] # "provider:model" entries switch provider — last resort is off-Google entirely
# all-NVIDIA fallback: OPT_PROVIDER="nvidia", OPT_MODEL="deepseek-ai/deepseek-v4-flash-0731" or "google/gemma-4-31b-it"
# NOT "moonshotai/kimi-k2.6": listed in the catalog but its NIM backend 404s ("Function ... Not Found")
# fallbacks: "deepseek-ai/deepseek-v4-pro-0813", "nvidia/nemotron-3-super-120b-a12b" (thinking off)
# avoid "mistralai/mistral-large-2-instruct" — its NIM endpoint returned empty completions
# gemini alternative: OPT_PROVIDER="gemini", OPT_MODEL="gemini-2.5-flash" (GEMINI_API_KEY from aistudio.google.com)
# thinking flags are Nemotron-specific — gate on the MODEL, never the provider
INFER_EXTRA = {"chat_template_kwargs": {"enable_thinking": True}} if "nemotron" in INFER_MODEL else None # native mode; False half-breaks the 3.5 template
OPT_EXTRA = {"chat_template_kwargs": {"enable_thinking": False}} if "nemotron" in OPT_MODEL else None
OPT_MAX_TOKENS = 3000 # brevity is enforced in the prompts; long generations were the slow part
# deepseek → "deepseek-chat" both roles; anthropic → "claude-haiku-4-5" / "claude-sonnet-4-5"
# nvidia catalog is large — list live ids with the optional cell below; some model families need one-time
# registration on their build.nvidia.com page ("Try API") or you get a 403.
# gemini → "gemini-2.5-flash" / "gemini-2.5-pro"
# cross-provider transfer test: evaluate the evolved skills with a different provider+model, e.g. ("deepseek", "deepseek-chat")
TRANSFER = None
N_TRAIN, N_VAL, N_TEST = 16, 8, 10 # more training evidence for rule decomposition; rollouts are cheap now
ITERS = 3
WORKERS = 4 # per arm. 4 x 2 arms = 8 in flight already saturates RPM_CAP; fewer open connections is kinder to a struggling backend
RPM_CAP = 36 # just under NVIDIA's ~40 req/min free-tier cap
SEED = 0
_clients = {}
def get_client(provider):
if provider not in _clients:
p = PROVIDERS[provider]
_clients[provider] = OpenAI(api_key=userdata.get(p["secret"]), base_url=p["base_url"],
timeout=120, max_retries=0) # fail fast; our own loop retries. Default was a silent 10-minute hang.
return _clients[provider]
client = get_client(INFER_PROVIDER) # default client: rollouts
opt_client = get_client(OPT_PROVIDER)
def probe(provider, model, timeout=25):
"""Is this endpoint answering right now? Tiny request, short timeout."""
try:
r = get_client(provider).with_options(timeout=timeout).chat.completions.create(
model=model, messages=[{"role": "user", "content": "Reply with OK."}], max_tokens=5)
return bool((r.choices[0].message.content or getattr(r.choices[0].message, "reasoning_content", "") or "").strip())
except Exception as e:
print(f" probe {provider}/{model}: {type(e).__name__}"); return False
def pick_live(primary, fallbacks, default_provider, role):
for entry in [primary] + fallbacks:
prov, _, m = entry.partition(":") if ":" in entry else (default_provider, None, entry)
if probe(prov, m):
print(f" {role}: {prov}/{m} is live"); return prov, m
print(f" {role}: nothing answered the probe — keeping {primary}"); return default_provider, primary
INFER_PROVIDER, INFER_MODEL = pick_live(INFER_MODEL, INFER_FALLBACKS, INFER_PROVIDER, "rollouts")
OPT_PROVIDER, OPT_MODEL = pick_live(OPT_MODEL, OPT_FALLBACKS, OPT_PROVIDER, "optimiser")
client, opt_client = get_client(INFER_PROVIDER), get_client(OPT_PROVIDER)
INFER_EXTRA = {"chat_template_kwargs": {"enable_thinking": True}} if "nemotron" in INFER_MODEL else None
OPT_EXTRA = {"chat_template_kwargs": {"enable_thinking": False}} if "nemotron" in OPT_MODEL else None
print(f"rollouts: {INFER_PROVIDER}/{INFER_MODEL} | optimiser: {OPT_PROVIDER}/{OPT_MODEL} | "
f"splits {N_TRAIN}/{N_VAL}/{N_TEST} x {ITERS} iters | workers {WORKERS}")
import threading, collections as _collections
STOP = threading.Event() # set on interrupt; workers check it and bail out instead of draining their queue
_rl_lock = threading.Lock(); _rl_times = _collections.deque()
def rate_limit():
"""Token bucket shared by every thread: at most RPM_CAP requests start per rolling 60s."""
while True:
with _rl_lock:
now = time.time()
while _rl_times and now - _rl_times[0] > 60: _rl_times.popleft()
if len(_rl_times) < RPM_CAP:
_rl_times.append(now); return
wait = 60 - (now - _rl_times[0]) + 0.05
time.sleep(max(wait, 0.05))
PY_TOOL = [{"type": "function", "function": {
"name": "python",
"description": "Execute Python code and return its stdout. print() anything you need to see.",
"parameters": {"type": "object", "properties": {"code": {"type": "string", "description": "Python source to execute"}}, "required": ["code"]}}}]
def _salvage_tool_call(e):
"""Groq rejects GPT-OSS tool calls whose arguments aren't valid JSON, but includes the raw
generation in the error. The code the model wanted to run is in there — recover it."""
body = getattr(e, "body", None)
fg = (body.get("error") or {}).get("failed_generation") if isinstance(body, dict) else None
if not fg:
m = re.search(r"failed_generation.: .(.*)", str(e), re.S)
fg = m.group(1) if m else None
if not fg: return None
m = re.search(r'"arguments"\s*:\s*(.*)', fg, re.S)
if not m: return None
code = m.group(1).strip()
for tail in ("'}", "}"):
if code.endswith(tail): code = code[: -len(tail)].rstrip()
code = code.strip('"').strip()
if code.startswith("{"):
try: code = json.loads(code).get("code", "")
except Exception: pass
if not code: return None
tc = type("TC", (), {})(); tc.id = "salvaged_0"
tc.function = type("F", (), {})(); tc.function.name = "python"; tc.function.arguments = json.dumps({"code": code})
msg = type("M", (), {})(); msg.content = ""; msg.tool_calls = [tc]
return msg
def chat_full(messages, model, temperature=0.0, max_tokens=2500, cl=None, extra_body=None, tools=None):
cl = cl or client; err = None
for attempt in range(8): # unattended-safe: more patience, longer backoffs
if STOP.is_set(): raise RuntimeError("stopped")
try:
kw = dict(model=model, messages=messages, temperature=temperature, max_tokens=max_tokens)
if extra_body: kw["extra_body"] = extra_body
if tools: kw["tools"] = tools
rate_limit()
r = cl.chat.completions.create(**kw)
m = r.choices[0].message
if not getattr(m, "tool_calls", None) and not (m.content or "").strip():
rc = (getattr(m, "reasoning_content", None) or "").strip()
if rc: # hybrid-reasoning models can put the whole answer in the reasoning channel
m.content = rc
return m
fr = getattr(r.choices[0], "finish_reason", "?")
print(f" [debug] empty completion from {model} (finish_reason={fr}, no reasoning_content)")
if len(messages) == 2 and messages[0]["role"] == "system": # some serving templates mishandle the system role
messages = [{"role": "user", "content": f"[Instructions]\n{messages[0]['content']}\n\n[Input]\n{messages[1]['content']}"}]
continue
temperature = 0.4; continue
return m
except Exception as e:
err = e
msg = str(e).lower()
print(f" [debug] {model} call failed (attempt {attempt+1}/8): {type(e).__name__}: {str(e)[:120]}")
if "timed out" in msg or "timeout" in type(e).__name__.lower():
timeouts = locals().get("timeouts", 0) + 1
if timeouts >= 2: raise RuntimeError(f"{model}: endpoint unresponsive (2 consecutive timeouts)") from e
if extra_body and ("400" in msg or "invalid" in msg or "unsupported" in msg or "unknown" in msg) and "tool" not in msg:
extra_body = None; continue # provider rejected an optional param — retry clean
if "tool_use_failed" in msg or ("tool" in msg and "400" in msg):
sal = _salvage_tool_call(e)
if sal: return sal # recover the code from the rejected generation
tools = None # endpoint may not support tools at all — degrade to the markdown protocol
temperature = 0.5; continue
if "402" in msg or "payment_required" in msg:
raise RuntimeError(f"{model}: not covered by this provider's free quota — pick a model from the account's usage/limits page") from e
if "model_not_found" in msg or "does not exist" in msg or "404" in msg or "not found" in msg:
raise RuntimeError(f"{model}: 404 from provider (not in catalog, or listed but not deployed) — pick another id") from e
time.sleep((10 if any(s in msg for s in ("429", "rate", "503", "high demand", "overloaded", "529")) else 2) * (attempt + 1))
raise err
def chat(messages, model, **kw):
return chat_full(messages, model, **kw).content or ""
def tc_code(tc):
a = tc.function.arguments or ""
try: return json.loads(a).get("code", "")
except Exception: return a # some models emit raw code as the arguments string
def serialize_assistant(m):
d = {"role": "assistant", "content": m.content or ""}
if getattr(m, "tool_calls", None):
d["tool_calls"] = [{"id": tc.id, "type": "function",
"function": {"name": tc.function.name, "arguments": tc.function.arguments}}
for tc in m.tool_calls]
return d
def llm(system, user, model, **kw): # kw may include cl= to route to a specific provider
return chat([{"role": "system", "content": system}, {"role": "user", "content": user}], model, **kw)
def parse_json(text):
text = text.strip()
fences = re.findall(r"```json\s*(.*?)```", text, re.S) # prefer explicit json fences, take the LAST
if not fences:
fences = [f for f in re.findall(r"```\s*(.*?)```", text, re.S) if f.lstrip().startswith("{")]
if fences: text = fences[-1]
start, end = text.find("{"), text.rfind("}")
blob = text[start:end + 1]
try:
return json.loads(blob)
except Exception:
return _ast.literal_eval(blob) # tolerate single-quoted / Python-literal dicts
# live model catalogs — set INFER_MODEL / OPT_MODEL from these lists (a listing is a claim, not a capability: probe before trusting)
for label, cl_, filt in (("rollout provider", client, None), ("optimiser provider", opt_client, "gemini")):
try:
ids = sorted({m.id for m in cl_.models.list().data})
if filt: ids = [i for i in ids if filt in i.lower()]
print(f"{label} — models available to this key:")
for i in ids[:40]: print(" ", i)
except Exception as e:
print(f"{label} listing failed:", e)
print("\nBEFORE A FULL RUN: Runtime -> Restart session (interrupted runs leave worker threads alive, still calling the API).")
# ======================================================================
# Tasks (synthetic, deterministic ground truth, hidden rules)
# ======================================================================
ITEMS = [("kopi-o", "1.40"), ("teh tarik", "1.70"), ("Milo dinosaur", "3.20"),
("kaya toast set", "3.30"), ("half-boiled eggs (pair)", "1.90"), ("nasi lemak", "4.00"),
("chicken rice", "4.50"), ("char kway teow", "5.00"), ("mee siam", "3.50"),
("curry puff", "1.80"), ("otah", "1.20"), ("bandung", "1.60")]
def make_task(i, rng):
n = rng.randint(2, 4)
lines = [(name, rng.choice([1, 2, 3, 4, 5, 6, 8, 10, 12, 15]), Decimal(p))
for name, p in rng.sample(ITEMS, n)]
member = rng.random() < 0.5
sub = Decimal("0")
for _, qty, price in lines:
line = price * qty
if qty >= 10: line *= Decimal("0.95") # HIDDEN rule 1: bulk 5% off lines with qty >= 10
sub += line
if member: sub *= Decimal("0.97") # HIDDEN rule 2: members 3% off subtotal
delivery = Decimal("8") if sub < 50 else Decimal("0") # stated
total = ((sub + delivery) * Decimal("1.09")).quantize(Decimal("0.01"), ROUND_HALF_UP) # stated
order = "; ".join(f"{qty} x {name} @ S${price} each" for name, qty, price in lines)
prompt = (f"Prepare the final bill for a kopitiam catering order in Singapore.\n"
f"Order: {order}.\n"
f"Customer is {'a loyalty member' if member else 'not a member'}.\n"
f"Store policy: delivery is S$8.00 if the discounted subtotal is below S$50.00, otherwise free. "
f"GST of 9% applies to subtotal plus delivery. Round the final amount half-up to the nearest cent.\n"
f"Give the amount the customer pays. Your last line must be exactly:\n"
f"FINAL: TOTAL: SGD <amount with 2 decimals>")
return {"id": f"t{i:03d}", "prompt": prompt, "answer": f"{total:.2f}"}
def grade(pred, answer):
m = re.search(r"SGD\s*\$?\s*([\d,]+\.\d{2})", pred or "")
return 1.0 if m and m.group(1).replace(",", "") == answer else 0.0
rng = random.Random(SEED)
ALL = [make_task(i, rng) for i in range(N_TRAIN + N_VAL + N_TEST)]
def _has_hidden(t): # any bulk-qty line, or a member
return bool(re.search(r"\b(10|12|15) x", t["prompt"])) or "not a member" not in t["prompt"]
easy = [t for t in ALL if not _has_hidden(t)]; hard = [t for t in ALL if _has_hidden(t)]
mixed = [] # interleave so each split inherits the same easy/hard ratio
while easy or hard:
for _ in range(max(1, round(len(hard) / max(1, len(easy))))):
if hard: mixed.append(hard.pop(0))
if easy: mixed.append(easy.pop(0))
TRAIN, VAL, TEST = mixed[:N_TRAIN], mixed[N_TRAIN:N_TRAIN + N_VAL], mixed[N_TRAIN + N_VAL:]
print(TRAIN[0]["prompt"], "\n-> answer:", TRAIN[0]["answer"])
# ======================================================================
# Inference Agent (ReAct with a Python tool)
# ======================================================================
AGENT_SYSTEM = """You are a careful kopitiam billing assistant.
Use the python tool to run code (or, if you cannot call tools, write exactly one ```python code block).
Do the arithmetic in Python, never in your head, and make your code PRINT the final line itself, exactly: TOTAL: SGD <amount with 2 decimals>. Be brief: no explanations, no restating the task. If you must answer without code, reply with one last line starting with 'FINAL: ' containing the actual amount. Never write a FINAL: line with placeholder text."""
def run_python(code):
buf = io.StringIO()
g = {"__name__": "__main__"}
try:
with contextlib.redirect_stdout(buf):
try:
tree = _ast.parse(code)
except SyntaxError as e:
print(f"[error] SyntaxError: {e}"); tree = None
if tree is not None:
if tree.body and isinstance(tree.body[-1], _ast.Expr): # Jupyter-style: echo a trailing bare expression
exec(compile(_ast.Module(body=tree.body[:-1], type_ignores=[]), "<cell>", "exec"), g)
val = eval(compile(_ast.Expression(tree.body[-1].value), "<cell>", "eval"), g)
if val is not None: print(repr(val))
else:
exec(compile(tree, "<cell>", "exec"), g)
except Exception as e:
buf.write(f"\n[error] {type(e).__name__}: {e}")
return buf.getvalue()[-3000:] or "(no output)"
def run_agent(task, skills_text, model, max_steps=3, cl=None):
if STOP.is_set():
return {"id": task["id"], "prompt": task["prompt"], "pred": "", "gt": task["answer"], "score": 0.0, "trace": []}
system = AGENT_SYSTEM + (f"\n\n# SKILLS — follow these procedures\n{skills_text}" if skills_text else "")
msgs = [{"role": "system", "content": system}, {"role": "user", "content": task["prompt"]}]
trace, pred = [], ""
for _ in range(max_steps):
m = None
for entry in [f"{INFER_PROVIDER}:{model}"] + (INFER_FALLBACKS if cl is None else []):
prov, _, mdl = entry.partition(":")
try:
m = chat_full(msgs, mdl, cl=(cl or get_client(prov)),
extra_body=({"chat_template_kwargs": {"enable_thinking": True}} if "nemotron" in mdl else None), tools=PY_TOOL)
if mdl != model: trace.append({"fallback_model": f"{prov}/{mdl}"})
break
except Exception as e:
print(f" rollout: {prov}/{mdl} unavailable ({type(e).__name__}); trying next", flush=True)
if m is None: break
if getattr(m, "tool_calls", None):
msgs.append(serialize_assistant(m))
done = None
for tc in m.tool_calls:
res = run_python(tc_code(tc))
msgs.append({"role": "tool", "tool_call_id": tc.id, "content": res})
trace.append({"tool_call": tc_code(tc), "tool_output": res})
hit = re.findall(r"TOTAL:\s*SGD\s*[\d,]+\.\d{2}", res)
if hit: done = hit[-1]
if done: pred = done; break # single-shot: the code printed the answer
continue
out = m.content or ""
msgs.append({"role": "assistant", "content": out}); trace.append({"assistant": out})
code = re.search(r"```python\n(.*?)```", out, re.S)
if code: # text-protocol fallback for models that don't call tools
res = run_python(code.group(1))
trace.append({"tool_output": res})
hit = re.findall(r"TOTAL:\s*SGD\s*[\d,]+\.\d{2}", res)
if hit: pred = hit[-1]; break # single-shot: the code printed the answer
msgs.append({"role": "user", "content": f"[python stdout]\n{res}\n\nContinue. Finish with a FINAL: line."})
continue
finals = [f.strip() for f in re.findall(r"FINAL:\s*(.*)", out) if "<" not in f]
if finals: pred = finals[-1]; break
msgs.append({"role": "user", "content": "No code and no usable FINAL line found. Run Python or give the FINAL: line with the actual amount."})
return {"id": task["id"], "prompt": task["prompt"], "pred": pred, "gt": task["answer"],
"score": grade(pred, task["answer"]), "trace": trace}
def evaluate(tasks, skills_text, model, cl=None):
print(f" rollouts: {len(tasks)} tasks, skills={'yes' if skills_text else 'none'}...", flush=True)
t0 = time.time()
with ThreadPoolExecutor(WORKERS) as ex:
res = list(ex.map(lambda t: run_agent(t, skills_text, model, cl=cl), tasks))
print(f" [timing] {len(tasks)} rollouts in {time.time()-t0:.0f}s")
return res
def mean_score(results):
return round(100 * statistics.mean(r["score"] for r in results), 1)
# ======================================================================
# Workspace: raw/ · wiki/ · skills/
# ======================================================================
WS = "/content/wikiskill"
def ws(arm, *p): return os.path.join(WS, arm, *p)
def init_ws(arm):
shutil.rmtree(ws(arm), ignore_errors=True)
for d in ["raw", "wiki/patterns", "skills"]: os.makedirs(ws(arm, d))
for f in ["wiki/index.md", "wiki/logs.md", "wiki/skill-impact.md"]:
open(ws(arm, f), "w").write("")
def save_traces(arm, k, split, results):
for r in results:
json.dump(r, open(ws(arm, "raw", f"iter{k}_{split}_{r['id']}.json"), "w"), indent=1)
def read_skills(arm):
out = []
for name in sorted(os.listdir(ws(arm, "skills"))):
p = ws(arm, "skills", name, "SKILL.md")
if os.path.exists(p): out.append(f"## skill: {name}\n" + open(p).read())
return "\n\n".join(out)
def read_wiki(arm):
parts = ["# wiki/index.md\n" + open(ws(arm, "wiki/index.md")).read()]
for f in sorted(os.listdir(ws(arm, "wiki/patterns"))):
parts.append(f"# wiki/patterns/{f}\n" + open(ws(arm, "wiki/patterns", f)).read())
parts.append("# wiki/logs.md\n" + open(ws(arm, "wiki/logs.md")).read())
parts.append("# wiki/skill-impact.md\n" + open(ws(arm, "wiki/skill-impact.md")).read())
return "\n\n".join(parts)
def fmt_trace(r, max_chars=1200):
body = "\n".join(f"[{k}] {v}" for step in r["trace"] for k, v in step.items())
return (f"### {r['id']} — {'PASS' if r['score'] else 'FAIL'} | pred={r['pred']!r} | gt={r['gt']}\n"
f"TASK: {r['prompt']}\n{body[:max_chars]}")
def sample_traces(results, n_fail=3, n_pass=1):
fails = [r for r in results if not r["score"]][:n_fail]
passes = [r for r in results if r["score"]][:n_pass]
return "\n\n".join(fmt_trace(r) for r in fails + passes)
def outcome_summary(results):
return "\n".join(f"{r['id']}: {'PASS' if r['score'] else 'FAIL'} pred={r['pred']!r} gt={r['gt']}" for r in results)
# ======================================================================
# Wiki Maintainer + Skill Proposer + Gate
# ======================================================================
MAINTAINER_SYSTEM = """You are the Wiki Maintainer for an agent-skill evolution system.
From the evidence table and traces, do root-cause analysis of FAILED tasks and extract what worked from PASSED ones.
Each pattern page documents ONE failure mode or strategy: evidence (task ids), root cause, actionable workaround.
Update existing pages (rewrite in full) or add new ones. Do not delete knowledge. Ground every claim in the table; never invent numbers. Be concise: each page under 120 words, at most 3 pages per iteration.
Output format — plain text, EXACTLY these sentinels, no JSON:
=== FILE: kebab-slug.md
<full page content>
(repeat === FILE: blocks as needed)
=== LOG
<one-paragraph summary of this iteration's findings>"""
PROPOSER_SYSTEM = """You are the Skill Proposer for an agent-skill evolution system.
Propose exactly ONE atomic change: create a new skill, or edit one existing skill (give its full new content).
Skills are concise procedural instructions the billing agent follows verbatim from its system prompt. Keep the skill under 200 words: numbered steps, no commentary.
Ground every number in the evidence table (the ratio column shows how ground truth deviates from the naive stated-rules total). Never invent a number. If ratios vary across tasks, do not average them into an approximate rate: look for an exact rule that reproduces EVERY ground truth — it may depend on individual line quantities, not only on membership — and reject any hypothesis that fails even one task. If a wiki and impact history are given, build on them and NEVER re-propose a rejected change.
Output format — plain text, EXACTLY these sentinels, no JSON:
=== SKILL: kebab-slug
=== ACTION: create or edit
=== PURPOSE: <which evidence motivated this>
=== CONTENT
<full SKILL.md markdown>"""
def opt_llm(system, user, arm, role):
"""Optimiser call with a fallback chain; entries may be 'model' (optimiser provider) or 'provider:model'. Returns (text, model_used)."""
for entry in [OPT_MODEL] + OPT_FALLBACKS:
prov, _, m = entry.partition(":") if ":" in entry else (OPT_PROVIDER, None, entry)
try:
return llm(system, user, m, max_tokens=OPT_MAX_TOKENS, cl=get_client(prov),
extra_body=({"chat_template_kwargs": {"enable_thinking": False}} if "nemotron" in m else None)), f"{prov}/{m}"
except Exception as e:
print(f"[{arm}] {role}: {prov}/{m} gave up ({type(e).__name__}); trying next fallback", flush=True)
raise RuntimeError(f"{role}: all optimiser models failed")
def evidence_table(results):
"""Pre-computed analysis substrate: per task, parsed lines, member flag, naive stated-rules total,
ground truth, prediction, and gt/naive ratio. Harness-side feature engineering so the optimiser
can infer hidden rules by inspection instead of needing a Python tool."""
rows = ["id | lines (qty x price) | member | naive_total | ground_truth | predicted | pass | gt/naive"]
for r in results:
lines = re.findall(r"(\d+) x [^@]+ @ S\$(\d+\.\d+)", r["prompt"])
member = "not a member" not in r["prompt"]
sub = sum(Decimal(p) * int(q) for q, p in lines)
delivery = Decimal("8") if sub < 50 else Decimal("0")
naive = ((sub + delivery) * Decimal("1.09")).quantize(Decimal("0.01"), ROUND_HALF_UP)
ratio = (Decimal(r["gt"]) / naive).quantize(Decimal("0.0001"))
rows.append(f'{r["id"]} | {", ".join(f"{q}x{p}" for q, p in lines)} | {"Y" if member else "N"} | '
f'{naive} | {r["gt"]} | {r["pred"] or "(none)"} | {"P" if r["score"] else "F"} | {ratio}')
return "\n".join(rows)
def sentinel_blocks(text, key):
return re.findall(rf"^=== {key}:?\s*(.*?)\n(.*?)(?=^=== |\Z)", text or "", re.S | re.M)
def maintain_wiki(arm, k, train_res):
user = (f"# CURRENT WIKI\n{read_wiki(arm)}\n\n# EVIDENCE TABLE (iteration {k})\n{evidence_table(train_res)}\n\n"
f"# SAMPLED TRACES\n{sample_traces(train_res)}")
print(f"[{arm}] maintainer: calling {OPT_MODEL} ({len(user)//4:,} tok prompt)...", flush=True)
t0 = time.time()
out, used = opt_llm(MAINTAINER_SYSTEM, user, arm, "maintainer")
print(f"[{arm}] maintainer: {len(out or '')} chars back from {used} in {time.time()-t0:.0f}s", flush=True)
files = sentinel_blocks(out, "FILE")
logm = re.search(r"^=== LOG\s*\n(.*)\Z", out or "", re.S | re.M)
if not files and not logm:
print(" [debug] maintainer output had no sentinels, head:", repr((out or "")[:200])); return
for fn, content in files:
fn = re.sub(r"[^a-z0-9\-]", "-", fn.strip().lower().replace(".md", "")) + ".md"
open(ws(arm, "wiki/patterns", fn), "w").write(content.strip())
print(f"[{arm}] wiki: wrote {fn} — {content.strip().splitlines()[0][:70] if content.strip() else ''}", flush=True)
idx = "\n".join(f"- {f}: {open(ws(arm,'wiki/patterns',f)).readline().strip('# \n')}"
for f in sorted(os.listdir(ws(arm, "wiki/patterns"))))
open(ws(arm, "wiki/index.md"), "w").write(idx)
open(ws(arm, "wiki/logs.md"), "a").write(f"\n## iteration {k}\n{logm.group(1).strip() if logm else ''}\n")
def propose_skill(arm, k, train_res, use_wiki):
ctx = (f"# WIKI\n{read_wiki(arm)}\n\n" if use_wiki else "")
user = (f"{ctx}# CURRENT SKILLS\n{read_skills(arm) or '(none)'}\n\n"
f"# EVIDENCE TABLE (iteration {k})\n{evidence_table(train_res)}\n\n"
f"# SAMPLED TRACES\n{sample_traces(train_res)}")
print(f"[{arm}] proposer: calling {OPT_MODEL} ({len(user)//4:,} tok prompt)...", flush=True)
t0 = time.time()
out, used = opt_llm(PROPOSER_SYSTEM, user, arm, "proposer")
print(f"[{arm}] proposer: {len(out or '')} chars back from {used} in {time.time()-t0:.0f}s", flush=True)
def grab(key):
m = re.search(rf"^=== {key}:?\s*(.*?)$", out or "", re.M); return m.group(1).strip() if m else ""
m = re.search(r"^=== CONTENT\s*\n(.*)\Z", out or "", re.S | re.M)
prop = {"skill": grab("SKILL"), "action": grab("ACTION") or "create",
"purpose": grab("PURPOSE"), "content": m.group(1).strip() if m else ""}
if not (prop["skill"] and prop["content"]):
print(" [debug] proposer output missing sentinels, head:", repr((out or "")[:200])); return None
print(f"[{arm}] proposer: {prop['action']} `{prop['skill']}` — {prop['purpose'][:90]}", flush=True)
return prop
def apply_proposal(arm, prop):
slug = re.sub(r"[^a-z0-9\-]", "-", prop["skill"].lower())
d = ws(arm, "skills", slug); os.makedirs(d, exist_ok=True)
old = open(os.path.join(d, "SKILL.md")).read() if os.path.exists(os.path.join(d, "SKILL.md")) else ""
open(os.path.join(d, "SKILL.md"), "w").write(prop["content"])
open(os.path.join(d, "PURPOSE.md"), "w").write(prop.get("purpose", ""))
diff = "".join(difflib.unified_diff(old.splitlines(1), prop["content"].splitlines(1), "old", "new"))
return slug, diff[:1500]
def log_impact(arm, k, slug, action, diff, val_score, accepted):
open(ws(arm, "wiki/skill-impact.md"), "a").write(
f"\n## iteration {k} — {action} `{slug}` — val={val_score} — {'ACCEPTED' if accepted else 'REJECTED'}\n```diff\n{diff}\n```\n")
# ======================================================================
# The evolution loop
# ======================================================================
def evolve(arm, use_wiki=True, iters=ITERS, base_val=None):
init_ws(arm)
best = mean_score(evaluate(VAL, "", INFER_MODEL)) if base_val is None else base_val
history = [("baseline", best, True)]
print(f"[{arm}] baseline val = {best}")
for k in range(iters):
skills = read_skills(arm)
train_res = evaluate(TRAIN, skills, INFER_MODEL)
save_traces(arm, k, "train", train_res)
print(f"[{arm}] iter {k}: train = {mean_score(train_res)}")
t0 = time.time()
try:
if use_wiki: maintain_wiki(arm, k, train_res)
except Exception as e:
print(f"[{arm}] maintainer failed, continuing: {type(e).__name__}: {str(e)[:100]}")
t1 = time.time()
try:
prop = propose_skill(arm, k, train_res, use_wiki)
except Exception as e:
print(f"[{arm}] proposer failed, skipping iteration: {type(e).__name__}: {str(e)[:100]}"); prop = None
print(f"[{arm}] [timing] maintainer {t1-t0:.0f}s, proposer {time.time()-t1:.0f}s")
if not prop or not all(isinstance(prop.get(f), str) and prop.get(f) for f in ("skill", "content")):
print(f"[{arm}] iter {k}: proposal malformed, skipped -> {str(prop)[:120]}"); continue
backup = ws(arm, f"_skills_backup"); shutil.rmtree(backup, ignore_errors=True)
shutil.copytree(ws(arm, "skills"), backup)
slug, diff = apply_proposal(arm, prop)
val_res = evaluate(VAL, read_skills(arm), INFER_MODEL); s = mean_score(val_res)
accepted = s > best
if accepted: best = s
else:
shutil.rmtree(ws(arm, "skills")); shutil.copytree(backup, ws(arm, "skills"))
log_impact(arm, k, slug, prop.get("action"), diff, s, accepted) # wiki is never rolled back
history.append((f"iter{k}:{slug}", s, accepted))
print(f"[{arm}] proposal {prop.get('action')} `{slug}` → val = {s} ({'ACCEPTED' if accepted else 'rejected'}, best={best})")
return read_skills(arm), history
# ======================================================================
# Run the three arms + test
# ======================================================================
results = {}
STOP.clear()
import atexit; atexit.register(STOP.set)
try:
results["baseline"] = mean_score(evaluate(TEST, "", INFER_MODEL))
BASE_VAL = mean_score(evaluate(VAL, "", INFER_MODEL)) # measured once, shared by both arms
with ThreadPoolExecutor(2) as arms: # the two arms are independent — run them side by side
f_wiki = arms.submit(evolve, "wikiskill", True, ITERS, BASE_VAL)
f_nowiki = arms.submit(evolve, "nowiki", False, ITERS, BASE_VAL)
wiki_skills, wiki_hist = f_wiki.result()
nowiki_skills, nowiki_hist = f_nowiki.result()
results["wikiskill"] = mean_score(evaluate(TEST, wiki_skills, INFER_MODEL))
results["nowiki (ablation)"] = mean_score(evaluate(TEST, nowiki_skills, INFER_MODEL))
except KeyboardInterrupt:
STOP.set()
raise SystemExit("Interrupted: stop flag set — in-flight workers will exit at their next check instead of draining the queue.")
print("\n=== TEST accuracy ===")
for k, v in results.items(): print(f"{k:22s} {v}")
# ======================================================================
# Inspect what it learned
# ======================================================================
print("=== evolved skills (wikiskill arm) ===\n", wiki_skills)
print("\n=== wiki index ===\n", open(ws("wikiskill", "wiki/index.md")).read())
print("\n=== skill-impact log ===\n", open(ws("wikiskill", "wiki/skill-impact.md")).read()[:3000])
report = ["# WikiSkill run " + HARNESS_VERSION, f"rollouts {INFER_PROVIDER}/{INFER_MODEL} · optimiser {OPT_PROVIDER}/{OPT_MODEL} · {N_TRAIN}/{N_VAL}/{N_TEST} × {ITERS}",
"", "## TEST accuracy", *[f"- {k}: {v}" for k, v in results.items()],
"", "## wikiskill history", *[f"- {n}: val={s} {'ACCEPTED' if a else 'rejected'}" for n, s, a in wiki_hist],
"## nowiki history", *[f"- {n}: val={s} {'ACCEPTED' if a else 'rejected'}" for n, s, a in nowiki_hist],
"", "## evolved skills (wikiskill)", wiki_skills or "(none accepted)",
"", "## evolved skills (nowiki)", nowiki_skills or "(none accepted)",
"", "## wiki index", open(ws("wikiskill", "wiki/index.md")).read(),
"", "## wiki patterns"]
for f in sorted(os.listdir(ws("wikiskill", "wiki/patterns"))):
report += [f"### {f}", open(ws("wikiskill", "wiki/patterns", f)).read(), ""]
report += ["## skill-impact log (wikiskill)", open(ws("wikiskill", "wiki/skill-impact.md")).read(),
"## skill-impact log (nowiki)", open(ws("nowiki", "wiki/skill-impact.md")).read()]
open(os.path.join(WS, "RESULTS.md"), "w").write("\n".join(report))
print(f"\nFull report written to {WS}/RESULTS.md (Colab file browser -> wikiskill -> RESULTS.md)")
# ======================================================================
# Optional: cross-provider transfer
# ======================================================================
if TRANSFER:
t_prov, t_model = TRANSFER
tc = get_client(t_prov)
t_base = mean_score(evaluate(TEST, "", t_model, cl=tc))
t_skill = mean_score(evaluate(TEST, wiki_skills, t_model, cl=tc))
print(f"transfer → {t_prov}:{t_model} no-skill {t_base} with-evolved-skills {t_skill}")
# WikiSkill — minimal Colab replication. FIRST LINE of output must say v49. Same as v48 (startup health check, fallbacks, limiter, report) with WORKERS=4 — no throughput loss, fewer in-flight connections.
# Secrets: NVIDIA_API_KEY + GEMINI_API_KEY. Restart session once before running.
# ======================================================================
# WikiSkill — minimal Colab replication
# ======================================================================
%pip -q install openai
import os, re, io, json, time, random, shutil, difflib, contextlib, statistics
from decimal import Decimal, ROUND_HALF_UP
from concurrent.futures import ThreadPoolExecutor
from openai import OpenAI
from google.colab import userdata
import ast as _ast
# ---- pick a provider (one flag). Keys go in Colab Secrets (key icon, left sidebar).
PROVIDERS = {
"anthropic": dict(base_url="https://api.anthropic.com/v1/", secret="ANTHROPIC_API_KEY"),
"deepseek": dict(base_url="https://api.deepseek.com", secret="DEEPSEEK_API_KEY"),
"gemini": dict(base_url="https://generativelanguage.googleapis.com/v1beta/openai/", secret="GEMINI_API_KEY"),
"openai": dict(base_url=None, secret="OPENAI_API_KEY"),
"nvidia": dict(base_url="https://integrate.api.nvidia.com/v1", secret="NVIDIA_API_KEY"),
"cerebras": dict(base_url="https://api.cerebras.ai/v1", secret="CEREBRAS_API_KEY"),
"groq": dict(base_url="https://api.groq.com/openai/v1", secret="GROQ_API_KEY"),
"mistral": dict(base_url="https://api.mistral.ai/v1", secret="MISTRAL_API_KEY"),
"sambanova": dict(base_url="https://api.sambanova.ai/v1", secret="SAMBANOVA_API_KEY"),
}
# roles can live on different providers so free-tier rate caps add up
# NVIDIA free tier caps REQUESTS (~40/min) not tokens — the right shape for token-heavy agent loops.
# Groq free tier caps tokens (8k TPM on gpt-oss-20b): too small for this workload, kept here as reference.
HARNESS_VERSION = "v49"
print(f"=== WikiSkill harness {HARNESS_VERSION} ===")
import threading as _th
_zombies = [t for t in _th.enumerate() if t.name.startswith("ThreadPoolExecutor")]
if _zombies:
raise SystemExit(f"REFUSING TO START: {len(_zombies)} worker threads from an interrupted run are still alive and calling the API. "
"Runtime -> Restart session (or Disconnect and delete runtime), then run this cell again.")
INFER_PROVIDER = "nvidia"
INFER_MODEL = "google/gemma-4-31b-it" # rollouts: plain non-reasoning instruction-follower — short outputs, no thinking channel
INFER_FALLBACKS = ["nvidia:deepseek-ai/deepseek-v4-flash-0731", "nvidia:nvidia/nemotron-3.5-lightning-30b-a3b",
"nvidia:google/gemma-3-12b-it", "nvidia:mistralai/mistral-nemotron", "nvidia:microsoft/phi-3.5-moe-instruct",
"gemini:gemini-flash-lite-latest"] # probed at startup (first live one wins) and used per call if the chosen one dies
# reasoning alternative: "nvidia/nemotron-3.5-lightning-30b-a3b" (thinking-on; slower, answers can land in reasoning_content)
OPT_PROVIDER = "gemini" # analyst on separate infrastructure: ~20 calls/run sits well inside Gemini's free tier
OPT_MODEL = "gemini-3.5-flash" # an older Flash generation: separate free-tier quota bucket, far less demand than the newest
OPT_FALLBACKS = ["gemini-flash-lite-latest", "gemini-3.1-flash-lite", "gemini-flash-latest",
"nvidia:google/gemma-4-31b-it"] # "provider:model" entries switch provider — last resort is off-Google entirely
# all-NVIDIA fallback: OPT_PROVIDER="nvidia", OPT_MODEL="deepseek-ai/deepseek-v4-flash-0731" or "google/gemma-4-31b-it"
# NOT "moonshotai/kimi-k2.6": listed in the catalog but its NIM backend 404s ("Function ... Not Found")
# fallbacks: "deepseek-ai/deepseek-v4-pro-0813", "nvidia/nemotron-3-super-120b-a12b" (thinking off)
# avoid "mistralai/mistral-large-2-instruct" — its NIM endpoint returned empty completions
# gemini alternative: OPT_PROVIDER="gemini", OPT_MODEL="gemini-2.5-flash" (GEMINI_API_KEY from aistudio.google.com)
# thinking flags are Nemotron-specific — gate on the MODEL, never the provider
INFER_EXTRA = {"chat_template_kwargs": {"enable_thinking": True}} if "nemotron" in INFER_MODEL else None # native mode; False half-breaks the 3.5 template
OPT_EXTRA = {"chat_template_kwargs": {"enable_thinking": False}} if "nemotron" in OPT_MODEL else None
OPT_MAX_TOKENS = 3000 # brevity is enforced in the prompts; long generations were the slow part
# deepseek → "deepseek-chat" both roles; anthropic → "claude-haiku-4-5" / "claude-sonnet-4-5"
# nvidia catalog is large — list live ids with the optional cell below; some model families need one-time
# registration on their build.nvidia.com page ("Try API") or you get a 403.
# gemini → "gemini-2.5-flash" / "gemini-2.5-pro"
# cross-provider transfer test: evaluate the evolved skills with a different provider+model, e.g. ("deepseek", "deepseek-chat")
TRANSFER = None
N_TRAIN, N_VAL, N_TEST = 16, 8, 10 # more training evidence for rule decomposition; rollouts are cheap now
ITERS = 3
WORKERS = 4 # per arm. 4 x 2 arms = 8 in flight already saturates RPM_CAP; fewer open connections is kinder to a struggling backend
RPM_CAP = 36 # just under NVIDIA's ~40 req/min free-tier cap
SEED = 0
_clients = {}
def get_client(provider):
if provider not in _clients:
p = PROVIDERS[provider]
_clients[provider] = OpenAI(api_key=userdata.get(p["secret"]), base_url=p["base_url"],
timeout=120, max_retries=0) # fail fast; our own loop retries. Default was a silent 10-minute hang.
return _clients[provider]
client = get_client(INFER_PROVIDER) # default client: rollouts
opt_client = get_client(OPT_PROVIDER)
def probe(provider, model, timeout=25):
"""Is this endpoint answering right now? Tiny request, short timeout."""
try:
r = get_client(provider).with_options(timeout=timeout).chat.completions.create(
model=model, messages=[{"role": "user", "content": "Reply with OK."}], max_tokens=5)
return bool((r.choices[0].message.content or getattr(r.choices[0].message, "reasoning_content", "") or "").strip())
except Exception as e:
print(f" probe {provider}/{model}: {type(e).__name__}"); return False
def pick_live(primary, fallbacks, default_provider, role):
for entry in [primary] + fallbacks:
prov, _, m = entry.partition(":") if ":" in entry else (default_provider, None, entry)
if probe(prov, m):
print(f" {role}: {prov}/{m} is live"); return prov, m
print(f" {role}: nothing answered the probe — keeping {primary}"); return default_provider, primary
INFER_PROVIDER, INFER_MODEL = pick_live(INFER_MODEL, INFER_FALLBACKS, INFER_PROVIDER, "rollouts")
OPT_PROVIDER, OPT_MODEL = pick_live(OPT_MODEL, OPT_FALLBACKS, OPT_PROVIDER, "optimiser")
client, opt_client = get_client(INFER_PROVIDER), get_client(OPT_PROVIDER)
INFER_EXTRA = {"chat_template_kwargs": {"enable_thinking": True}} if "nemotron" in INFER_MODEL else None
OPT_EXTRA = {"chat_template_kwargs": {"enable_thinking": False}} if "nemotron" in OPT_MODEL else None
print(f"rollouts: {INFER_PROVIDER}/{INFER_MODEL} | optimiser: {OPT_PROVIDER}/{OPT_MODEL} | "
f"splits {N_TRAIN}/{N_VAL}/{N_TEST} x {ITERS} iters | workers {WORKERS}")
import threading, collections as _collections
STOP = threading.Event() # set on interrupt; workers check it and bail out instead of draining their queue
_rl_lock = threading.Lock(); _rl_times = _collections.deque()
def rate_limit():
"""Token bucket shared by every thread: at most RPM_CAP requests start per rolling 60s."""
while True:
with _rl_lock:
now = time.time()
while _rl_times and now - _rl_times[0] > 60: _rl_times.popleft()
if len(_rl_times) < RPM_CAP:
_rl_times.append(now); return
wait = 60 - (now - _rl_times[0]) + 0.05
time.sleep(max(wait, 0.05))
PY_TOOL = [{"type": "function", "function": {
"name": "python",
"description": "Execute Python code and return its stdout. print() anything you need to see.",
"parameters": {"type": "object", "properties": {"code": {"type": "string", "description": "Python source to execute"}}, "required": ["code"]}}}]
def _salvage_tool_call(e):
"""Groq rejects GPT-OSS tool calls whose arguments aren't valid JSON, but includes the raw
generation in the error. The code the model wanted to run is in there — recover it."""
body = getattr(e, "body", None)
fg = (body.get("error") or {}).get("failed_generation") if isinstance(body, dict) else None
if not fg:
m = re.search(r"failed_generation.: .(.*)", str(e), re.S)
fg = m.group(1) if m else None
if not fg: return None
m = re.search(r'"arguments"\s*:\s*(.*)', fg, re.S)
if not m: return None
code = m.group(1).strip()
for tail in ("'}", "}"):
if code.endswith(tail): code = code[: -len(tail)].rstrip()
code = code.strip('"').strip()
if code.startswith("{"):
try: code = json.loads(code).get("code", "")
except Exception: pass
if not code: return None
tc = type("TC", (), {})(); tc.id = "salvaged_0"
tc.function = type("F", (), {})(); tc.function.name = "python"; tc.function.arguments = json.dumps({"code": code})
msg = type("M", (), {})(); msg.content = ""; msg.tool_calls = [tc]
return msg
def chat_full(messages, model, temperature=0.0, max_tokens=2500, cl=None, extra_body=None, tools=None):
cl = cl or client; err = None
for attempt in range(8): # unattended-safe: more patience, longer backoffs
if STOP.is_set(): raise RuntimeError("stopped")
try:
kw = dict(model=model, messages=messages, temperature=temperature, max_tokens=max_tokens)
if extra_body: kw["extra_body"] = extra_body
if tools: kw["tools"] = tools
rate_limit()
r = cl.chat.completions.create(**kw)
m = r.choices[0].message
if not getattr(m, "tool_calls", None) and not (m.content or "").strip():
rc = (getattr(m, "reasoning_content", None) or "").strip()
if rc: # hybrid-reasoning models can put the whole answer in the reasoning channel
m.content = rc
return m
fr = getattr(r.choices[0], "finish_reason", "?")
print(f" [debug] empty completion from {model} (finish_reason={fr}, no reasoning_content)")
if len(messages) == 2 and messages[0]["role"] == "system": # some serving templates mishandle the system role
messages = [{"role": "user", "content": f"[Instructions]\n{messages[0]['content']}\n\n[Input]\n{messages[1]['content']}"}]
continue
temperature = 0.4; continue
return m
except Exception as e:
err = e
msg = str(e).lower()
print(f" [debug] {model} call failed (attempt {attempt+1}/8): {type(e).__name__}: {str(e)[:120]}")
if "timed out" in msg or "timeout" in type(e).__name__.lower():
timeouts = locals().get("timeouts", 0) + 1
if timeouts >= 2: raise RuntimeError(f"{model}: endpoint unresponsive (2 consecutive timeouts)") from e
if extra_body and ("400" in msg or "invalid" in msg or "unsupported" in msg or "unknown" in msg) and "tool" not in msg:
extra_body = None; continue # provider rejected an optional param — retry clean
if "tool_use_failed" in msg or ("tool" in msg and "400" in msg):
sal = _salvage_tool_call(e)
if sal: return sal # recover the code from the rejected generation
tools = None # endpoint may not support tools at all — degrade to the markdown protocol
temperature = 0.5; continue
if "402" in msg or "payment_required" in msg:
raise RuntimeError(f"{model}: not covered by this provider's free quota — pick a model from the account's usage/limits page") from e
if "model_not_found" in msg or "does not exist" in msg or "404" in msg or "not found" in msg:
raise RuntimeError(f"{model}: 404 from provider (not in catalog, or listed but not deployed) — pick another id") from e
time.sleep((10 if any(s in msg for s in ("429", "rate", "503", "high demand", "overloaded", "529")) else 2) * (attempt + 1))
raise err
def chat(messages, model, **kw):
return chat_full(messages, model, **kw).content or ""
def tc_code(tc):
a = tc.function.arguments or ""
try: return json.loads(a).get("code", "")
except Exception: return a # some models emit raw code as the arguments string
def serialize_assistant(m):
d = {"role": "assistant", "content": m.content or ""}
if getattr(m, "tool_calls", None):
d["tool_calls"] = [{"id": tc.id, "type": "function",
"function": {"name": tc.function.name, "arguments": tc.function.arguments}}
for tc in m.tool_calls]
return d
def llm(system, user, model, **kw): # kw may include cl= to route to a specific provider
return chat([{"role": "system", "content": system}, {"role": "user", "content": user}], model, **kw)
def parse_json(text):
text = text.strip()
fences = re.findall(r"```json\s*(.*?)```", text, re.S) # prefer explicit json fences, take the LAST
if not fences:
fences = [f for f in re.findall(r"```\s*(.*?)```", text, re.S) if f.lstrip().startswith("{")]
if fences: text = fences[-1]
start, end = text.find("{"), text.rfind("}")
blob = text[start:end + 1]
try:
return json.loads(blob)
except Exception:
return _ast.literal_eval(blob) # tolerate single-quoted / Python-literal dicts
# live model catalogs — set INFER_MODEL / OPT_MODEL from these lists (a listing is a claim, not a capability: probe before trusting)
for label, cl_, filt in (("rollout provider", client, None), ("optimiser provider", opt_client, "gemini")):
try:
ids = sorted({m.id for m in cl_.models.list().data})
if filt: ids = [i for i in ids if filt in i.lower()]
print(f"{label} — models available to this key:")
for i in ids[:40]: print(" ", i)
except Exception as e:
print(f"{label} listing failed:", e)
print("\nBEFORE A FULL RUN: Runtime -> Restart session (interrupted runs leave worker threads alive, still calling the API).")
# ======================================================================
# Tasks (synthetic, deterministic ground truth, hidden rules)
# ======================================================================
ITEMS = [("kopi-o", "1.40"), ("teh tarik", "1.70"), ("Milo dinosaur", "3.20"),
("kaya toast set", "3.30"), ("half-boiled eggs (pair)", "1.90"), ("nasi lemak", "4.00"),
("chicken rice", "4.50"), ("char kway teow", "5.00"), ("mee siam", "3.50"),
("curry puff", "1.80"), ("otah", "1.20"), ("bandung", "1.60")]
def make_task(i, rng):
n = rng.randint(2, 4)
lines = [(name, rng.choice([1, 2, 3, 4, 5, 6, 8, 10, 12, 15]), Decimal(p))
for name, p in rng.sample(ITEMS, n)]
member = rng.random() < 0.5
sub = Decimal("0")
for _, qty, price in lines:
line = price * qty
if qty >= 10: line *= Decimal("0.95") # HIDDEN rule 1: bulk 5% off lines with qty >= 10
sub += line
if member: sub *= Decimal("0.97") # HIDDEN rule 2: members 3% off subtotal
delivery = Decimal("8") if sub < 50 else Decimal("0") # stated
total = ((sub + delivery) * Decimal("1.09")).quantize(Decimal("0.01"), ROUND_HALF_UP) # stated
order = "; ".join(f"{qty} x {name} @ S${price} each" for name, qty, price in lines)
prompt = (f"Prepare the final bill for a kopitiam catering order in Singapore.\n"
f"Order: {order}.\n"
f"Customer is {'a loyalty member' if member else 'not a member'}.\n"
f"Store policy: delivery is S$8.00 if the discounted subtotal is below S$50.00, otherwise free. "
f"GST of 9% applies to subtotal plus delivery. Round the final amount half-up to the nearest cent.\n"
f"Give the amount the customer pays. Your last line must be exactly:\n"
f"FINAL: TOTAL: SGD <amount with 2 decimals>")
return {"id": f"t{i:03d}", "prompt": prompt, "answer": f"{total:.2f}"}
def grade(pred, answer):
m = re.search(r"SGD\s*\$?\s*([\d,]+\.\d{2})", pred or "")
return 1.0 if m and m.group(1).replace(",", "") == answer else 0.0
rng = random.Random(SEED)
ALL = [make_task(i, rng) for i in range(N_TRAIN + N_VAL + N_TEST)]
def _has_hidden(t): # any bulk-qty line, or a member
return bool(re.search(r"\b(10|12|15) x", t["prompt"])) or "not a member" not in t["prompt"]
easy = [t for t in ALL if not _has_hidden(t)]; hard = [t for t in ALL if _has_hidden(t)]
mixed = [] # interleave so each split inherits the same easy/hard ratio
while easy or hard:
for _ in range(max(1, round(len(hard) / max(1, len(easy))))):
if hard: mixed.append(hard.pop(0))
if easy: mixed.append(easy.pop(0))
TRAIN, VAL, TEST = mixed[:N_TRAIN], mixed[N_TRAIN:N_TRAIN + N_VAL], mixed[N_TRAIN + N_VAL:]
print(TRAIN[0]["prompt"], "\n-> answer:", TRAIN[0]["answer"])
# ======================================================================
# Inference Agent (ReAct with a Python tool)
# ======================================================================
AGENT_SYSTEM = """You are a careful kopitiam billing assistant.
Use the python tool to run code (or, if you cannot call tools, write exactly one ```python code block).
Do the arithmetic in Python, never in your head, and make your code PRINT the final line itself, exactly: TOTAL: SGD <amount with 2 decimals>. Be brief: no explanations, no restating the task. If you must answer without code, reply with one last line starting with 'FINAL: ' containing the actual amount. Never write a FINAL: line with placeholder text."""
def run_python(code):
buf = io.StringIO()
g = {"__name__": "__main__"}
try:
with contextlib.redirect_stdout(buf):
try:
tree = _ast.parse(code)
except SyntaxError as e:
print(f"[error] SyntaxError: {e}"); tree = None
if tree is not None:
if tree.body and isinstance(tree.body[-1], _ast.Expr): # Jupyter-style: echo a trailing bare expression
exec(compile(_ast.Module(body=tree.body[:-1], type_ignores=[]), "<cell>", "exec"), g)
val = eval(compile(_ast.Expression(tree.body[-1].value), "<cell>", "eval"), g)
if val is not None: print(repr(val))
else:
exec(compile(tree, "<cell>", "exec"), g)
except Exception as e:
buf.write(f"\n[error] {type(e).__name__}: {e}")
return buf.getvalue()[-3000:] or "(no output)"
def run_agent(task, skills_text, model, max_steps=3, cl=None):
if STOP.is_set():
return {"id": task["id"], "prompt": task["prompt"], "pred": "", "gt": task["answer"], "score": 0.0, "trace": []}
system = AGENT_SYSTEM + (f"\n\n# SKILLS — follow these procedures\n{skills_text}" if skills_text else "")
msgs = [{"role": "system", "content": system}, {"role": "user", "content": task["prompt"]}]
trace, pred = [], ""
for _ in range(max_steps):
m = None
for entry in [f"{INFER_PROVIDER}:{model}"] + (INFER_FALLBACKS if cl is None else []):
prov, _, mdl = entry.partition(":")
try:
m = chat_full(msgs, mdl, cl=(cl or get_client(prov)),
extra_body=({"chat_template_kwargs": {"enable_thinking": True}} if "nemotron" in mdl else None), tools=PY_TOOL)
if mdl != model: trace.append({"fallback_model": f"{prov}/{mdl}"})
break
except Exception as e:
print(f" rollout: {prov}/{mdl} unavailable ({type(e).__name__}); trying next", flush=True)
if m is None: break
if getattr(m, "tool_calls", None):
msgs.append(serialize_assistant(m))
done = None
for tc in m.tool_calls:
res = run_python(tc_code(tc))
msgs.append({"role": "tool", "tool_call_id": tc.id, "content": res})
trace.append({"tool_call": tc_code(tc), "tool_output": res})
hit = re.findall(r"TOTAL:\s*SGD\s*[\d,]+\.\d{2}", res)
if hit: done = hit[-1]
if done: pred = done; break # single-shot: the code printed the answer
continue
out = m.content or ""
msgs.append({"role": "assistant", "content": out}); trace.append({"assistant": out})
code = re.search(r"```python\n(.*?)```", out, re.S)
if code: # text-protocol fallback for models that don't call tools
res = run_python(code.group(1))
trace.append({"tool_output": res})
hit = re.findall(r"TOTAL:\s*SGD\s*[\d,]+\.\d{2}", res)
if hit: pred = hit[-1]; break # single-shot: the code printed the answer
msgs.append({"role": "user", "content": f"[python stdout]\n{res}\n\nContinue. Finish with a FINAL: line."})
continue
finals = [f.strip() for f in re.findall(r"FINAL:\s*(.*)", out) if "<" not in f]
if finals: pred = finals[-1]; break
msgs.append({"role": "user", "content": "No code and no usable FINAL line found. Run Python or give the FINAL: line with the actual amount."})
return {"id": task["id"], "prompt": task["prompt"], "pred": pred, "gt": task["answer"],
"score": grade(pred, task["answer"]), "trace": trace}
def evaluate(tasks, skills_text, model, cl=None):
print(f" rollouts: {len(tasks)} tasks, skills={'yes' if skills_text else 'none'}...", flush=True)
t0 = time.time()
with ThreadPoolExecutor(WORKERS) as ex:
res = list(ex.map(lambda t: run_agent(t, skills_text, model, cl=cl), tasks))
print(f" [timing] {len(tasks)} rollouts in {time.time()-t0:.0f}s")
return res
def mean_score(results):
return round(100 * statistics.mean(r["score"] for r in results), 1)
# ======================================================================
# Workspace: raw/ · wiki/ · skills/
# ======================================================================
WS = "/content/wikiskill"
def ws(arm, *p): return os.path.join(WS, arm, *p)
def init_ws(arm):
shutil.rmtree(ws(arm), ignore_errors=True)
for d in ["raw", "wiki/patterns", "skills"]: os.makedirs(ws(arm, d))
for f in ["wiki/index.md", "wiki/logs.md", "wiki/skill-impact.md"]:
open(ws(arm, f), "w").write("")
def save_traces(arm, k, split, results):
for r in results:
json.dump(r, open(ws(arm, "raw", f"iter{k}_{split}_{r['id']}.json"), "w"), indent=1)
def read_skills(arm):
out = []
for name in sorted(os.listdir(ws(arm, "skills"))):
p = ws(arm, "skills", name, "SKILL.md")
if os.path.exists(p): out.append(f"## skill: {name}\n" + open(p).read())
return "\n\n".join(out)
def read_wiki(arm):
parts = ["# wiki/index.md\n" + open(ws(arm, "wiki/index.md")).read()]
for f in sorted(os.listdir(ws(arm, "wiki/patterns"))):
parts.append(f"# wiki/patterns/{f}\n" + open(ws(arm, "wiki/patterns", f)).read())
parts.append("# wiki/logs.md\n" + open(ws(arm, "wiki/logs.md")).read())
parts.append("# wiki/skill-impact.md\n" + open(ws(arm, "wiki/skill-impact.md")).read())
return "\n\n".join(parts)
def fmt_trace(r, max_chars=1200):
body = "\n".join(f"[{k}] {v}" for step in r["trace"] for k, v in step.items())
return (f"### {r['id']} — {'PASS' if r['score'] else 'FAIL'} | pred={r['pred']!r} | gt={r['gt']}\n"
f"TASK: {r['prompt']}\n{body[:max_chars]}")
def sample_traces(results, n_fail=3, n_pass=1):
fails = [r for r in results if not r["score"]][:n_fail]
passes = [r for r in results if r["score"]][:n_pass]
return "\n\n".join(fmt_trace(r) for r in fails + passes)
def outcome_summary(results):
return "\n".join(f"{r['id']}: {'PASS' if r['score'] else 'FAIL'} pred={r['pred']!r} gt={r['gt']}" for r in results)
# ======================================================================
# Wiki Maintainer + Skill Proposer + Gate
# ======================================================================
MAINTAINER_SYSTEM = """You are the Wiki Maintainer for an agent-skill evolution system.
From the evidence table and traces, do root-cause analysis of FAILED tasks and extract what worked from PASSED ones.
Each pattern page documents ONE failure mode or strategy: evidence (task ids), root cause, actionable workaround.
Update existing pages (rewrite in full) or add new ones. Do not delete knowledge. Ground every claim in the table; never invent numbers. Be concise: each page under 120 words, at most 3 pages per iteration.
Output format — plain text, EXACTLY these sentinels, no JSON:
=== FILE: kebab-slug.md
<full page content>
(repeat === FILE: blocks as needed)
=== LOG
<one-paragraph summary of this iteration's findings>"""
PROPOSER_SYSTEM = """You are the Skill Proposer for an agent-skill evolution system.
Propose exactly ONE atomic change: create a new skill, or edit one existing skill (give its full new content).
Skills are concise procedural instructions the billing agent follows verbatim from its system prompt. Keep the skill under 200 words: numbered steps, no commentary.
Ground every number in the evidence table (the ratio column shows how ground truth deviates from the naive stated-rules total). Never invent a number. If ratios vary across tasks, do not average them into an approximate rate: look for an exact rule that reproduces EVERY ground truth — it may depend on individual line quantities, not only on membership — and reject any hypothesis that fails even one task. If a wiki and impact history are given, build on them and NEVER re-propose a rejected change.
Output format — plain text, EXACTLY these sentinels, no JSON:
=== SKILL: kebab-slug
=== ACTION: create or edit
=== PURPOSE: <which evidence motivated this>
=== CONTENT
<full SKILL.md markdown>"""
def opt_llm(system, user, arm, role):
"""Optimiser call with a fallback chain; entries may be 'model' (optimiser provider) or 'provider:model'. Returns (text, model_used)."""
for entry in [OPT_MODEL] + OPT_FALLBACKS:
prov, _, m = entry.partition(":") if ":" in entry else (OPT_PROVIDER, None, entry)
try:
return llm(system, user, m, max_tokens=OPT_MAX_TOKENS, cl=get_client(prov),
extra_body=({"chat_template_kwargs": {"enable_thinking": False}} if "nemotron" in m else None)), f"{prov}/{m}"
except Exception as e:
print(f"[{arm}] {role}: {prov}/{m} gave up ({type(e).__name__}); trying next fallback", flush=True)
raise RuntimeError(f"{role}: all optimiser models failed")
def evidence_table(results):
"""Pre-computed analysis substrate: per task, parsed lines, member flag, naive stated-rules total,
ground truth, prediction, and gt/naive ratio. Harness-side feature engineering so the optimiser
can infer hidden rules by inspection instead of needing a Python tool."""
rows = ["id | lines (qty x price) | member | naive_total | ground_truth | predicted | pass | gt/naive"]
for r in results:
lines = re.findall(r"(\d+) x [^@]+ @ S\$(\d+\.\d+)", r["prompt"])
member = "not a member" not in r["prompt"]
sub = sum(Decimal(p) * int(q) for q, p in lines)
delivery = Decimal("8") if sub < 50 else Decimal("0")
naive = ((sub + delivery) * Decimal("1.09")).quantize(Decimal("0.01"), ROUND_HALF_UP)
ratio = (Decimal(r["gt"]) / naive).quantize(Decimal("0.0001"))
rows.append(f'{r["id"]} | {", ".join(f"{q}x{p}" for q, p in lines)} | {"Y" if member else "N"} | '
f'{naive} | {r["gt"]} | {r["pred"] or "(none)"} | {"P" if r["score"] else "F"} | {ratio}')
return "\n".join(rows)
def sentinel_blocks(text, key):
return re.findall(rf"^=== {key}:?\s*(.*?)\n(.*?)(?=^=== |\Z)", text or "", re.S | re.M)
def maintain_wiki(arm, k, train_res):
user = (f"# CURRENT WIKI\n{read_wiki(arm)}\n\n# EVIDENCE TABLE (iteration {k})\n{evidence_table(train_res)}\n\n"
f"# SAMPLED TRACES\n{sample_traces(train_res)}")
print(f"[{arm}] maintainer: calling {OPT_MODEL} ({len(user)//4:,} tok prompt)...", flush=True)
t0 = time.time()
out, used = opt_llm(MAINTAINER_SYSTEM, user, arm, "maintainer")
print(f"[{arm}] maintainer: {len(out or '')} chars back from {used} in {time.time()-t0:.0f}s", flush=True)
files = sentinel_blocks(out, "FILE")
logm = re.search(r"^=== LOG\s*\n(.*)\Z", out or "", re.S | re.M)
if not files and not logm:
print(" [debug] maintainer output had no sentinels, head:", repr((out or "")[:200])); return
for fn, content in files:
fn = re.sub(r"[^a-z0-9\-]", "-", fn.strip().lower().replace(".md", "")) + ".md"
open(ws(arm, "wiki/patterns", fn), "w").write(content.strip())
print(f"[{arm}] wiki: wrote {fn} — {content.strip().splitlines()[0][:70] if content.strip() else ''}", flush=True)
idx = "\n".join(f"- {f}: {open(ws(arm,'wiki/patterns',f)).readline().strip('# \n')}"
for f in sorted(os.listdir(ws(arm, "wiki/patterns"))))
open(ws(arm, "wiki/index.md"), "w").write(idx)
open(ws(arm, "wiki/logs.md"), "a").write(f"\n## iteration {k}\n{logm.group(1).strip() if logm else ''}\n")
def propose_skill(arm, k, train_res, use_wiki):
ctx = (f"# WIKI\n{read_wiki(arm)}\n\n" if use_wiki else "")
user = (f"{ctx}# CURRENT SKILLS\n{read_skills(arm) or '(none)'}\n\n"
f"# EVIDENCE TABLE (iteration {k})\n{evidence_table(train_res)}\n\n"
f"# SAMPLED TRACES\n{sample_traces(train_res)}")
print(f"[{arm}] proposer: calling {OPT_MODEL} ({len(user)//4:,} tok prompt)...", flush=True)
t0 = time.time()
out, used = opt_llm(PROPOSER_SYSTEM, user, arm, "proposer")
print(f"[{arm}] proposer: {len(out or '')} chars back from {used} in {time.time()-t0:.0f}s", flush=True)
def grab(key):
m = re.search(rf"^=== {key}:?\s*(.*?)$", out or "", re.M); return m.group(1).strip() if m else ""
m = re.search(r"^=== CONTENT\s*\n(.*)\Z", out or "", re.S | re.M)
prop = {"skill": grab("SKILL"), "action": grab("ACTION") or "create",
"purpose": grab("PURPOSE"), "content": m.group(1).strip() if m else ""}
if not (prop["skill"] and prop["content"]):
print(" [debug] proposer output missing sentinels, head:", repr((out or "")[:200])); return None
print(f"[{arm}] proposer: {prop['action']} `{prop['skill']}` — {prop['purpose'][:90]}", flush=True)
return prop
def apply_proposal(arm, prop):
slug = re.sub(r"[^a-z0-9\-]", "-", prop["skill"].lower())
d = ws(arm, "skills", slug); os.makedirs(d, exist_ok=True)
old = open(os.path.join(d, "SKILL.md")).read() if os.path.exists(os.path.join(d, "SKILL.md")) else ""
open(os.path.join(d, "SKILL.md"), "w").write(prop["content"])
open(os.path.join(d, "PURPOSE.md"), "w").write(prop.get("purpose", ""))
diff = "".join(difflib.unified_diff(old.splitlines(1), prop["content"].splitlines(1), "old", "new"))
return slug, diff[:1500]
def log_impact(arm, k, slug, action, diff, val_score, accepted):
open(ws(arm, "wiki/skill-impact.md"), "a").write(
f"\n## iteration {k} — {action} `{slug}` — val={val_score} — {'ACCEPTED' if accepted else 'REJECTED'}\n```diff\n{diff}\n```\n")
# ======================================================================
# The evolution loop
# ======================================================================
def evolve(arm, use_wiki=True, iters=ITERS, base_val=None):
init_ws(arm)
best = mean_score(evaluate(VAL, "", INFER_MODEL)) if base_val is None else base_val
history = [("baseline", best, True)]
print(f"[{arm}] baseline val = {best}")
for k in range(iters):
skills = read_skills(arm)
train_res = evaluate(TRAIN, skills, INFER_MODEL)
save_traces(arm, k, "train", train_res)
print(f"[{arm}] iter {k}: train = {mean_score(train_res)}")
t0 = time.time()
try:
if use_wiki: maintain_wiki(arm, k, train_res)
except Exception as e:
print(f"[{arm}] maintainer failed, continuing: {type(e).__name__}: {str(e)[:100]}")
t1 = time.time()
try:
prop = propose_skill(arm, k, train_res, use_wiki)
except Exception as e:
print(f"[{arm}] proposer failed, skipping iteration: {type(e).__name__}: {str(e)[:100]}"); prop = None
print(f"[{arm}] [timing] maintainer {t1-t0:.0f}s, proposer {time.time()-t1:.0f}s")
if not prop or not all(isinstance(prop.get(f), str) and prop.get(f) for f in ("skill", "content")):
print(f"[{arm}] iter {k}: proposal malformed, skipped -> {str(prop)[:120]}"); continue
backup = ws(arm, f"_skills_backup"); shutil.rmtree(backup, ignore_errors=True)
shutil.copytree(ws(arm, "skills"), backup)
slug, diff = apply_proposal(arm, prop)
val_res = evaluate(VAL, read_skills(arm), INFER_MODEL); s = mean_score(val_res)
accepted = s > best
if accepted: best = s
else:
shutil.rmtree(ws(arm, "skills")); shutil.copytree(backup, ws(arm, "skills"))
log_impact(arm, k, slug, prop.get("action"), diff, s, accepted) # wiki is never rolled back
history.append((f"iter{k}:{slug}", s, accepted))
print(f"[{arm}] proposal {prop.get('action')} `{slug}` → val = {s} ({'ACCEPTED' if accepted else 'rejected'}, best={best})")
return read_skills(arm), history
# ======================================================================
# Run the three arms + test
# ======================================================================
results = {}
STOP.clear()
import atexit; atexit.register(STOP.set)
try:
results["baseline"] = mean_score(evaluate(TEST, "", INFER_MODEL))
BASE_VAL = mean_score(evaluate(VAL, "", INFER_MODEL)) # measured once, shared by both arms
with ThreadPoolExecutor(2) as arms: # the two arms are independent — run them side by side
f_wiki = arms.submit(evolve, "wikiskill", True, ITERS, BASE_VAL)
f_nowiki = arms.submit(evolve, "nowiki", False, ITERS, BASE_VAL)
wiki_skills, wiki_hist = f_wiki.result()
nowiki_skills, nowiki_hist = f_nowiki.result()
results["wikiskill"] = mean_score(evaluate(TEST, wiki_skills, INFER_MODEL))
results["nowiki (ablation)"] = mean_score(evaluate(TEST, nowiki_skills, INFER_MODEL))
except KeyboardInterrupt:
STOP.set()
raise SystemExit("Interrupted: stop flag set — in-flight workers will exit at their next check instead of draining the queue.")
print("\n=== TEST accuracy ===")
for k, v in results.items(): print(f"{k:22s} {v}")
# ======================================================================
# Inspect what it learned
# ======================================================================
print("=== evolved skills (wikiskill arm) ===\n", wiki_skills)
print("\n=== wiki index ===\n", open(ws("wikiskill", "wiki/index.md")).read())
print("\n=== skill-impact log ===\n", open(ws("wikiskill", "wiki/skill-impact.md")).read()[:3000])
report = ["# WikiSkill run " + HARNESS_VERSION, f"rollouts {INFER_PROVIDER}/{INFER_MODEL} · optimiser {OPT_PROVIDER}/{OPT_MODEL} · {N_TRAIN}/{N_VAL}/{N_TEST} × {ITERS}",
"", "## TEST accuracy", *[f"- {k}: {v}" for k, v in results.items()],
"", "## wikiskill history", *[f"- {n}: val={s} {'ACCEPTED' if a else 'rejected'}" for n, s, a in wiki_hist],
"## nowiki history", *[f"- {n}: val={s} {'ACCEPTED' if a else 'rejected'}" for n, s, a in nowiki_hist],
"", "## evolved skills (wikiskill)", wiki_skills or "(none accepted)",
"", "## evolved skills (nowiki)", nowiki_skills or "(none accepted)",
"", "## wiki index", open(ws("wikiskill", "wiki/index.md")).read(),
"", "## wiki patterns"]
for f in sorted(os.listdir(ws("wikiskill", "wiki/patterns"))):
report += [f"### {f}", open(ws("wikiskill", "wiki/patterns", f)).read(), ""]
report += ["## skill-impact log (wikiskill)", open(ws("wikiskill", "wiki/skill-impact.md")).read(),
"## skill-impact log (nowiki)", open(ws("nowiki", "wiki/skill-impact.md")).read()]
open(os.path.join(WS, "RESULTS.md"), "w").write("\n".join(report))
print(f"\nFull report written to {WS}/RESULTS.md (Colab file browser -> wikiskill -> RESULTS.md)")
# ======================================================================
# Optional: cross-provider transfer
# ======================================================================
if TRANSFER:
t_prov, t_model = TRANSFER
tc = get_client(t_prov)
t_base = mean_score(evaluate(TEST, "", t_model, cl=tc))
t_skill = mean_score(evaluate(TEST, wiki_skills, t_model, cl=tc))
print(f"transfer → {t_prov}:{t_model} no-skill {t_base} with-evolved-skills {t_skill}")
原文链接: Beyond Code: A New Artefact and a Second Pipeline for AI-DLC
汇智网翻译整理，转载请标明出处
