企业官网建设流程全解析

agno Environments SQL 生成任务集：用内存 SQLite 夹具做"执行结果评分"的完整实战

【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno

本文基于 agno 仓库cookbook/environments/_22_sql_generation/目录，讲解如何用可执行评分（executable scoring）评测"模型生成 SQL"这类存在大量等价正确解的生成任务：把模型输出的类型化 SQL 字符串直接扔进只读内存 SQLite 夹具执行，只比较返回结果行、不比较查询措辞。读完本文，你可以掌握 agno 的Environment/Task/CodeScorer/run_rollouts完整调用链，并拿到三个可复制运行的递归 CTE、多表 Join、窗口函数 SQL 评测样例。

设计哲学：对"结果行"评分，而不是对"查询措辞"评分

目录文档 README.md 开宗明义：

Generate a typed SQL string, execute it against an in-memory fixture, and score the result rows rather than the query's wording.

核心问题是：对同一个自然语言需求，存在许多语义等价的 SQL 写法（CTE 拆法不同、子查询与 Join 互换、ORDER BY 等价形式……），如果做精确文本比对，会把大量有效解误判为错误。因此这个任务集采用"可执行评分"：

1. Agent 按output_schema输出一个类型化的 SQL 字符串（Pydantic 模型包裹sql: str字段）；
2. 评分器在私有的:memory:SQLite 连接上执行建表脚本（fixture），然后执行模型生成的查询；
3. 把返回的结果行与预计算的expected["rows"]做逐行相等比较，行对了即满分。

文档同时说明了适用边界：

Use executable scoring when many SQL strings can be correct and exact text comparison would reject valid alternatives. These examples use read-only in-memory SQLite; for constrained code outputs, continue to_23_code_fixes/.

即：当"多解正确、文本比对会误杀"时用可执行评分；所有示例使用只读内存 SQLite，无需任何数据库服务；若你需要更受约束的代码输出评测，仓库提供了 代码修复任务集 作为后续延伸。

三个样例文件各自覆盖一类"强模型也不会饱和"的 SQL 能力（任务同时组合了时序规则、Join 和窗口函数，避免模型在简单的SELECT ... WHERE上直接拿满分）：

运行方式

.venvs/demo/bin/python cookbook/environments/_22_sql_generation/basic.py .venvs/demo/bin/python cookbook/environments/_22_sql_generation/joins.py .venvs/demo/bin/python cookbook/environments/_22_sql_generation/window_functions.py

• 需要环境变量OPENAI_API_KEY（示例使用OpenAIResponses(id="gpt-5.5", reasoning_effort="low", verbosity="low")）；
• 不需要任何数据库服务，SQLite 全部走进程内内存连接；
• 每个脚本调用run_rollouts(env, k=8, concurrency=4)：每个任务跑 8 次采样、最多 4 路并发，最后打印EnvironmentRunResult并渲染报告。

TEST_LOG.md 记录了 2026-07-20 针对gpt-5.5、agno 2.7.4 的实测：basic.py8 次 48 秒，inventory-state通过 7/8（0.875）；joins.py8 次 18 秒，business-minute-sla通过 7/8；window_functions.py16 次 85 秒，stateful-threshold-crossing通过 7/8，而更简单的final-state-audit饱和到 8/8。这个"7/8 的中段"正是任务设计者刻意追求的——下文会解释为什么。

样例一：basic.py — 递归状态机的库存回放

任务语义

inventory_events表按 SKU 独立回放事件（顺序为hanged_at, event_id），从 stock=0 开始：

• receive无条件加qty；
• reserve仅当当前库存 ≥ qty 才被接受（扣减 qty），被拒绝的 reserve 不改变库存；
• release仅当reserve_event_id指向同一 SKU、此前被接受、且尚未被有效 release 消耗的 reserve 才有效；有效 release 回补原始 reserve 数量并消耗该 reserve。重复、被拒、未知、指向未来或跨 SKU 的引用一律无效。

夹具数据里埋了典型陷阱：SKU A 的事件 4 和 5 两次指向 reserve 2（第二次无效），事件 15 指向不存在的 reserve 999，事件 6 的 reserve 10 因当时库存不足被拒。最终期望行是[["A", 4, 2, 1, 1, 2], ["B", 1, 2, 1, 1, 1]]（sku、final_stock、accepted_reserves、rejected_reserves、valid_releases、invalid_releases）。

这类任务的难点在于：release 的判定依赖"哪些 reserve 曾被接受"这一路径状态，模型必须在单条只读查询里用递归 CTE（配合 SQLite JSON 函数携带被接受 reserve 的 id 和数量）把状态一路推下去。

关键代码

Agent 定义（basic.py#L42-L49）：

class Query(BaseModel): sql: str = Field(..., description="One read-only SQLite query") agent = Agent( model=OpenAIResponses(id="gpt-5.5", reasoning_effort="low", verbosity="low"), instructions=( "Return one read-only SQLite query. Follow every temporal rule literally. " "Do not assume facts not present in the schema." ), output_schema=Query, )

output_schema=Query让run.content直接是Query实例而非字符串，评分器可以放心取run.content.sql。

Environment 装配（basic.py#L94-L108）把建表脚本与期望行一起放进Task.expected，由评分器在运行时消费：

env = Environment( name="inventory-state-sql", agent=agent, tasks=( Task( id="inventory-state", input=prompt, expected={ "setup": setup, # CREATE TABLE + INSERT 的 fixture 脚本 "rows": [["A", 4, 2, 1, 1, 2], ["B", 1, 2, 1, 1, 1]], }, ), ), scorer=CodeScorer(executes_to_expected_rows), )

评分器：三个样例共享的可执行判分函数

三个脚本共用同一套评分骨架executes_to_expected_rows(run, expected)（如 basic.py#L23-L39）：

def executes_to_expected_rows(run, expected): sql = run.content.sql.strip() if not sql.lower().startswith(("select", "with")): return Score(0.0, False, reason="query must start with SELECT or WITH") connection = sqlite3.connect(":memory:") try: connection.executescript(expected["setup"]) connection.execute("PRAGMA query_only = ON") actual = [list(row) for row in connection.execute(sql).fetchall()] except sqlite3.Error as exc: return Score(0.0, False, reason=f"SQLite rejected the query: {exc}") finally: connection.close() passed = actual == expected["rows"] return Score(1.0 if passed else 0.0, passed, reason=f"returned rows: {actual}")

四个值得注意的工程细节：

1. 只读双重保险：先检查语句必须以SELECT/WITH开头（WITH即 CTE，递归状态机几乎必然要用到），随后打开PRAGMA query_only = ON，即使模型绕过了前缀检查也无法写库；
2. 隔离性：每次评分都新开一个:memory:连接，夹具脚本executescript(expected["setup"])现搭现拆，attempt 之间互不污染，评分器因此可以在多路并发下安全运行；
3. 失败也计分：SQLite 拒绝执行的查询得Score(0.0, False, reason=...)而不是抛异常——从 TEST_LOG.md 可以看到，8 次采样里那次失败正是因为查询"递归形状对了但json_object()标签用了非文本导致 SQLite 拒绝"，这类失败被完整计入统计；
4. reason 携带实际返回行，方便逐 attempt 复盘失败原因。

源码级对照：Score 与 CodeScorer

评分器类型定义在 libs/agno/agno/scorer/base.py#L27-L40：

@dataclass class Score: """The result of scoring one run. `value` is always in [0, 1].""" value: float passed: bool reason: Optional[str] = None detail: Optional[Dict[str, Any]] = None

__post_init__会强制0.0 <= value <= 1.0，越界直接抛错——防止按 1-10 心算的评分器悄悄给所有 attempt 打绿灯。

CodeScorer 则是"把任意 callable 包装成 Scorer"的适配器：可接受函数签名(run, expected) -> bool | float | Score。bool直接映射为 1.0/0.0；float原样透传并以value >= pass_threshold判定通过；同步函数在ascore里经asyncio.to_thread执行，异步函数直接 await。这正是本任务集三个脚本能以三行代码接入判分的底层原因。

另一个容易忽略的点：CodeScorer.digest() 会对评分函数的去缩进源码 +pass_threshold做 sha256，作为Environment环境指纹（env fingerprint）的一部分。也就是说"换了判分逻辑或换通过阈值"会被识别为环境变化——对评测结果的可复现性比对很重要。

样例二：joins.py — 业务分钟 SLA 的多表 Join

joins.py 的任务是：对每个至少两张工单的组织，返回name, ticket_count, within_sla_count, within_sla_rate（保留三位小数），按 rate 降序、name 升序。规则密度明显更高：

• 工单的"响应"是最早的actor_type='agent'且created_at >= opened_at的响应；bot 响应和早于开单的响应都是干扰项（夹具里responses5 号响应就早于 201 号工单开单时间 5 分钟）；
• 无有效响应即 SLA 失败；
• 时长只算业务分钟：周一至周五、排除holidays表中的日期、09:00 含至 17:00 不含；
• 分钟数被精确定义为"落在业务时段内且opened_at <= m < response_at的整分钟时刻 m 的个数"，与组织级sla_minutes用<=比较，超过即超时。

夹具跨越了节假日（2025-07-07）和隔夜边界（103 号工单 16:30 开单、次日 12:00 才有人工响应），期望结果为[["Atlas", 3, 2, 0.667], ["Boreal", 2, 1, 0.5]]。Agent 的 instructions 相应调整为"时序规则用 CTE 显式表达时请优先用 CTE，并保持要求的输出顺序"（joins.py#L41-L48）。

TEST_LOG.md 里记录了这次失败 attempt 的病灶：返回了两个组织都是 100%，暴露的是业务分钟边界错误（半开区间取错），而不是 SQL 语法问题——这正是"对结果行评分"能暴露措辞比对永远发现不了的逻辑偏差。

样例三：window_functions.py — 递归轨迹 + 窗口比较

window_functions.py 在一个 Environment 里放了两个任务（window_functions.py#L111-L140）：

1. stateful-threshold-crossing：与 basic.py 相同的回放规则，但要求保留每个事件后的stock_after，派生delta_stock = stock_after - 上一事件 stock（首事件前为 0），返回所有"stock_after ≥ 5 且上一时刻 < 5"的向上穿越事件，输出sku, event_id, delta_stock, stock_after。期望 6 行：

   [["A", 1, 10, 10], ["A", 4, 7, 10], ["A", 9, 2, 6], ["B", 10, 5, 5], ["B", 13, 3, 5], ["B", 16, 5, 6]]

   它把"递归状态推导"和"对保留轨迹做窗口比较"（前一行 stock 与当前行 stock 的LAG式对比）拼在一条查询里，是三个样例中结构最复杂的。

2. final-state-audit：基本等同 basic.py 的最终状态审计（final_state_prompt），期望行[["A", 6, 2, 1, 1, 2], ["B", 6, 2, 1, 1, 1]]——注意夹具比 basic.py 多了事件 9（A 再 receive 2）和 16（B receive 5），所以最终库存从 4/1 变为 6/6。

TEST_LOG 显示：final-state-audit饱和 8/8，而stateful-threshold-crossing落在 7/8（0.875），失败那次依旧是"非文本 JSON 标签导致 SQLite 拒绝"。

运行引擎：Environment、Task 与 run_rollouts 的语义

三个脚本的收尾都是同一行：

results = run_rollouts(env, k=8, concurrency=4) print(results) results.print_report()

结合 libs/agno/agno/environments/runner.py#L1221-L1228 的签名可以读出各参数含义：

def run_rollouts( env: Environment, *, k: int = 8, # 每个任务的采样次数 tasks: Optional[Sequence[Task]] = None, # 可选地只跑 env.tasks 的子集 model: Optional[Model] = None, # 可选的模型覆盖 concurrency: int = 4, # attempt 并发度 ) -> EnvironmentRunResult:

run_rollouts是arun_rollouts的同步门面（asyncio.run包装），不能在已有事件循环内调用。它的实现还包含两个对本任务集很关键的行为：

• 每 attempt 深拷贝 Agent：Environment 的文档字符串 明确说明Environment是frozen=True，冻结的是"接线"而非状态——字段不可重绑，保证结果可证明来自这套任务集 + 评分器 + agent；而活体agent会在每次 attempt 时被深拷贝，避免跨 attempt 的状态泄漏（对 SQL 生成这种无工具、无会话状态的任务，这一点保证了 8 次采样是 8 次独立采样）；
• attempt 超时由Environment.timeout_seconds（默认 120 秒）控制，超时 attempt 记为"未评分"而非 0 分。

统计口径同样值得注意。TaskResult 上：

@property def pass_rate(self) -> Optional[float]: # Unscored attempts are excluded from statistics, never coerced to zero: # a timeout is not a wrong answer. if self.n_scored == 0: return None return self.n_passed / self.n_scored

未评分的 attempt 从统计中剔除而不是算 0——超时不是错误答案。TEST_LOG 中"8 attempts, all scored"说明三个脚本的采样全部进入了判分，0.875 是真实的通过率而不是被超时稀释的数。

TaskResult上还有一个in_learning_zone属性（runner.py#L79-L83）：只要"部分通过、部分失败"即为真。TEST_LOG 里反复出现的措辞——"created the useful middle band"、"escaped the wall of full bars"——说明这些样例在入库前做过任务难度校准：早期版本（固定 4 小时墙钟 SLA、简单时序台账等）会让模型 8/8 饱和，失去区分度；加入"按组织阈值 + 节假日 + 隔夜跨度 + bot/早于开单干扰项 + 精确分钟语义"后才落到 7/8 的"学习区"。这正是该目录文档说的"so a strong model does not saturate onSELECT ... WHERE"的具体含义。

最后，Environment.env_fingerprint()/policy_fingerprint()（environment.py#L150-L158）会在运行开始时把环境身份（任务集、评分器 digest、声明的工具 schema、prompt 字段）与模型策略身份（模型类、id、请求参数）分别盖章到结果上。对本任务集而言，这意味着：改动prompt、setup夹具、期望行或判分函数中的任何一个，都会产生不同的环境指纹——当你之后用EnvironmentRunResult的 diff 能力对比两批结果时，框架会拒绝跨环境指纹的比较（env_matches返回 False），防止拿"换题后的成绩"冒充"同题复测"。

小结：何时该用这套可执行 SQL 评分

把 README 的"When to use"展开成可操作的判断标准：

• 用：输出是"可以执行、且结果可确定性比对"的代码/查询，且自然语言到代码的映射存在大量等价解——SQL 生成（本文）、代码修复（_23_code_fixes）、以及仓库environments目录里的其他可执行任务；
• 评分器三要素照抄即可：前置合法性检查（语句形状 +PRAGMA query_only）→ 私有内存夹具执行 → 结果行相等比较，失败路径一律返回带reason的Score(0.0, False)而不是抛异常；
• 任务难度要主动校准：如果某任务连续 k 次全对（8/8），说明它对当前模型已饱和，应按 TEST_LOG 记录的方法叠加时序规则、干扰项与边界语义，把它拉回 0.5~1.0 之间的学习区，评测才有区分度。

运行前提再确认一遍：只需OPENAI_API_KEY与 agno 库本身，无需任何外部数据库服务；三个脚本均可独立运行、互相无依赖，按本文"运行方式"一节直接执行即可复现 TEST_LOG 中的报告。

【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno