尧图网站建设 
Agno Environments 结构化抽取实战用类型化记录 来源优先级规则处理相互冲突的文本证据【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno本文基于 Agno 官方 Cookbook 的_24_structured_extraction环境演示讲解如何把多来源、互相冲突的散文证据抽取为一个完整、类型化、可逐字段核验的记录对象。读完后你将掌握如何为抽取任务定义 Pydantic 输出模式并内嵌来源优先级precedence策略、如何用Environment/Task/run_rollouts搭建可重复评测的抽取环境以及如何用CodeScorer对整条记录全字段精确匹配进行打分——这套方法适用于合同账户、物流发货单、订单快照等正确性即完整结构化对象的高可靠抽取场景。何时使用类型化抽取而非散文摘要cookbook/environments/_24_structured_extraction/README.md 给出的核心判断标准是当正确的定义是完整的结构化对象、而不是看起来合理的文字摘要时就应使用类型化抽取。原文档进一步给出两条实践建议值得完整保留把优先级规则precedence rules直接写进策略instructions中让模型在生成时就按规则裁决字段归属对每个字段都进行评分score every field——因为简单、无冲突的记录往往会让评分饱和saturate从而掩盖真正有价值的区分度区间。该目录的三个示例脚本正好对应这一思想的递进层次文件演示主题难度递增点basic.py从签署文档与不生效草稿中抽取生效的账户记录修订/撤回/条件生效/as-of 日期conflicting_fields.py按字段级来源优先级裁决发货单各字段并对被丢弃证据做审计校验和逐字段溯源 丢弃来源清单 审计校验和nested_records.py把修订、取消后的订单条目与发货项对账为排序好的嵌套对象嵌套模式 未发货余额对账 事件时间线本环境建立在前一环节 _23_code_fixes 的有界修复之上如需了解优先级更重的分类与升级escalation场景可继续阅读 _25_support_triage。示例一账户记录抽取basic.pybasic.py的任务是给定一个基线签署文档 BASE 若干修订amendment 若干噪音草稿、邮件、聊天记录抽取当前生效operative的账户记录。输出模式把约束编码进 Schemafrom typing import Literal from pydantic import BaseModel, Field class AccountRecord(BaseModel): account_id: str plan: Literal[starter, pro, enterprise] seats: int Field(ge1) renewal_date: str auto_renew: bool applied_source_ids: list[str] Field( descriptionOperative source ids in lexical order )注意applied_source_ids字段——它要求模型返回所有贡献了至少一个当前字段的来源 id精确大写、按字典序。这正是 README 所强调的对每个字段评分的落地方式只核对字段值时评测会饱和要求精确的来源溯源provenance才暴露出真正有区分度的中间分数带。TEST_LOG.md 中记录实测仅校验字段值的版本在前 3 行任务上全部 4/4 饱和加入精确来源溯源后later-narrow-amendment任务落到 2/4 的学习区评测信号变得可用。策略即指令优先级规则直接写进 instructionsAgent 的关键设计是把完整的裁决策略写进instructions而不是留给模型自行发挥agent Agent( modelOpenAIResponses(idgpt-5.5, reasoning_effortlow), output_schemaAccountRecord, instructions( Extract the operative account record. A later signed amendment overrides only the fields it changes. Signed documents outrank email and chat. Ignore drafts, proposals, quoted old text, and explicitly rejected changes. # ... 条件生效修订、撤回rescission、as-of 日期规则 ... Return the ids of every source that contributes at least one current field, using the exact uppercase ids in lexical order. ), )从这段策略可以提炼出五条核心裁决规则部分覆盖原则后签的修订只覆盖它实际变更的字段其余字段保留基线文档的值来源权威性分级签署文档 邮件/聊天噪音排除草稿draft、提案、被引用的旧文本、被明确拒绝REJECTED的变更一律不生效条件生效带前置条件的修订只有在签署证据确认条件成立后才生效例如安全审批邮件说大概批了但签署的安全决定写 NOT APPROVED则条件未满足as-of 时点给定提取时点时未来生效future-effective的变更不适用。任务集与打分environment定义了 5 个任务partial-amendment、rejected-revision、later-narrow-amendment、contingent-rescission、as-of-effective-date每个Task携带input冲突文本与expected黄金记录字典。以contingent-rescission为例BASE 为 Starter/18 席修订 A 改为 Pro/30 席/自动续费修订 B 有条件改席位与续费日条件未确认修订 C 只撤回 A 的 plan 变更——最终生效记录是starter/30 席/自动续费开来源为[A, BASE, C]。打分采用精确匹配def record_matches(run, expected) - bool: return ( isinstance(run.content, AccountRecord) and run.content.model_dump() expected ) environment Environment( nameaccount-record-extraction, agentagent, tasks(...), scorerCodeScorer(record_matches), ) if __name__ __main__: results run_rollouts(environment, k4, concurrency4)run_rollouts对每个任务做k4次重复 rollout、并发 4打印每任务的通过数n_passed/n_scored。这种多次重复 逐任务通过率的形式让评测结果呈现真实的可靠性分布而不是一次性运气。示例二字段级冲突裁决与审计校验和conflicting_fields.pyconflicting_fields.py把问题推进到逐字段来源裁决同一发货单的目的地、服务类型、申报价值、签收要求可能来自权威性不同的来源manifest 清单、printed label 标签、carrier acceptance scan 承运商签收扫描、signed shipping correction 签署的更正。逐字段优先级与溯源字段输出模式比示例一多了三块审计结构class ShipmentRecord(BaseModel): shipment_id: str destination_country: str service: Literal[ground, priority, express] declared_value_usd: int signature_required: bool source_by_field: dict[str, str] # 每个字段 - 生效来源 id discarded_source_ids: list[str] # 明确不生效的来源 discarded_source_checksum: int # 被丢弃证据的审计校验和策略要点摘自 instructions总优先级最新的签署更正 承运商签收扫描 打印标签 manifest。最新按生效时间戳最大判断而非在文本中出现的顺序只改它显式提供的字段来源未提及申报价值或签收条款时绝不改动这些字段不生效来源未签署的聊天、计划中的变更、作废VOIDED扫描、被引用的历史记录都不生效必须列入discarded_source_ids字典序注意不要把仍是有效兜底来源的低优先级来源也算进丢弃清单审计校验和这是一个刻意设计的防饱和机制——对每个被丢弃来源 id 的数字后缀n按字典序一基位置i计算h0 271828 sum(i*n)然后迭代 8 轮h_r (h_(r-1)^2 97*r 31) mod 10000019最终返回h_8。从 TEST_LOG 的校准记录看这个校验和是逃离 6/6 饱和墙的关键早期版本中字段值、来源 id、丢弃清单甚至 12 来源的时间戳审计全部饱和最终改为从来源 id 派生种子 8 轮模运算递推的丢弃证据校验和后timestamped-source-audit任务才落到 3/6 这类真实信号区间整个环境在 98 秒内跑完。任务示例以scan-and-correction为例清单 MCanada/ground/$900/需签收、标签 LUS/priority、签收扫描 SCanada/express、签署更正 C1只改申报价值为 $750、未签聊天 U1要求免签收但仍在待批。黄金答案为 destinationCanada来源 S、serviceexpress来源 S、价值750来源 C1、签收True来源 M丢弃来源仅[U1]校验和8635946。该环境共 4 个任务run_rollouts(environment, k6, concurrency6)以 6 次重复运行实测通过率为2/6、4/6、0/6、3/6——其中0/6行被保留作为无信号失败边界3/6等行则落入真实的部分通过率区间。示例三嵌套订单快照nested_records.pynested_records.py演示嵌套结构的抽取与对账订单items 多发货单shipments 未发货余额unshipped_items 生效事件时间线applied_event_ids。class LineItem(BaseModel): sku: str quantity: int Field(ge1) unit_price_cents: int Field(ge0) class Shipment(BaseModel): shipment_id: str items: list[ShipmentItem] class OrderSnapshot(BaseModel): order_id: str currency: str items: list[LineItem] shipments: list[Shipment] unshipped_items: list[ShipmentItem] applied_event_ids: list[str]策略要点包括修订替换命名行确认的修订只替换它点名的行未提及的行保留取消的行与作废voided的发货单整体排除发货单包含当前数量发货单若只写 SKU 不写数量则包含该行当前的完整数量稳定输出排序items 按 SKU 排序、shipments 按 shipment_id 排序、每个 shipment 内 items 按 SKU 排序——保证model_dump()可以逐字段精确比对价格规整小数价格统一转为整数美分cents如$19.95 → 1995未发货对账unshipped_items 每项当前订单数量减去非作废发货单的分配量按 SKU 排序零余额省略事件时间线applied_event_ids按时间顺序列出所有实际改变最终快照的生效修订/取消/作废/替换/重装repack事件省略初始声明与不生效的提案。TEST_LOG 记录把已不再影响规范化快照的事件从审计 id 中移除后才保留了两个真实的 2/4 学习区任务。最复杂的multi-stage-fulfillment任务5 个 SKU 的订单经两轮修订A1 改量/取消 B/加 FA2 恢复 B/改 D/取消 E、两个初始发货单、一次作废V1 作废 S2、一次替换R1 创建 S4与一次重装P1 移动一个 C最终要输出 4 个发货单、4 项未发货余额和事件时间线[A1, A2, V1, R1, P1]。底层实现Agno Environments 的组件三个脚本都只依赖 Agno 的四个组件全部可用当前仓库源码验证Agentoutput_schema指定 Pydantic 模式后run.content是反序列化好的模式实例CodeScorer的判定函数可以直接调用run.content.model_dump()与黄金字典做全量相等比较——这就是整条记录、逐字段打分的机制基础Task/Environment定义于 environment.pyTask类约 L34、Environment类约 L89Task携带id/input/expectedEnvironment聚合 agent、任务集与 scorerCodeScorer定义于 code.pyL12接收一个(run, expected) - bool的判定函数把正确性交给你写的确定性代码而不是模型自评run_rollouts定义于 runner.py约 L1221负责按k每任务重复次数与concurrency并发执行 rollout产出每个TaskResult的n_passed/n_scored支撑通过率区间式的评测结论而非一次性通过/失败。运行方式与适用前提运行命令README 原文python cookbook/environments/_24_structured_extraction/basic.py python cookbook/environments/_24_structured_extraction/conflicting_fields.py python cookbook/environments/_24_structured_extraction/nested_records.py前提条件需要设置OPENAI_API_KEY所有模型调用均使用OpenAIResponses且模型为gpt-5.5reasoning_effortlow依赖 Pydantic 模式定义BaseModel/Field/Literal无需额外外部服务。TEST_LOG.md 记录的实测环境为 Agno 2.7.4、gpt-5.5三个脚本状态均为 PASS并详细记录了每任务的通过率分布与两次校准过程修正签署更正的黄金答案、移除不再影响快照的审计事件 id。关键实践总结把优先级规则写进 instructions而不是事后用代码过滤来源权威性、只改显式字段、as-of 时点、条件生效等规则是策略本身应由模型在生成阶段执行用溯源字段防止评测饱和applied_source_ids、source_by_field、discarded_source_ids这类字段把答对值升级为答对值 答对为什么让简单任务不再无差别全对刻意加入有计算成本的正确性维度如审计校验和的 8 轮模运算递推能稳定地把评测推入 0–1 之间的真实信号区间实测 2/6、3/6、0/6 分布嵌套输出必须规定排序与数值规整SKU 排序、整数美分、零余额省略否则model_dump()精确比较会因顺序抖动而失真保留 0/k 与 k/k 边界任务TEST_LOG 显示团队刻意保留无信号失败边界和部分通过率区间的任务使评测集能持续暴露模型的薄弱能力带而不是追求全部饱和。【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Agno Environments 结构化抽取实战：用类型化记录 + 来源优先级规则处理相互冲突的文本证据
Agno Environments 结构化抽取实战用类型化记录 来源优先级规则处理相互冲突的文本证据【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno本文基于 Agno 官方 Cookbook 的_24_structured_extraction环境演示讲解如何把多来源、互相冲突的散文证据抽取为一个完整、类型化、可逐字段核验的记录对象。读完后你将掌握如何为抽取任务定义 Pydantic 输出模式并内嵌来源优先级precedence策略、如何用Environment/Task/run_rollouts搭建可重复评测的抽取环境以及如何用CodeScorer对整条记录全字段精确匹配进行打分——这套方法适用于合同账户、物流发货单、订单快照等正确性即完整结构化对象的高可靠抽取场景。何时使用类型化抽取而非散文摘要cookbook/environments/_24_structured_extraction/README.md 给出的核心判断标准是当正确的定义是完整的结构化对象、而不是看起来合理的文字摘要时就应使用类型化抽取。原文档进一步给出两条实践建议值得完整保留把优先级规则precedence rules直接写进策略instructions中让模型在生成时就按规则裁决字段归属对每个字段都进行评分score every field——因为简单、无冲突的记录往往会让评分饱和saturate从而掩盖真正有价值的区分度区间。该目录的三个示例脚本正好对应这一思想的递进层次文件演示主题难度递增点basic.py从签署文档与不生效草稿中抽取生效的账户记录修订/撤回/条件生效/as-of 日期conflicting_fields.py按字段级来源优先级裁决发货单各字段并对被丢弃证据做审计校验和逐字段溯源 丢弃来源清单 审计校验和nested_records.py把修订、取消后的订单条目与发货项对账为排序好的嵌套对象嵌套模式 未发货余额对账 事件时间线本环境建立在前一环节 _23_code_fixes 的有界修复之上如需了解优先级更重的分类与升级escalation场景可继续阅读 _25_support_triage。示例一账户记录抽取basic.pybasic.py的任务是给定一个基线签署文档 BASE 若干修订amendment 若干噪音草稿、邮件、聊天记录抽取当前生效operative的账户记录。输出模式把约束编码进 Schemafrom typing import Literal from pydantic import BaseModel, Field class AccountRecord(BaseModel): account_id: str plan: Literal[starter, pro, enterprise] seats: int Field(ge1) renewal_date: str auto_renew: bool applied_source_ids: list[str] Field( descriptionOperative source ids in lexical order )注意applied_source_ids字段——它要求模型返回所有贡献了至少一个当前字段的来源 id精确大写、按字典序。这正是 README 所强调的对每个字段评分的落地方式只核对字段值时评测会饱和要求精确的来源溯源provenance才暴露出真正有区分度的中间分数带。TEST_LOG.md 中记录实测仅校验字段值的版本在前 3 行任务上全部 4/4 饱和加入精确来源溯源后later-narrow-amendment任务落到 2/4 的学习区评测信号变得可用。策略即指令优先级规则直接写进 instructionsAgent 的关键设计是把完整的裁决策略写进instructions而不是留给模型自行发挥agent Agent( modelOpenAIResponses(idgpt-5.5, reasoning_effortlow), output_schemaAccountRecord, instructions( Extract the operative account record. A later signed amendment overrides only the fields it changes. Signed documents outrank email and chat. Ignore drafts, proposals, quoted old text, and explicitly rejected changes. # ... 条件生效修订、撤回rescission、as-of 日期规则 ... Return the ids of every source that contributes at least one current field, using the exact uppercase ids in lexical order. ), )从这段策略可以提炼出五条核心裁决规则部分覆盖原则后签的修订只覆盖它实际变更的字段其余字段保留基线文档的值来源权威性分级签署文档 邮件/聊天噪音排除草稿draft、提案、被引用的旧文本、被明确拒绝REJECTED的变更一律不生效条件生效带前置条件的修订只有在签署证据确认条件成立后才生效例如安全审批邮件说大概批了但签署的安全决定写 NOT APPROVED则条件未满足as-of 时点给定提取时点时未来生效future-effective的变更不适用。任务集与打分environment定义了 5 个任务partial-amendment、rejected-revision、later-narrow-amendment、contingent-rescission、as-of-effective-date每个Task携带input冲突文本与expected黄金记录字典。以contingent-rescission为例BASE 为 Starter/18 席修订 A 改为 Pro/30 席/自动续费修订 B 有条件改席位与续费日条件未确认修订 C 只撤回 A 的 plan 变更——最终生效记录是starter/30 席/自动续费开来源为[A, BASE, C]。打分采用精确匹配def record_matches(run, expected) - bool: return ( isinstance(run.content, AccountRecord) and run.content.model_dump() expected ) environment Environment( nameaccount-record-extraction, agentagent, tasks(...), scorerCodeScorer(record_matches), ) if __name__ __main__: results run_rollouts(environment, k4, concurrency4)run_rollouts对每个任务做k4次重复 rollout、并发 4打印每任务的通过数n_passed/n_scored。这种多次重复 逐任务通过率的形式让评测结果呈现真实的可靠性分布而不是一次性运气。示例二字段级冲突裁决与审计校验和conflicting_fields.pyconflicting_fields.py把问题推进到逐字段来源裁决同一发货单的目的地、服务类型、申报价值、签收要求可能来自权威性不同的来源manifest 清单、printed label 标签、carrier acceptance scan 承运商签收扫描、signed shipping correction 签署的更正。逐字段优先级与溯源字段输出模式比示例一多了三块审计结构class ShipmentRecord(BaseModel): shipment_id: str destination_country: str service: Literal[ground, priority, express] declared_value_usd: int signature_required: bool source_by_field: dict[str, str] # 每个字段 - 生效来源 id discarded_source_ids: list[str] # 明确不生效的来源 discarded_source_checksum: int # 被丢弃证据的审计校验和策略要点摘自 instructions总优先级最新的签署更正 承运商签收扫描 打印标签 manifest。最新按生效时间戳最大判断而非在文本中出现的顺序只改它显式提供的字段来源未提及申报价值或签收条款时绝不改动这些字段不生效来源未签署的聊天、计划中的变更、作废VOIDED扫描、被引用的历史记录都不生效必须列入discarded_source_ids字典序注意不要把仍是有效兜底来源的低优先级来源也算进丢弃清单审计校验和这是一个刻意设计的防饱和机制——对每个被丢弃来源 id 的数字后缀n按字典序一基位置i计算h0 271828 sum(i*n)然后迭代 8 轮h_r (h_(r-1)^2 97*r 31) mod 10000019最终返回h_8。从 TEST_LOG 的校准记录看这个校验和是逃离 6/6 饱和墙的关键早期版本中字段值、来源 id、丢弃清单甚至 12 来源的时间戳审计全部饱和最终改为从来源 id 派生种子 8 轮模运算递推的丢弃证据校验和后timestamped-source-audit任务才落到 3/6 这类真实信号区间整个环境在 98 秒内跑完。任务示例以scan-and-correction为例清单 MCanada/ground/$900/需签收、标签 LUS/priority、签收扫描 SCanada/express、签署更正 C1只改申报价值为 $750、未签聊天 U1要求免签收但仍在待批。黄金答案为 destinationCanada来源 S、serviceexpress来源 S、价值750来源 C1、签收True来源 M丢弃来源仅[U1]校验和8635946。该环境共 4 个任务run_rollouts(environment, k6, concurrency6)以 6 次重复运行实测通过率为2/6、4/6、0/6、3/6——其中0/6行被保留作为无信号失败边界3/6等行则落入真实的部分通过率区间。示例三嵌套订单快照nested_records.pynested_records.py演示嵌套结构的抽取与对账订单items 多发货单shipments 未发货余额unshipped_items 生效事件时间线applied_event_ids。class LineItem(BaseModel): sku: str quantity: int Field(ge1) unit_price_cents: int Field(ge0) class Shipment(BaseModel): shipment_id: str items: list[ShipmentItem] class OrderSnapshot(BaseModel): order_id: str currency: str items: list[LineItem] shipments: list[Shipment] unshipped_items: list[ShipmentItem] applied_event_ids: list[str]策略要点包括修订替换命名行确认的修订只替换它点名的行未提及的行保留取消的行与作废voided的发货单整体排除发货单包含当前数量发货单若只写 SKU 不写数量则包含该行当前的完整数量稳定输出排序items 按 SKU 排序、shipments 按 shipment_id 排序、每个 shipment 内 items 按 SKU 排序——保证model_dump()可以逐字段精确比对价格规整小数价格统一转为整数美分cents如$19.95 → 1995未发货对账unshipped_items 每项当前订单数量减去非作废发货单的分配量按 SKU 排序零余额省略事件时间线applied_event_ids按时间顺序列出所有实际改变最终快照的生效修订/取消/作废/替换/重装repack事件省略初始声明与不生效的提案。TEST_LOG 记录把已不再影响规范化快照的事件从审计 id 中移除后才保留了两个真实的 2/4 学习区任务。最复杂的multi-stage-fulfillment任务5 个 SKU 的订单经两轮修订A1 改量/取消 B/加 FA2 恢复 B/改 D/取消 E、两个初始发货单、一次作废V1 作废 S2、一次替换R1 创建 S4与一次重装P1 移动一个 C最终要输出 4 个发货单、4 项未发货余额和事件时间线[A1, A2, V1, R1, P1]。底层实现Agno Environments 的组件三个脚本都只依赖 Agno 的四个组件全部可用当前仓库源码验证Agentoutput_schema指定 Pydantic 模式后run.content是反序列化好的模式实例CodeScorer的判定函数可以直接调用run.content.model_dump()与黄金字典做全量相等比较——这就是整条记录、逐字段打分的机制基础Task/Environment定义于 environment.pyTask类约 L34、Environment类约 L89Task携带id/input/expectedEnvironment聚合 agent、任务集与 scorerCodeScorer定义于 code.pyL12接收一个(run, expected) - bool的判定函数把正确性交给你写的确定性代码而不是模型自评run_rollouts定义于 runner.py约 L1221负责按k每任务重复次数与concurrency并发执行 rollout产出每个TaskResult的n_passed/n_scored支撑通过率区间式的评测结论而非一次性通过/失败。运行方式与适用前提运行命令README 原文python cookbook/environments/_24_structured_extraction/basic.py python cookbook/environments/_24_structured_extraction/conflicting_fields.py python cookbook/environments/_24_structured_extraction/nested_records.py前提条件需要设置OPENAI_API_KEY所有模型调用均使用OpenAIResponses且模型为gpt-5.5reasoning_effortlow依赖 Pydantic 模式定义BaseModel/Field/Literal无需额外外部服务。TEST_LOG.md 记录的实测环境为 Agno 2.7.4、gpt-5.5三个脚本状态均为 PASS并详细记录了每任务的通过率分布与两次校准过程修正签署更正的黄金答案、移除不再影响快照的审计事件 id。关键实践总结把优先级规则写进 instructions而不是事后用代码过滤来源权威性、只改显式字段、as-of 时点、条件生效等规则是策略本身应由模型在生成阶段执行用溯源字段防止评测饱和applied_source_ids、source_by_field、discarded_source_ids这类字段把答对值升级为答对值 答对为什么让简单任务不再无差别全对刻意加入有计算成本的正确性维度如审计校验和的 8 轮模运算递推能稳定地把评测推入 0–1 之间的真实信号区间实测 2/6、3/6、0/6 分布嵌套输出必须规定排序与数值规整SKU 排序、整数美分、零余额省略否则model_dump()精确比较会因顺序抖动而失真保留 0/k 与 k/k 边界任务TEST_LOG 显示团队刻意保留无信号失败边界和部分通过率区间的任务使评测集能持续暴露模型的薄弱能力带而不是追求全部饱和。【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考