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

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 querys wording.核心问题是对同一个自然语言需求存在许多语义等价的 SQL 写法CTE 拆法不同、子查询与 Join 互换、ORDER BY 等价形式……如果做精确文本比对会把大量有效解误判为错误。因此这个任务集采用可执行评分Agent 按output_schema输出一个类型化的 SQL 字符串Pydantic 模型包裹sql: str字段评分器在私有的:memory:SQLite 连接上执行建表脚本fixture然后执行模型生成的查询把返回的结果行与预计算的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上直接拿满分文件考察点basic.py回放库存事件reserve/release 的有效性依赖此前已被接受的状态递归 CTE 状态机joins.py关联组织、工单、响应历史、SLA 策略、节假日日历计算业务分钟响应时长window_functions.py保留递归推导的库存轨迹再用窗口函数比较找出所有向上穿越阈值的事件运行方式.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(idgpt-5.5, reasoning_effortlow, verbositylow)不需要任何数据库服务SQLite 全部走进程内内存连接每个脚本调用run_rollouts(env, k8, concurrency4)每个任务跑 8 次采样、最多 4 路并发最后打印EnvironmentRunResult并渲染报告。TEST_LOG.md 记录了 2026-07-20 针对gpt-5.5、agno 2.7.4 的实测basic.py8 次 48 秒inventory-state通过 7/80.875joins.py8 次 18 秒business-minute-sla通过 7/8window_functions.py16 次 85 秒stateful-threshold-crossing通过 7/8而更简单的final-state-audit饱和到 8/8。这个7/8 的中段正是任务设计者刻意追求的——下文会解释为什么。样例一basic.py — 递归状态机的库存回放任务语义inventory_events表按 SKU 独立回放事件顺序为hanged_at, event_id从 stock0 开始receive无条件加qtyreserve仅当当前库存 ≥ 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-L49class Query(BaseModel): sql: str Field(..., descriptionOne read-only SQLite query) agent Agent( modelOpenAIResponses(idgpt-5.5, reasoning_effortlow, verbositylow), instructions( Return one read-only SQLite query. Follow every temporal rule literally. Do not assume facts not present in the schema. ), output_schemaQuery, )output_schemaQuery让run.content直接是Query实例而非字符串评分器可以放心取run.content.sql。Environment 装配basic.py#L94-L108把建表脚本与期望行一起放进Task.expected由评分器在运行时消费env Environment( nameinventory-state-sql, agentagent, tasks( Task( idinventory-state, inputprompt, expected{ setup: setup, # CREATE TABLE INSERT 的 fixture 脚本 rows: [[A, 4, 2, 1, 1, 2], [B, 1, 2, 1, 1, 1]], }, ), ), scorerCodeScorer(executes_to_expected_rows), )评分器三个样例共享的可执行判分函数三个脚本共用同一套评分骨架executes_to_expected_rows(run, expected)如 basic.py#L23-L39def executes_to_expected_rows(run, expected): sql run.content.sql.strip() if not sql.lower().startswith((select, with)): return Score(0.0, False, reasonquery 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, reasonfSQLite rejected the query: {exc}) finally: connection.close() passed actual expected[rows] return Score(1.0 if passed else 0.0, passed, reasonfreturned rows: {actual})四个值得注意的工程细节只读双重保险先检查语句必须以SELECT/WITH开头WITH即 CTE递归状态机几乎必然要用到随后打开PRAGMA query_only ON即使模型绕过了前缀检查也无法写库隔离性每次评分都新开一个:memory:连接夹具脚本executescript(expected[setup])现搭现拆attempt 之间互不污染评分器因此可以在多路并发下安全运行失败也计分SQLite 拒绝执行的查询得Score(0.0, False, reason...)而不是抛异常——从 TEST_LOG.md 可以看到8 次采样里那次失败正是因为查询递归形状对了但json_object()标签用了非文本导致 SQLite 拒绝这类失败被完整计入统计reason 携带实际返回行方便逐 attempt 复盘失败原因。源码级对照Score 与 CodeScorer评分器类型定义在 libs/agno/agno/scorer/base.py#L27-L40dataclass 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.0float原样透传并以value pass_threshold判定通过同步函数在ascore里经asyncio.to_thread执行异步函数直接 await。这正是本任务集三个脚本能以三行代码接入判分的底层原因。另一个容易忽略的点CodeScorer.digest() 会对评分函数的去缩进源码 pass_threshold做 sha256作为Environment环境指纹env fingerprint的一部分。也就是说换了判分逻辑或换通过阈值会被识别为环境变化——对评测结果的可复现性比对很重要。样例二joins.py — 业务分钟 SLA 的多表 Joinjoins.py 的任务是对每个至少两张工单的组织返回name, ticket_count, within_sla_count, within_sla_rate保留三位小数按 rate 降序、name 升序。规则密度明显更高工单的响应是最早的actor_typeagent且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-L140stateful-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式对比拼在一条查询里是三个样例中结构最复杂的。final-state-audit基本等同 basic.py 的最终状态审计final_state_prompt期望行[[A, 6, 2, 1, 1, 2], [B, 6, 2, 1, 1, 1]]——注意夹具比 basic.py 多了事件 9A 再 receive 2和 16B receive 5所以最终库存从 4/1 变为 6/6。TEST_LOG 显示final-state-audit饱和 8/8而stateful-threshold-crossing落在 7/80.875失败那次依旧是非文本 JSON 标签导致 SQLite 拒绝。运行引擎Environment、Task 与 run_rollouts 的语义三个脚本的收尾都是同一行results run_rollouts(env, k8, concurrency4) 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 深拷贝 AgentEnvironment 的文档字符串 明确说明Environment是frozenTrue冻结的是接线而非状态——字段不可重绑保证结果可证明来自这套任务集 评分器 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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考