本笔记本构建了一个用于智能体（agent）的改进飞轮。我们从真实的运行轨迹（traces）开始，加入人工和模型反馈，将反馈转化为评估（evals），并利用产生的证据建议 Codex 执行下一次的框架（harness）变更。
你将
- 创建一个基于 OpenAI Agents SDK 的财务分析师智能体
- 在合成的公司数据上运行它并捕获轨迹
- 为这些运行添加示例性的人工反馈和 LLM 生成的反馈
- 将反馈转化为以后可以重复运行的 Promptfoo 评估
- 使用 HALO 对下一次框架变更进行排名，并编写 Codex 可用的移交方案
在本笔记本中，框架（harness）是指围绕模型的完整契约，包括指令、工具、路由、输出要求和验证检查。
这个飞轮保留了你从每次运行中学到的东西。轨迹展示了发生了什么，反馈解释了重要之处，评估使这些期望可复用，Codex 可以基于产生的变更集采取行动。
你将构建什么
最终，你将获得
- 一个基于 OpenAI Agents SDK 的财务分析师，它在五次追踪运行中审查了一家虚构公司的尽职调查材料
- 针对这些运行轨迹的人工和 LLM 生成的反馈
- 一套自动生成的 Promptfoo 评估套件
- 针对当前智能体行为的 Promptfoo 验证门禁
- 对轨迹、反馈和评估结果的 HALO 优化过程
- 面向开发者的移交方案，以便 Codex 实施建议的框架变更
该智能体支持一家虚构公司的收购尽职调查。它审查财务导出数据、客户数据、合同、安全说明、董事会材料和管理层叙述，然后回答尽职调查问题，并提供引用和可供审查的制品。
该循环编写了一个推动工作向前发展的单一文件：位于 ARTIFACT_DIR 下的生成的 codex_handoff.md 文件。它包含了完整的 HALO 诊断、排名后的建议、背后的证据以及 Codex 进行下一次框架更新所需的实施指南。
自动化的程度由开发者决定。你可以使用这个循环来提出经过审查的变更集，或者将其连接到自动打开、合并和部署合并请求的工作流程。一个常见的起点是经过审查的循环，即系统提出变更集，开发者在合并前批准差异。随着评估门禁的可信度提高，同样的移交方案可以支持更深度的自动化。无论哪种方式，核心工作流程都是一致的：轨迹加上人工和模型反馈，最终转化为具体的框架变更，而不是仅仅作为不相关的评论。
与那些仅停留在轨迹或评估阶段的示例相比，本笔记本将轨迹、评审判断、生成的评估、优化以及实施移交保持在一个可运行的改进循环中。
先决条件
在安装示例使用的 Python 依赖项后，从仓库根目录运行此笔记本
python -m venv .venv
source .venv/bin/activate
pip install openai openai-agents halo-enginePromptfoo 通过 npx 运行，因此你还需要在路径中配置 Node.js 和 npx。
在运行笔记本之前设置 API 密钥
export OPENAI_API_KEY=...该示例特意设定为仅限实时运行。轨迹生成、模型批评、评估生成、验证和优化步骤均使用最新的模型输出，以便笔记本演示实际的循环，而不是脚本化的预览。下一个单元格将模型选择集中展示，以便你可以根据需要通过替换更便宜的模型来权衡质量与成本。
按默认的五次轨迹计算，完整运行预计需要 20 分钟左右，不过模型延迟和网络状况会影响此时间。最长的部分通常是第 3 步（运行追踪的智能体调用）和第 7 步（HALO 分析完整循环）。反馈、评估生成和 Promptfoo 单元格也会进行实时调用，但通常较短。运行时间较长的单元格会在工作时打印进度或经过的时间。
%%capture
# Install or upgrade the Python dependencies used by this notebook.
%pip install --quiet --upgrade openai openai-agents halo-enginefrom __future__ import annotations
import asyncio
import hashlib
import json
import os
import re
import shutil
import subprocess
import sys
import tempfile
import time
import textwrap
import threading
from contextlib import contextmanager
from dataclasses import asdict, dataclass, field
from datetime import datetime, timezone
from importlib.metadata import version
from pathlib import Path
from typing import Any, Iterable, Iterator, Mapping
from IPython.display import Markdown, display
from openai import OpenAI
def find_project_root(start: Path | None = None) -> Path:
current = (start or Path.cwd()).resolve()
for candidate in [current, *current.parents]:
if (candidate / "registry.yaml").exists():
return candidate
return current
PROJECT_ROOT = find_project_root()
if not os.getenv("OPENAI_API_KEY"):
raise RuntimeError("Set OPENAI_API_KEY before running this live notebook.")
if shutil.which("npx") is None:
raise RuntimeError("Install Node.js with npx before running the Promptfoo eval gate.")
# Edit these in one place if you want to use lower-cost models for part of the loop.
AGENT_MODEL = os.getenv("OPENAI_AGENT_MODEL", "gpt-5.5")
ANALYSIS_MODEL = os.getenv("OPENAI_ANALYSIS_MODEL", "gpt-5.5")
EVAL_GENERATION_MODEL = os.getenv("OPENAI_EVAL_GENERATION_MODEL", ANALYSIS_MODEL)
JUDGE_MODEL = os.getenv("OPENAI_JUDGE_MODEL", ANALYSIS_MODEL)
HALO_MODEL = os.getenv("OPENAI_HALO_MODEL", ANALYSIS_MODEL)
PROMPTFOO_VERSION = os.getenv("PROMPTFOO_VERSION", "0.121.9")
client = OpenAI()
def format_duration(seconds: float) -> str:
minutes, remainder = divmod(int(round(seconds)), 60)
return f"{minutes}m {remainder:02d}s" if minutes else f"{remainder}s"
ARTIFACT_DIR = PROJECT_ROOT / "examples" / "agents_sdk" / "agent_improvement_loop_artifacts"
TRACE_DIR = ARTIFACT_DIR / "traces"
HALO_TRACE_PATH = ARTIFACT_DIR / "halo_traces" / "traces.jsonl"
if ARTIFACT_DIR.exists():
shutil.rmtree(ARTIFACT_DIR)
ARTIFACT_DIR.mkdir(exist_ok=True)
TRACE_DIR.mkdir(exist_ok=True)
HALO_TRACE_PATH.parent.mkdir(exist_ok=True)
print("Project root detected.")
print("Models:", {
"agent": AGENT_MODEL,
"analysis": ANALYSIS_MODEL,
"eval_generation": EVAL_GENERATION_MODEL,
"judge": JUDGE_MODEL,
"halo": HALO_MODEL,
"promptfoo": PROMPTFOO_VERSION,
})Project root detected.
Models: {'agent': 'gpt-5.5', 'analysis': 'gpt-5.5', 'eval_generation': 'gpt-5.5', 'judge': 'gpt-5.5', 'halo': 'gpt-5.5', 'promptfoo': '0.121.9'}第 1 步：创建合成的公司数据
本笔记本为一家可能在收购中被审查的公司创建了虚构的尽职调查材料。这些数据混合了结构化导出数据和叙述性 Markdown 文档，因此智能体必须决定哪些来源值得更多关注。
合成数据中的叙述性 Markdown 文件
| 文件 | 包含原因 |
|---|---|
overview.md | 管理层最高级别的公司总结 |
product_strategy.md | 路线图背景及未经核实的 NRR（净收入留存率）估算 |
go_to_market.md | 销售动作背景，应与管道数据核对 |
board_deck.md | 与结构化导出数据可能存在冲突的精炼管理层叙述 |
financials/revenue_recognition_notes.md | 针对启动阶段 ARR（年度经常性收入）处理的会计背景 |
legal/contracts_summary.md | 合同层面的风险背景 |
legal/open_issues.md | 应保持可见的未决法律事项 |
security/security_overview.md | 安全状况和认证措辞 |
sales/security_faq.md | 面向销售的安全用语，可能夸大了证据 |
hr/org_chart.md | 领导层和员工的运营背景 |
sales/pipeline_notes.md | 定性管道评论 |
notes/qa_log.md | 尽职调查问题及未决的后续事项 |
示例在运行时生成合成的公司数据，使其保持独立，同时仍为智能体提供结构化导出数据和叙述性文档的真实混合以供分析。
定义合成源文件
下一个折叠单元格包含用于构建虚构公司数据的源文档。
from textwrap import dedent
WORKSPACE_FILES = {
"overview.md": """
# FictionalCorp XYZ
FictionalCorp XYZ is a revenue intelligence software company with annual SaaS subscriptions, usage add-ons, and launch-stage commitments.
Management reports FY2025 ARR of $43.0M and year-over-year growth of 71%.
Management reports no legal-entity customer above 15% of booked ARR after excluding launch-stage usage add-ons.
Legal summary: Management states legal matters are ordinary course and no contract terms should affect valuation.
""",
"product_strategy.md": """
# Product Strategy
Core product lines:
- Forecast Assist
- Pipeline Quality Monitor
- Renewal Risk Workbench
Product roadmap priority is enterprise workflow depth. Management expects usage add-ons to increase expansion revenue.
Sales leadership references a 122% NRR estimate in planning materials, but finance has not published official NRR and the estimate excludes selected downsell and churn adjustments.
""",
"go_to_market.md": """
# Go To Market
FictionalCorp XYZ sells to CRO and RevOps buyers through a direct sales motion.
The current plan assumes larger enterprise ACVs and partner-sourced pipeline. Pipeline conversion evidence should be checked against `sales/pipeline.csv`.
""",
"board_deck.md": """
# Board Packet - December 2025
- FY2025 ending ARR: $43.0M
- ARR growth: 71%
- Gross margin: 69%
- Cash burn: $2.9M per month
- Runway: 11 months
Management narrative: the company is positioned for efficient enterprise expansion.
ARR note: the headline ARR view includes signed launch-stage commitments and a usage true-up view used for board planning.
Management narrative: customer concentration is manageable when measured by legal entity and booked ARR.
""",
"financials/revenue_recognition_notes.md": """
# Revenue Recognition Notes
Finance treats `financials/arr_bridge.csv` as the controlled FY2025 ARR bridge.
The board deck ARR includes $2.8M of signed launch-stage commitments that were not live by 2025-12-31 and $1.1M of usage true-ups that finance does not classify as recurring ARR.
RevOps also circulates a bookings-adjusted ARR view of $40.8M. That view is useful for pipeline planning but should not be silently reconciled with the controlled ARR bridge.
""",
"legal/contracts_summary.md": """
# Contracts Summary
Standard customer contracts are annual SaaS agreements with security and DPA exhibits. The largest five customers account for $25.1M of ARR.
Management summary: legal matters are ordinary course and no contract terms should affect valuation.
Clause inventory has not been fully reconciled with this summary. Two strategic customer agreements are flagged for non-standard terms in `legal/clause_inventory.csv`.
""",
"legal/open_issues.md": """
# Open Legal Issues
Former reseller DataHarbor filed a breach-of-contract claim seeking $3.2M plus accelerated commissions. Counsel estimates loss is possible but not probable. A clause review also identified two strategic customer MSAs with non-standard change-of-control notice rights and uncapped confidentiality indemnity language.
""",
"security/security_overview.md": """
# Security Overview
SOC 2 Type I is complete. SOC 2 Type II fieldwork is in progress, and the Type II report has not been issued.
Customer security reviews should verify the exact certification status before relying on SOC 2 claims.
""",
"sales/security_faq.md": """
# Sales Security FAQ
Field guidance says Aurora is "SOC 2 complete" for late-stage enterprise deals.
Security team note: this wording was intended to refer to Type I readiness, not an issued Type II report. Do not use this FAQ as certification evidence without checking `security/security_overview.md`.
""",
"hr/org_chart.md": """
# Org Chart
- CEO
- CFO
- VP Sales
- VP Product
- Head of Security
Hiring plan assumes 14 net new GTM hires in 2026.
""",
"sales/pipeline_notes.md": """
# Pipeline Notes
Commit-stage pipeline includes $1.6M of DataHarbor-sourced opportunities that may be affected by the reseller dispute.
Northstar expansion pipeline assumes completion of SOC 2 Type II before procurement review. Finance has not included this expansion in controlled FY2025 ARR.
""",
"notes/qa_log.md": """
# Diligence Q&A Log
- NRR was requested. RevOps provided a 122% management estimate, but finance has not validated official NRR and says the estimate excludes downsold Northstar entities and a churned reseller-sourced account.
- CAC payback was requested but not provided.
- Top-two customer ARR equals $12.4M, or 34% of FY2025 ARR based on `customers/top_customers.csv`.
- Northstar Holdings parent-account ARR equals $12.4M, or 34% of FY2025 ARR based on `customers/account_hierarchy.csv`.
- Board ARR should not be silently reconciled to finance ARR; use `financials/revenue_recognition_notes.md` for the difference.
""",
"financials/arr_bridge.csv": """
metric,value_m
opening_arr_2025_m,21.58
new_arr_m,8.1
expansion_arr_m,3.2
contraction_arr_m,1.1
churn_arr_m,2.7
ending_arr_2025_m,36.9
bookings_adjusted_arr_m,40.8
""",
"financials/monthly_kpis.csv": """
month,ending_arr_m,new_arr_m,expansion_arr_m,churn_arr_m,gross_margin
2025-01,21.58,0.55,0.35,0.18,0.69
2025-02,23.28,0.59,0.37,0.20,0.69
2025-03,24.98,0.63,0.39,0.21,0.69
2025-04,26.69,0.67,0.41,0.22,0.69
2025-05,28.39,0.71,0.43,0.24,0.69
2025-06,30.09,0.75,0.45,0.26,0.69
2025-07,31.79,0.79,0.47,0.27,0.69
2025-09,33.50,0.83,0.49,0.28,0.69
2025-10,35.20,0.87,0.51,0.30,0.69
2025-12,36.90,0.91,0.53,0.32,0.69
""",
"financials/p_and_l.csv": """
period,revenue_m,gross_margin,opex_m,cash_burn_m,runway_months
FY2025,30.26,0.69,47.71,2.9,11
""",
"financials/retention_extract.csv": """
metric,value,status,notes
net_revenue_retention,122%,management_estimate_unvalidated,Sales deck estimate; excludes downsold Northstar entities and one churned reseller-sourced account.
gross_revenue_retention,84%,finance_partial,Preliminary 2025 cohort; usage feeds incomplete for two enterprise customers.
logo_retention,91%,finance_partial,"Includes legal entities, not parent-account rollups."
cac_payback_months,,not_provided,Requested by diligence team; no source schedule in dataroom.
""",
"customers/top_customers.csv": """
customer,parent_account,arr_m,arr_share,segment,renewal_date,inclusion_basis
Northstar Bank,Northstar Holdings,7.8,0.2114,Enterprise,2026-02-15,controlled_arr_bridge
Northstar Capital Markets,Northstar Holdings,4.6,0.1247,Enterprise,2026-04-01,controlled_arr_bridge
Helio Retail,Helio Retail,6.9,0.1870,Enterprise,2026-05-15,controlled_arr_bridge
BluePeak Logistics,BluePeak Logistics,3.6,0.0976,Mid-market,2026-06-30,controlled_arr_bridge
Summit Foods,Summit Foods,2.2,0.0596,Mid-market,2026-02-28,controlled_arr_bridge
""",
"customers/account_hierarchy.csv": """
legal_entity,parent_account,parent_arr_m,note
Northstar Bank,Northstar Holdings,12.4,Same procurement parent as Northstar Capital Markets.
Northstar Capital Markets,Northstar Holdings,12.4,Managed by separate RevOps owner but same parent renewal committee.
Helio Retail,Helio Retail,6.9,Standalone parent account.
BluePeak Logistics,BluePeak Logistics,3.6,Standalone parent account; renewal issue open.
""",
"customers/renewal_calendar.csv": """
customer,renewal_date,renewal_risk,notes
Northstar Bank,2026-02-15,medium,Expansion depends on completed SOC 2 Type II.
Northstar Capital Markets,2026-04-01,medium,Same parent procurement committee as Northstar Bank.
Helio Retail,2026-05-15,medium,Adoption below plan; forecast latency escalation remains in monitoring.
BluePeak Logistics,2026-06-30,high,Open CRM sync errors and renewal risk.
""",
"customers/customer_health.csv": """
customer,health,primary_risk,signal_date,caveat
Northstar Bank,green,none flagged,2025-10-31,"Northstar health is recorded by legal entity, not parent account."
Northstar Capital Markets,yellow,monitor adoption,2025-10-31,"Northstar health is recorded by legal entity, not parent account."
Helio Retail,yellow,monitor adoption,2025-12-15,
BluePeak Logistics,red,renewal risk,2025-12-15,
Summit Foods,yellow,monitor adoption,2025-12-15,
""",
"legal/clause_inventory.csv": """
customer,issue,exposure,confidence
Northstar Bank,change_of_control_notice,customer may request transition plan within 10 days of a control transaction,medium
Helio Retail,uncapped_confidentiality_indemnity,uncapped liability for confidentiality breach; not reflected in management summary,high
BluePeak Logistics,service_credit_carveout,credits can exceed one month fees if CRM sync SLA missed for two consecutive months,medium
""",
"sales/pipeline.csv": """
stage,pipeline_m,historical_close_rate,quality_note
commit,6.1,0.39,Includes security-dependent Northstar expansion.
best_case,9.7,0.28,Includes DataHarbor-sourced opportunities under dispute.
early,18.2,0.08,High volume but low conversion quality.
""",
"support/escalations.csv": """
customer,severity,issue,status
Northstar Capital Markets,medium,Forecast latency,monitoring
BluePeak Logistics,high,CRM sync errors,open
Northstar Bank,medium,Security questionnaire blocked pending SOC 2 Type II report,open
""",
}实例化合成数据
将源文件写入磁盘，添加清单，并检查生成的数据集。
def write_workspace_file(path: Path, content: str) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(dedent(content).strip() + "\n", encoding="utf-8")
def generate_acquisition_diligence_workspace() -> Path:
"""Create the synthetic acquisition-diligence workspace directly from notebook data."""
dataroom = ARTIFACT_DIR / "synthetic_dataroom"
shutil.rmtree(dataroom, ignore_errors=True)
for relative_path, content in WORKSPACE_FILES.items():
write_workspace_file(dataroom / relative_path, content)
manifest = {
"company_name": "FictionalCorp XYZ",
"scenario": "adversarial_diligence",
"files": sorted(str(path.relative_to(dataroom)) for path in dataroom.rglob("*") if path.is_file()),
}
write_workspace_file(dataroom / "manifest.json", json.dumps(manifest, indent=2))
return dataroom
dataset = generate_acquisition_diligence_workspace()
files = sorted(str(path.relative_to(dataset)) for path in dataset.rglob("*") if path.is_file())
print(f"Dataset created: {len(files)} files")Dataset created: 24 files第 2 步：定义基于 Agents SDK 的分析师
示例智能体对一家作为可能收购目标的虚构 SaaS 公司执行尽职调查。案例材料包含结构化导出数据和管理层叙述。部分来源存在共识，部分存在冲突，有些重要结论仅得到部分支持。这为我们随着时间推移改进框架提供了现实理由。
智能体仅使用提供的公司数据为投资团队回答问题。当结构化财务证据与叙述总结发生冲突时，它应优先选择前者；在证据缺失时应保留不确定性；并留下可供其他评审员检查的制品。
OpenAI Agents SDK 提供了此工作流程所需的托管运行程序、沙盒执行、模型设置和追踪钩子。综上所述，提示词、工具、路由规则、输出要求和验证检查共同构成了当前的智能体框架（agent harness）。
智能体生成的制品
| 制品 | 智能体编写它的原因 |
|---|---|
summary_answer.md | 返回给用户的简洁答案 |
investment_memo.md | 为尽职调查阅读者准备的更完整的评审制品 |
risk_register.json | 包含证据的结构化风险，下游系统可对其进行审查 |
open_questions.md | 应保持可见的缺失证据或未解决的问题 |
citations.json | 从结论到源文件的机器可读链接 |
evidence_table.csv | 结论与支持来源的表格化审计追踪 |
这些制品通过在最终答案旁边保留支持证据、未决问题和必需文件，使工作保持可审查性。
需要留意的失败模式
本笔记本旨在揭示诸如以下之类的失败：
- 当结构化导出数据不一致时，将管理层叙述视为官方指标
- 报告未经财务核实的 NRR 估算
- 将母账户集中度归并为较弱的法律实体视角
- 当证据仅支持 Type I 时，声称“SOC 2 已完成”
- 在撰写精炼答案的同时，遗漏引用、风险文件或证据制品
定义框架模式
从模型设置和提升后的智能体配置的小型数据结构开始。这使得框架明确，以便以后的优化可以针对除了提示词措辞之外的内容。
@dataclass(frozen=True)
class ModelSettings:
agent_model: str
reasoning_effort: str
@dataclass(frozen=True)
class AgentConfig:
version: str
system_prompt: str
model_settings: ModelSettings
tool_policy: dict[str, Any]
eval_metadata: dict[str, Any]
path: Path = field(default_factory=lambda: Path("notebook_defined_agent_config"))
@property
def required_artifacts(self) -> list[str]:
return self.tool_policy["required_artifacts"]
def build_instructions(self) -> str:
return "\n\n".join([
self.system_prompt,
format_policy_section("Tool policy", self.tool_policy),
f"Runtime config:\n- Config version: `{self.version}`.\n- Treat this config as the promoted runtime contract.\n- Do not modify the runtime config during the run.",
]) + "\n"
def format_policy_section(title: str, policy: dict[str, Any]) -> str:
lines = [f"{title}:"]
for key, value in policy.items():
lines.extend(format_policy_value(key, value))
return "\n".join(lines)
def format_policy_value(key: str, value: Any, indent: int = 0) -> list[str]:
prefix = " " * indent
if isinstance(value, dict):
lines = [f"{prefix}- {key}:"]
for child_key, child_value in value.items():
lines.extend(format_policy_value(child_key, child_value, indent + 1))
return lines
if isinstance(value, list):
lines = [f"{prefix}- {key}:"]
for item in value:
if isinstance(item, dict):
lines.append(f"{prefix} -")
for child_key, child_value in item.items():
lines.extend(format_policy_value(child_key, child_value, indent + 2))
else:
lines.append(f"{prefix} - {item}")
return lines
return [f"{prefix}- {key}: {value}"]配置指令和策略
系统提示词陈述证据规则，工具策略定义智能体可读写的内容，评估元数据记录当前已提升的框架版本。
SYSTEM_PROMPT = """
You are a diligence analyst reviewing a synthetic company dataroom.
Evidence scope:
- Use only files under `data/`.
- Do not use outside knowledge or assumptions.
- Prefer structured CSV/JSON exports over narrative files when they conflict.
Runtime tools:
- The sandbox starts in the mounted workspace root. Use workspace-relative paths such as `data/...` and `outputs/...`; when running shell commands, omit `workdir` or use a relative path only. Never pass absolute temporary paths.
- `data/tools/check_evidence_coverage.py`: use this before finalizing answers with material claims. Create a JSON list of claims with `claim`, `claim_type`, and `citations`, then run `python data/tools/check_evidence_coverage.py --claims-json <path> --dataset-root data --output outputs/evidence_coverage.json`.
- `data/tools/validate_output_contract.py`: run this after writing the required artifacts and before final response with `python data/tools/validate_output_contract.py --outputs outputs --dataset-root data --output outputs/output_contract_validation.json`.
- If either tool reports unsupported claims, missing citations, missing files, malformed JSON, or empty artifacts, revise the answer/artifacts before finalizing. If the evidence is unavailable, say the claim is unknown or unsupported.
Citation rules:
- Every material claim must cite one or more source filenames.
- Cite filenames exactly as workspace-relative paths, for example `financials/arr_bridge.csv`.
- Do not cite files that do not support the claim.
Unknown-handling rules:
- If evidence is missing, state that the answer is unknown or unsupported.
- Never fabricate missing numbers.
- If evidence conflicts, state the conflict explicitly instead of reconciling silently.
Output rules:
- Write `outputs/summary_answer.md`.
- Write `outputs/investment_memo.md`.
- Write `outputs/risk_register.json`.
- Write `outputs/open_questions.md`.
- Write `outputs/citations.json`.
- Write `outputs/evidence_table.csv`.
""".strip()
MODEL_SETTINGS = {
"agent_model": AGENT_MODEL,
"reasoning_effort": "medium",
}
TOOL_POLICY = {
"allowed_data_root": "data",
"writable_output_root": "outputs",
"required_artifacts": [
"summary_answer.md",
"investment_memo.md",
"risk_register.json",
"open_questions.md",
"citations.json",
"evidence_table.csv",
],
"evidence_preference": [
"Prefer structured CSV or JSON exports over narrative summaries when sources conflict.",
"Treat board materials as useful narrative evidence, not the final system of record for metrics.",
"Surface unresolved conflicts instead of silently reconciling them.",
],
"runtime_tools": [
{
"path": "data/tools/check_evidence_coverage.py",
"purpose": "Audit drafted material claims against cited dataroom files before final answer.",
"recommended_command": "python data/tools/check_evidence_coverage.py --claims-json outputs/claim_audit_input.json --dataset-root data --output outputs/evidence_coverage.json",
},
{
"path": "data/tools/validate_output_contract.py",
"purpose": "Validate required output artifacts, JSON shape, and citation/source file references.",
"recommended_command": "python data/tools/validate_output_contract.py --outputs outputs --dataset-root data --output outputs/output_contract_validation.json",
},
],
"unknown_handling": [
"Say unknown or unsupported when a metric is absent.",
"Do not infer missing values from adjacent metrics.",
"Keep facts, inferences, and open questions separate.",
],
"mutation_policy": [
"Write only to the configured outputs directory.",
"Do not modify dataroom inputs.",
"Do not modify runtime agent configuration during a run.",
],
}
EVAL_METADATA = {
"version": "v001",
"status": "promoted",
"created_by": "manual_baseline",
"promotion_gate": "manual_review",
"description": "Baseline diligence analyst config with strict dataroom grounding, citation, unknown-handling, and artifact rules.",
}
agent_config = AgentConfig(
version=EVAL_METADATA["version"],
system_prompt=SYSTEM_PROMPT,
model_settings=ModelSettings(**MODEL_SETTINGS),
tool_policy=TOOL_POLICY,
eval_metadata=EVAL_METADATA,
)检查智能体配置
此紧凑视图显示了已提升的配置版本、所选模型、必需制品以及智能体可使用的运行时工具。
required_artifacts_md = "\n".join(
f"- `{artifact}`" for artifact in agent_config.required_artifacts
)
runtime_tools_md = "\n".join(
f"- `{tool['path']}` — {tool['purpose']}"
for tool in agent_config.tool_policy["runtime_tools"]
)
display(Markdown(f"""
### Agent config summary
- **Version:** `{agent_config.version}`
- **Agent model:** `{agent_config.model_settings.agent_model}`
- **Reasoning effort:** `{agent_config.model_settings.reasoning_effort}`
**Required artifacts**
{required_artifacts_md}
**Runtime tools**
{runtime_tools_md}
"""))智能体配置摘要
- 版本:
v001 - 智能体模型:
gpt-5.5 - 推理努力:
medium
必需制品
summary_answer.mdinvestment_memo.mdrisk_register.jsonopen_questions.mdcitations.jsonevidence_table.csv
运行时工具
data/tools/check_evidence_coverage.py— 在给出最终答案前，根据引用的数据室文件核对草拟的材料声明。data/tools/validate_output_contract.py— 验证必需的输出制品、JSON 形状以及引用/源文件参考。
添加验证工具
接下来的辅助工具在工作区内创建了两个本地工具：一个用于检查草拟的结论是否引用了真实的数据室文件，另一个验证所需的输出制品是否存在并具有预期的形状。代码默认隐藏以节省空间，但如果你想检查实现细节，可以展开它。
CHECK_EVIDENCE_COVERAGE = r'''#!/usr/bin/env python3
import argparse
import json
from pathlib import Path
def main() -> None:
parser = argparse.ArgumentParser(description="Audit whether drafted claims cite existing dataroom files.")
parser.add_argument("--claims-json", type=Path, required=True)
parser.add_argument("--dataset-root", type=Path, default=Path("data"))
parser.add_argument("--output", type=Path, default=Path("outputs/evidence_coverage.json"))
args = parser.parse_args()
claims = json.loads(args.claims_json.read_text(encoding="utf-8"))
if not isinstance(claims, list):
raise ValueError("--claims-json must contain a JSON list of claim objects")
result = check_evidence_coverage(claims, args.dataset_root)
args.output.parent.mkdir(parents=True, exist_ok=True)
args.output.write_text(json.dumps(result, indent=2) + "\n", encoding="utf-8")
print(json.dumps(result, indent=2))
def check_evidence_coverage(claims: list[dict], dataset_root: Path) -> dict:
supported = []
unsupported = []
missing_citations = []
for raw in claims:
claim = str(raw.get("claim") or "").strip()
claim_type = str(raw.get("claim_type") or "claim")
citations = [str(item).strip().removeprefix("data/") for item in raw.get("citations") or [] if str(item).strip()]
row = {"claim": claim, "claim_type": claim_type, "citations": citations}
if not citations:
missing_citations.append({**row, "issue": "No citation provided."})
continue
missing = [citation for citation in citations if not (dataset_root / citation).exists()]
if missing:
unsupported.append({**row, "issue": f"Missing cited file(s): {', '.join(missing)}"})
else:
supported.append(row)
return {
"supported_claims": supported,
"unsupported_claims": unsupported,
"missing_citations": missing_citations,
"recommended_caveats": [
"Add valid source filenames or mark unsupported claims as unknown before final answer."
],
"passed": not unsupported and not missing_citations,
}
if __name__ == "__main__":
main()
'''
VALIDATE_OUTPUT_CONTRACT = r'''#!/usr/bin/env python3
import argparse
import csv
import json
from pathlib import Path
REQUIRED_FILES = [
"summary_answer.md",
"investment_memo.md",
"risk_register.json",
"open_questions.md",
"citations.json",
"evidence_table.csv",
]
def main() -> None:
parser = argparse.ArgumentParser(description="Validate diligence output artifacts before final answer.")
parser.add_argument("--outputs", type=Path, default=Path("outputs"))
parser.add_argument("--dataset-root", type=Path, default=Path("data"))
parser.add_argument("--output", type=Path, default=Path("outputs/output_contract_validation.json"))
args = parser.parse_args()
result = validate_output_contract(args.outputs, args.dataset_root)
args.output.parent.mkdir(parents=True, exist_ok=True)
args.output.write_text(json.dumps(result, indent=2) + "\n", encoding="utf-8")
print(json.dumps(result, indent=2))
def validate_output_contract(outputs: Path, dataset_root: Path) -> dict:
issues = []
for filename in REQUIRED_FILES:
path = outputs / filename
if not path.exists():
issues.append({"file": filename, "issue": "missing required artifact"})
elif path.stat().st_size == 0:
issues.append({"file": filename, "issue": "empty required artifact"})
risks = _read_json(outputs / "risk_register.json", default=[])
citations = _read_json(outputs / "citations.json", default=[])
if not isinstance(risks, list):
issues.append({"file": "risk_register.json", "issue": "must be a JSON list"})
risks = []
if not isinstance(citations, list):
issues.append({"file": "citations.json", "issue": "must be a JSON list"})
citations = []
for index, risk in enumerate(risks):
evidence = risk.get("evidence") if isinstance(risk, dict) else None
if not evidence:
issues.append({"file": "risk_register.json", "risk_index": index, "issue": "risk lacks evidence"})
continue
missing = [str(item).removeprefix("data/") for item in evidence if not (dataset_root / str(item).removeprefix("data/")).exists()]
if missing:
issues.append({"file": "risk_register.json", "risk_index": index, "issue": f"missing evidence file(s): {', '.join(missing)}"})
for index, citation in enumerate(citations):
sources = citation.get("sources") if isinstance(citation, dict) else None
if not sources:
issues.append({"file": "citations.json", "citation_index": index, "issue": "citation lacks sources"})
continue
missing = [str(item).removeprefix("data/") for item in sources if not (dataset_root / str(item).removeprefix("data/")).exists()]
if missing:
issues.append({"file": "citations.json", "citation_index": index, "issue": f"missing source file(s): {', '.join(missing)}"})
try:
with (outputs / "evidence_table.csv").open(newline="", encoding="utf-8") as handle:
rows = list(csv.DictReader(handle))
if rows and not {"claim_id", "claim", "sources"}.issubset(rows[0].keys()):
issues.append({"file": "evidence_table.csv", "issue": "must include claim_id, claim, and sources columns"})
except FileNotFoundError:
pass
return {"passed": not issues, "issues": issues, "required_files": REQUIRED_FILES}
def _read_json(path: Path, default):
if not path.exists():
return default
try:
return json.loads(path.read_text(encoding="utf-8"))
except json.JSONDecodeError as exc:
return {"error": str(exc)}
if __name__ == "__main__":
main()
'''
def write_runtime_tools(dataset_dir: Path) -> list[str]:
tools_dir = dataset_dir / "tools"
tools_dir.mkdir(parents=True, exist_ok=True)
files = {
"check_evidence_coverage.py": CHECK_EVIDENCE_COVERAGE,
"validate_output_contract.py": VALIDATE_OUTPUT_CONTRACT,
}
written: list[str] = []
for filename, content in files.items():
path = tools_dir / filename
path.write_text(content, encoding="utf-8")
path.chmod(0o755)
written.append(str(path.relative_to(dataset_dir)))
return written构建每个用户轮次
提示词构建器仅在需要时才添加特定任务的指南，例如备忘录格式、单独的风险类别，或针对不支持的 NRR 结论的严格处理。
def build_user_prompt(question: str, agent_config: Any | None = None) -> str:
config_line = ""
if agent_config is not None:
config_line = f"\nActive agent config: `{agent_config.version}` from `{agent_config.path}`.\n"
memo_instruction = ""
if _asks_for_memo(question):
memo_instruction = (
"\nThe user asked for a memo-style deliverable. Return the memo content inline in "
"your final answer and also write the required output artifacts. Do not answer only "
"with a status update or artifact path list.\n"
)
risk_category_instruction = ""
if _asks_for_top_risk_categories(question):
risk_category_instruction = (
"\nStructure the final answer with separate sections for Financial, Legal, and "
"Customer concentration risks. Do not collapse customer concentration into the "
"financial category.\n"
)
unsupported_metric_instruction = ""
if _asks_for_net_revenue_retention(question):
unsupported_metric_instruction = (
"\nFor net revenue retention, report the metric only if the dataroom directly "
"provides NRR/net revenue retention. Do not derive or estimate an NRR percentage "
"from ARR bridge components unless the user explicitly asks for an estimate. If "
"the metric is absent, say it is unknown or unsupported, cite the searched "
"source files, and separate missing evidence from any directional inference.\n"
)
return f"""
Answer this diligence question using only the mounted dataroom:
{question}
{config_line}
{memo_instruction}
{risk_category_instruction}
{unsupported_metric_instruction}
Also write the required output artifacts. Keep the answer concise, grounded, and citation-heavy.
Use workspace-relative paths for shell commands and omit `workdir`; do not pass absolute temporary paths.
"""
def _asks_for_memo(question: str) -> bool:
lower = question.lower()
return "memo" in lower or "ic-style" in lower or "investment committee" in lower
def _asks_for_top_risk_categories(question: str) -> bool:
lower = question.lower()
return all(term in lower for term in ("financial", "legal", "customer")) and "risk" in lower
def _asks_for_net_revenue_retention(question: str) -> bool:
lower = question.lower()
return "net revenue retention" in lower or "nrr" in lower导出轨迹以供后续优化
本地导出器将 Agents SDK 事件转换为 HALO 稍后可读取的 OpenTelemetry 风格的 JSONL。由于实现较为繁重，代码默认保持折叠状态。
配置轨迹导出器
设置接收 Agents SDK 跨度（spans）并将每个跨度写入一行 JSONL 的导出器对象。
EXPORT_SCHEMA_VERSION = 1
OBSERVATION_KIND_BY_TYPE = {
"agent": "AGENT",
"generation": "LLM",
"response": "LLM",
"function": "TOOL",
"mcp_tools": "TOOL",
"handoff": "CHAIN",
"guardrail": "GUARDRAIL",
"custom": "SPAN",
"task": "SPAN",
"turn": "SPAN",
"transcription": "SPAN",
"speech": "SPAN",
"speech_group": "SPAN",
}
@dataclass(frozen=True)
class HaloExportContext:
project_id: str
service_name: str
service_version: str | None = None
deployment_environment: str | None = None
extra_resource_attributes: Mapping[str, Any] | None = None
def setup_halo_tracing(
path: str | Path,
*,
project_id: str = "synthetic-dataroom-agent",
service_name: str = "financial-diligence-analyst",
service_version: str | None = None,
deployment_environment: str | None = None,
extra_resource_attributes: Mapping[str, Any] | None = None,
):
from agents import set_trace_processors
trace_path = Path(path)
trace_path.parent.mkdir(parents=True, exist_ok=True)
processor = HaloJsonlTraceProcessor(
trace_path,
ctx=HaloExportContext(
project_id=project_id,
service_name=service_name,
service_version=service_version,
deployment_environment=deployment_environment,
extra_resource_attributes=extra_resource_attributes,
),
)
# Use only the local exporter for this cookbook workflow.
# Hosted trace ingestion may be unavailable in some environments (for example ZDR orgs).
set_trace_processors([processor])
return processor
class HaloJsonlTraceProcessor:
def __init__(self, path: Path, *, ctx: HaloExportContext):
self._path = path
self._ctx = ctx
self._lock = threading.Lock()
self._handle = path.open("a", encoding="utf-8")
self._trace_meta: dict[str, tuple[str | None, str | None, dict[str, Any]]] = {}
def on_trace_start(self, trace) -> None: # noqa: ANN001
data = trace.export() or {}
trace_id = _strip_prefix(data.get("id"), "trace_") or ""
metadata = data.get("metadata") if isinstance(data.get("metadata"), dict) else {}
self._trace_meta[trace_id] = (
data.get("workflow_name"),
data.get("group_id"),
metadata,
)
def on_trace_end(self, trace) -> None: # noqa: ANN001
data = trace.export() or {}
trace_id = _strip_prefix(data.get("id"), "trace_") or ""
self._trace_meta.pop(trace_id, None)
def on_span_start(self, span) -> None: # noqa: ANN001
return None
def on_span_end(self, span) -> None: # noqa: ANN001
exported = span.export() or {}
trace_id = _strip_prefix(exported.get("trace_id"), "trace_") or ""
workflow_name, group_id, trace_metadata = self._trace_meta.get(trace_id, (None, None, {}))
line = span_to_halo_jsonl_line(
span,
ctx=self._ctx,
workflow_name=workflow_name,
group_id=group_id,
trace_metadata=trace_metadata,
)
encoded = json.dumps(line, separators=(",", ":"), ensure_ascii=False, default=str)
with self._lock:
self._handle.write(encoded)
self._handle.write("\n")
def shutdown(self) -> None:
with self._lock:
try:
self._handle.flush()
self._handle.close()
except Exception:
pass
def force_flush(self) -> None:
with self._lock:
self._handle.flush()将 SDK 跨度映射为 HALO 可读字段
这些辅助函数将每种 SDK 跨度类型转换为 HALO 稍后将检查的属性。
def span_to_halo_jsonl_line(
span,
*,
ctx: HaloExportContext,
workflow_name: str | None = None,
group_id: str | None = None,
trace_metadata: Mapping[str, Any] | None = None,
) -> dict[str, Any]:
raw = span.export() or {}
span_data = raw.get("span_data") or {}
span_type = str(span_data.get("type") or "custom")
error = raw.get("error")
resource_attributes: dict[str, Any] = {"service.name": ctx.service_name}
if ctx.service_version:
resource_attributes["service.version"] = ctx.service_version
if ctx.deployment_environment:
resource_attributes["deployment.environment"] = ctx.deployment_environment
if ctx.extra_resource_attributes:
resource_attributes.update(ctx.extra_resource_attributes)
attributes, projection = _attributes_for_span_type(span_type, span_data)
if workflow_name:
attributes["agent.workflow.name"] = workflow_name
if group_id:
attributes["agent.workflow.group_id"] = group_id
for key, value in (trace_metadata or {}).items():
if _json_safe(value):
attributes[f"agent.trace_metadata.{key}"] = value
else:
attributes[f"agent.trace_metadata.{key}"] = _json(value)
attributes.update(
{
"inference.export.schema_version": EXPORT_SCHEMA_VERSION,
"inference.project_id": ctx.project_id,
"inference.observation_kind": OBSERVATION_KIND_BY_TYPE.get(span_type, "SPAN"),
"inference.llm.provider": projection.get("llm_provider"),
"inference.llm.model_name": projection.get("llm_model_name"),
"inference.llm.input_tokens": projection.get("input_tokens"),
"inference.llm.output_tokens": projection.get("output_tokens"),
"inference.llm.cost.total": projection.get("cost_total"),
"inference.user_id": projection.get("user_id"),
"inference.session_id": group_id,
"inference.agent_name": projection.get("agent_name") or "",
}
)
return {
"trace_id": _strip_prefix(raw.get("trace_id"), "trace_") or "",
"span_id": _strip_prefix(raw.get("id"), "span_") or "",
"parent_span_id": _strip_prefix(raw.get("parent_id"), "span_") or "",
"trace_state": "",
"name": _span_name(span_type, span_data),
"kind": _span_kind(span_type),
"start_time": _to_otlp_timestamp(raw.get("started_at")),
"end_time": _to_otlp_timestamp(raw.get("ended_at")),
"status": {
"code": "STATUS_CODE_ERROR" if error else "STATUS_CODE_OK",
"message": str((error or {}).get("message") or ""),
},
"resource": {"attributes": resource_attributes},
"scope": {"name": "openai-agents-sdk", "version": _sdk_version()},
"attributes": {key: value for key, value in attributes.items() if value is not None},
}
def _attributes_for_span_type(
span_type: str,
data: Mapping[str, Any],
) -> tuple[dict[str, Any], dict[str, Any]]:
if span_type == "agent":
return _agent_attrs(data)
if span_type == "generation":
return _generation_attrs(data)
if span_type == "response":
return _response_attrs(data)
if span_type == "function":
return _function_attrs(data)
if span_type == "mcp_tools":
return _mcp_tools_attrs(data)
if span_type == "handoff":
return _handoff_attrs(data)
if span_type == "guardrail":
return _guardrail_attrs(data)
return _custom_attrs(span_type, data)
def _agent_attrs(data: Mapping[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]:
name = data.get("name") or ""
return _drop_none(
{
"openinference.span.kind": "AGENT",
"agent.name": name,
"agent.handoffs": _json(data.get("handoffs")),
"agent.tools": _json(data.get("tools")),
"agent.output_type": data.get("output_type"),
}
), {"agent_name": name}
def _generation_attrs(data: Mapping[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]:
usage = data.get("usage") or {}
input_messages = data.get("input") or []
output_messages = data.get("output") or []
attrs: dict[str, Any] = {
"openinference.span.kind": "LLM",
"llm.provider": "openai",
"llm.model_name": data.get("model"),
"llm.invocation_parameters": _json(data.get("model_config")),
"llm.input_messages": _json(list(input_messages)),
"llm.output_messages": _json(list(output_messages)),
"llm.token_count.prompt": _int(usage.get("input_tokens") or usage.get("prompt_tokens")),
"llm.token_count.completion": _int(
usage.get("output_tokens") or usage.get("completion_tokens")
),
"llm.token_count.total": _int(usage.get("total_tokens")),
}
attrs.update(_expand_messages("llm.input_messages", input_messages))
attrs.update(_expand_messages("llm.output_messages", output_messages))
return _drop_none(attrs), {
"llm_provider": "openai",
"llm_model_name": data.get("model"),
"input_tokens": _int(usage.get("input_tokens") or usage.get("prompt_tokens")),
"output_tokens": _int(usage.get("output_tokens") or usage.get("completion_tokens")),
}
def _response_attrs(data: Mapping[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]:
usage = data.get("usage") or {}
return _drop_none(
{
"openinference.span.kind": "LLM",
"llm.provider": "openai",
"llm.response.id": data.get("response_id"),
"llm.token_count.prompt": _int(usage.get("input_tokens") or usage.get("prompt_tokens")),
"llm.token_count.completion": _int(
usage.get("output_tokens") or usage.get("completion_tokens")
),
"llm.token_count.total": _int(usage.get("total_tokens")),
}
), {
"llm_provider": "openai",
"input_tokens": _int(usage.get("input_tokens") or usage.get("prompt_tokens")),
"output_tokens": _int(usage.get("output_tokens") or usage.get("completion_tokens")),
}
def _function_attrs(data: Mapping[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]:
return _drop_none(
{
"openinference.span.kind": "TOOL",
"tool.name": data.get("name"),
"input.value": data.get("input"),
"output.value": data.get("output"),
"mcp.data": _json(data.get("mcp_data")),
}
), {}
def _mcp_tools_attrs(data: Mapping[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]:
return _drop_none(
{
"openinference.span.kind": "TOOL",
"mcp.server": data.get("server"),
"mcp.tools.listed": _json(data.get("result")),
}
), {}
def _handoff_attrs(data: Mapping[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]:
return _drop_none(
{
"openinference.span.kind": "CHAIN",
"agent.handoff.from": data.get("from_agent"),
"agent.handoff.to": data.get("to_agent"),
}
), {"agent_name": data.get("to_agent")}
def _guardrail_attrs(data: Mapping[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]:
return _drop_none(
{
"openinference.span.kind": "GUARDRAIL",
"guardrail.name": data.get("name"),
"guardrail.triggered": bool(data.get("triggered")),
}
), {}
def _custom_attrs(span_type: str, data: Mapping[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]:
attrs: dict[str, Any] = {
"openinference.span.kind": "CHAIN",
"sdk.span.type": span_type,
}
if data.get("name"):
attrs["sdk.span.name"] = data.get("name")
payload = data.get("data") or {}
if isinstance(payload, Mapping):
for key, value in payload.items():
attrs[f"sdk.data.{key}"] = value if _json_safe(value) else _json(value)
if "usage" in data:
attrs["llm.token_count.total"] = _int((data.get("usage") or {}).get("total_tokens"))
return _drop_none(attrs), {}标准化辅助值
最后的辅助函数确保 ID、时间戳和序列化值在导出的跨度中保持一致。
def _strip_prefix(value: Any, prefix: str) -> str | None:
if not value:
return None
text = str(value)
return text[len(prefix) :] if text.startswith(prefix) else text
def _to_otlp_timestamp(value: str | None) -> str:
if not value:
return ""
parsed = datetime.fromisoformat(value)
if parsed.tzinfo is None:
parsed = parsed.replace(tzinfo=timezone.utc)
parsed = parsed.astimezone(timezone.utc)
return parsed.strftime("%Y-%m-%dT%H:%M:%S.") + f"{parsed.microsecond:06d}000Z"
def _span_kind(span_type: str) -> str:
return "SPAN_KIND_CLIENT" if span_type in {"generation", "response"} else "SPAN_KIND_INTERNAL"
def _span_name(span_type: str, data: Mapping[str, Any]) -> str:
if data.get("name"):
return f"{span_type}.{data['name']}"
if data.get("model"):
return f"{span_type}.{data['model']}"
return span_type
def _expand_messages(prefix: str, messages: Iterable[Mapping[str, Any]]) -> dict[str, Any]:
attrs: dict[str, Any] = {}
for index, message in enumerate(messages or []):
if not isinstance(message, Mapping):
continue
role = message.get("role")
content = message.get("content")
if role is not None:
attrs[f"{prefix}.{index}.message.role"] = role
if isinstance(content, str):
attrs[f"{prefix}.{index}.message.content"] = content
elif content is not None:
attrs[f"{prefix}.{index}.message.content"] = _json(content)
for tool_index, tool_call in enumerate(message.get("tool_calls") or []):
function = (tool_call or {}).get("function") or {}
attrs[f"{prefix}.{index}.message.tool_calls.{tool_index}.tool_call.id"] = (
tool_call or {}
).get("id")
attrs[
f"{prefix}.{index}.message.tool_calls.{tool_index}.tool_call.function.name"
] = function.get("name")
attrs[
f"{prefix}.{index}.message.tool_calls.{tool_index}.tool_call.function.arguments"
] = function.get("arguments")
if message.get("tool_call_id"):
attrs[f"{prefix}.{index}.message.tool_call_id"] = message["tool_call_id"]
if message.get("name"):
attrs[f"{prefix}.{index}.message.name"] = message["name"]
return {key: value for key, value in attrs.items() if value is not None}
def _json(value: Any) -> str | None:
if value is None:
return None
return json.dumps(value, default=str, separators=(",", ":"))
def _json_safe(value: Any) -> bool:
return isinstance(value, (str, int, float, bool)) or value is None
def _int(value: Any) -> int | None:
if value is None:
return None
try:
return int(value)
except (TypeError, ValueError):
return None
def _drop_none(values: Mapping[str, Any]) -> dict[str, Any]:
return {key: value for key, value in values.items() if value is not None}
def _sdk_version() -> str:
try:
return version("openai-agents")
except Exception:
return "unknown"运行 SDK 智能体
run_sdk_agent() 直接调用 Agents SDK 运行程序，同时处理每次追踪运行前后的重复设置：挂载数据、附加追踪、执行智能体并收集输出制品。
async def run_sdk_agent(
dataset_dir: Path,
output_dir: Path,
question: str,
model: str,
agent_config: AgentConfig,
trace_id: str | None = None,
trace_metadata: dict[str, Any] | None = None,
halo_trace_path: str | Path | None = None,
halo_project_id: str = "financial_diligence_analyst_optimization_context",
) -> str:
from agents import ModelSettings as SDKModelSettings
from agents import Runner, custom_span, flush_traces, trace
from agents.run import RunConfig
from agents.sandbox import Manifest, SandboxAgent, SandboxRunConfig
from agents.sandbox.entries import Dir, LocalDir
from agents.sandbox.sandboxes.unix_local import UnixLocalSandboxClient
from openai.types.shared import Reasoning
output_dir.mkdir(parents=True, exist_ok=True)
with staged_dataset_mount(dataset_dir) as staged_dataset_dir:
write_runtime_manifest(staged_dataset_dir)
reasoning = Reasoning(effort=agent_config.model_settings.reasoning_effort)
agent = SandboxAgent(
name="Synthetic dataroom diligence analyst",
model=model,
model_settings=SDKModelSettings(reasoning=reasoning),
instructions=agent_config.build_instructions(),
default_manifest=Manifest(
entries={
"data": LocalDir(src=staged_dataset_dir),
"outputs": Dir(),
}
),
)
client = UnixLocalSandboxClient()
session = None
halo_processor = None
if halo_trace_path is not None:
halo_processor = setup_halo_tracing(
halo_trace_path,
project_id=halo_project_id,
service_version=agent_config.version,
deployment_environment="notebook" if trace_metadata else None,
extra_resource_attributes={
"agent.config.version": agent_config.version,
"agent.config.path": str(agent_config.path),
},
)
trace_context = (
trace(
workflow_name="Synthetic dataroom diligence",
trace_id=trace_id,
metadata=trace_metadata,
)
if trace_id
else None
)
if trace_context is not None:
trace_context.__enter__()
try:
with custom_span(
"sandbox_workspace",
{
"tool.name": "sandbox_workspace",
"tool.input": {
"mounted": "data",
"writable": "outputs",
"dataset_dir": str(dataset_dir),
"staged_dataset_dir": str(staged_dataset_dir),
"agent_config": str(agent_config.path),
"agent_config_version": agent_config.version,
},
},
disabled=trace_context is None,
):
with custom_span(
"agent_config",
{
"tool.name": "agent_config",
"tool.input": {
"version": agent_config.version,
"required_artifacts": agent_config.required_artifacts,
},
},
disabled=trace_context is None,
):
pass
session = await client.create(manifest=agent.default_manifest)
async with session:
result = await Runner.run(
agent,
build_user_prompt(question, agent_config),
run_config=RunConfig(
sandbox=SandboxRunConfig(session=session),
workflow_name="Synthetic dataroom diligence",
trace_id=trace_id,
trace_metadata=trace_metadata,
tracing_disabled=trace_id is None,
),
max_turns=30,
)
for filename in agent_config.required_artifacts:
try:
with custom_span(
"artifact_write",
{
"tool.name": "artifact_write",
"tool.input": {"filename": filename},
},
disabled=trace_context is None,
):
with await session.read(Path("outputs") / filename) as handle:
(output_dir / filename).write_bytes(handle.read())
except Exception:
continue
return str(result.final_output)
finally:
delete = getattr(client, "delete", None)
if delete is not None and session is not None:
try:
await delete(session)
except Exception:
pass
if trace_context is not None:
trace_context.__exit__(None, None, None)
if halo_processor is not None:
try:
flush_traces()
except Exception:
pass
try:
halo_processor.shutdown()
except Exception:
pass
@contextmanager
def staged_dataset_mount(dataset_dir: Path) -> Iterator[Path]:
"""Prepare a writable SDK mount copy without mutating the source dataroom."""
with tempfile.TemporaryDirectory(prefix="synthetic-dataroom-mount-") as tmp:
staged_dir = Path(tmp) / dataset_dir.name
shutil.copytree(dataset_dir, staged_dir)
write_runtime_tools(staged_dir)
yield staged_dir.resolve()
def write_runtime_manifest(dataset_dir: Path) -> None:
manifest = {
"runtime_scope": "sdk_agent_visible_dataroom",
"files": sorted(
str(path.relative_to(dataset_dir))
for path in dataset_dir.rglob("*")
if path.is_file() and path.name != "manifest.json"
),
}
(dataset_dir / "manifest.json").write_text(
json.dumps(manifest, indent=2) + "\n",
encoding="utf-8",
)第 3 步：生成追踪运行
问题的设置经过深思熟虑，具有多样性，以便评估套件覆盖智能体出错的几种方式。笔记本默认运行五次轨迹，以在保持实时路径实用性的同时，覆盖几种不同的行为。如果你以后想要更广泛的覆盖范围，可以使用更大的题库。
每次运行都使用异步 Agents SDK 路径，并写入真实的追踪记录和必需的制品。
QUESTION_BANK = [
"What do runway and burn tell us about near-term financing risk?",
"How strong is revenue quality, and which ARR figure should we rely on?",
"What is the real customer concentration risk after parent-account rollups?",
"What legal exposure should an acquirer investigate first?",
"How ready is the company for enterprise security review?",
"Which contradictions appear across the board deck, finance exports, and management narratives?",
"What unsupported metrics should we refuse to infer from the dataroom?",
"What follow-up questions should management answer before an investment committee review?",
"What are the top three diligence risks, ranked by severity?",
"Which claims in the materials look directionally useful but still need stronger evidence?",
]
# Using 5 questions as the default, with more available if you want broader coverage later.
DEFAULT_TRACE_INDICES = [0, 1, 2, 4, 6]
TRACE_LIMIT = len(DEFAULT_TRACE_INDICES)
QUESTIONS = [QUESTION_BANK[index] for index in DEFAULT_TRACE_INDICES]
@dataclass
class TraceRecord:
trace_id: str
sdk_trace_id: str
trace_label: str
question: str
answer: str
output_dir: str
mode: str
def sdk_trace_id(label: str) -> str:
# Agents SDK trace uploads expect ids shaped like `trace_<hex>`.
return f"trace_{hashlib.sha256(label.encode('utf-8')).hexdigest()[:32]}"
def exported_trace_id(label: str) -> str:
# The local HALO exporter strips the SDK `trace_` prefix before writing JSONL.
return sdk_trace_id(label).removeprefix("trace_")
async def generate_traces(dataset: Path, questions: list[str]) -> list[TraceRecord]:
traces: list[TraceRecord] = []
for index, question in enumerate(questions, start=1):
label = f"trace-{index:02d}"
print(f"Running {label}/{len(questions):02d}: {question}")
output_dir = TRACE_DIR / f"trace_{index:02d}"
output_dir.mkdir(parents=True, exist_ok=True)
real_sdk_trace_id = sdk_trace_id(label)
real_exported_trace_id = exported_trace_id(label)
answer = await run_sdk_agent(
dataset_dir=dataset,
output_dir=output_dir,
question=question,
model=AGENT_MODEL,
agent_config=agent_config,
trace_id=real_sdk_trace_id,
trace_metadata={"notebook_trace_id": label},
halo_trace_path=HALO_TRACE_PATH,
)
traces.append(
TraceRecord(
trace_id=real_exported_trace_id,
sdk_trace_id=real_sdk_trace_id,
trace_label=label,
question=question,
answer=answer,
output_dir=str(output_dir.relative_to(PROJECT_ROOT)),
mode="sdk",
)
)
return traces
trace_generation_started = time.perf_counter()
traces = await generate_traces(dataset, QUESTIONS)
print(f"Trace generation completed in {format_duration(time.perf_counter() - trace_generation_started)}")
assert len(traces) == TRACE_LIMIT
for trace in traces:
print(f"{trace.trace_label}: {trace.question}")
print(textwrap.shorten(trace.answer.replace("\n", " "), width=180, placeholder="..."))
print()Running trace-01/05: What do runway and burn tell us about near-term financing risk?
Running trace-02/05: How strong is revenue quality, and which ARR figure should we rely on?
Running trace-03/05: What is the real customer concentration risk after parent-account rollups?
Running trace-04/05: How ready is the company for enterprise security review?
Running trace-05/05: What unsupported metrics should we refuse to infer from the dataroom?
Trace generation completed in 7m 59s
trace-01: What do runway and burn tell us about near-term financing risk?
Near-term financing risk is elevated. Finance reports `$2.9M` monthly cash burn and `11 months` runway, and the board packet corroborates both figures....
trace-02: How strong is revenue quality, and which ARR figure should we rely on?
**Answer** - Revenue quality is **moderate, not clean**: real scale and 69% gross margin, but ARR definition drift, unvalidated retention, concentration, and renewal risk weaken...
trace-03: What is the real customer concentration risk after parent-account rollups?
**Answer** - Real concentration risk is **high**: Northstar Bank + Northstar Capital Markets roll up to **Northstar Holdings at $12.4M**, or **33.6% of controlled FY2025 ARR**....
trace-04: How ready is the company for enterprise security review?
**Answer** - The company is **partially ready, but not ready for frictionless enterprise security review**: SOC 2 Type I is complete, but SOC 2 Type II fieldwork is still in...
trace-05: What unsupported metrics should we refuse to infer from the dataroom?
**Answer** Refuse to infer these unsupported or conflicted metrics from the dataroom: - `CAC payback`: explicitly `not_provided`; requested but not supplied....检查智能体制品
每次追踪运行都会写入框架所需的全部制品集。下方的第一次运行展示了智能体生成的文件，以便你可以同时检查答案、证据和未决问题。
def show_trace_artifacts(trace: TraceRecord) -> None:
output_dir = PROJECT_ROOT / trace.output_dir
for artifact in agent_config.required_artifacts:
path = output_dir / artifact
language = {
".md": "markdown",
".json": "json",
".csv": "csv",
}.get(path.suffix, "text")
display(Markdown(f"### `{artifact}`\n```{language}\n{path.read_text(encoding='utf-8').rstrip()}\n```"))
show_trace_artifacts(traces[0])summary_answer.md
# Summary Answer
Runway and burn indicate elevated near-term financing risk. Finance reports FY2025 cash burn of $2.9M per month and 11 months of runway, and the December board packet repeats the same burn and runway figures. (`financials/p_and_l.csv`, `board_deck.md`)
An 11-month runway is a sub-12-month financing window: unless burn is reduced, revenue conversion accelerates, or additional capital is secured, the company likely needs a financing plan in the near term. (`financials/p_and_l.csv`)
The financing story is somewhat weakened by ARR quality and source conflicts. The controlled FY2025 ARR bridge shows $36.9M ending ARR, while the board deck reports $43.0M because it includes $2.8M of launch-stage commitments and $1.1M of usage true-ups that finance does not classify as recurring ARR. (`financials/arr_bridge.csv`, `financials/revenue_recognition_notes.md`, `board_deck.md`)
The dataroom does not provide a cash balance, debt schedule, undrawn facility, covenant package, or financing plan, so the exact liquidity cushion and financing path are unknown from the provided evidence. (`financials/p_and_l.csv`, `manifest.json`)investment_memo.md
# Investment Memo: Runway and Burn
## Bottom Line
- Near-term financing risk is elevated because finance reports $2.9M of monthly cash burn and only 11 months of runway. (`financials/p_and_l.csv`)
- The board packet corroborates the same $2.9M monthly burn and 11-month runway. (`board_deck.md`)
- The exact liquidity cushion is unknown because the dataroom provides runway and burn but not cash balance, debt availability, covenant terms, or a financing plan. (`financials/p_and_l.csv`, `manifest.json`)
## Evidence
- FY2025 P&L reports $30.26M revenue, 69% gross margin, $47.71M opex, $2.9M cash burn per month, and 11 months of runway. (`financials/p_and_l.csv`)
- Finance-controlled ARR is $36.9M at FY2025 year-end. (`financials/arr_bridge.csv`)
- The board deck reports $43.0M FY2025 ending ARR, 71% ARR growth, 69% gross margin, $2.9M monthly burn, and 11 months of runway. (`board_deck.md`)
- Finance states the board ARR includes $2.8M signed launch-stage commitments not live by 2025-12-31 and $1.1M usage true-ups that finance does not classify as recurring ARR. (`financials/revenue_recognition_notes.md`)
## Interpretation
- A company burning $2.9M per month with 11 months of runway has less than one year to reduce burn, convert growth into cash-efficient revenue, or raise capital. (`financials/p_and_l.csv`)
- The growth narrative should be underwritten against finance-controlled ARR rather than board headline ARR because finance identifies specific non-recurring or not-yet-live components in the board figure. (`financials/arr_bridge.csv`, `financials/revenue_recognition_notes.md`, `board_deck.md`)
- Current evidence supports a financing-risk concern, but it does not support quantifying exact cash balance, facility availability, covenant headroom, or planned raise timing. (`financials/p_and_l.csv`, `manifest.json`)
## Diligence View
- Financing risk: High / elevated.
- Key dependency: management must show a credible plan to extend runway beyond the reported 11 months.
- Critical missing evidence: cash balance, monthly cash forecast, debt/facility details, covenant headroom, and financing plan.risk_register.json
[
{
"id": "R-001",
"risk": "Sub-12-month runway",
"severity": "High",
"rationale": "Finance reports 11 months of runway and $2.9M of monthly cash burn, which indicates a near-term need to reduce burn, improve cash generation, or secure financing.",
"evidence": [
"financials/p_and_l.csv",
"board_deck.md"
],
"open_questions": [
"What is current unrestricted cash?",
"What financing actions are planned before runway drops below 6 months?"
]
},
{
"id": "R-002",
"risk": "ARR quality may weaken financing narrative",
"severity": "Medium",
"rationale": "Finance-controlled FY2025 ending ARR is $36.9M, while the board deck reports $43.0M ARR because it includes launch-stage commitments and usage true-ups that finance does not classify as recurring ARR.",
"evidence": [
"financials/arr_bridge.csv",
"financials/revenue_recognition_notes.md",
"board_deck.md"
],
"open_questions": [
"Which ARR figure is used in lender or investor materials?",
"How much of the launch-stage commitments have since gone live?"
]
},
{
"id": "R-003",
"risk": "Liquidity structure is not evidenced",
"severity": "Medium",
"rationale": "The dataroom provides burn and runway but does not provide cash balance, debt availability, covenant headroom, or a financing plan, limiting confidence in the company\u2019s liquidity path.",
"evidence": [
"financials/p_and_l.csv",
"manifest.json"
],
"open_questions": [
"Is there an undrawn revolver or venture debt facility?",
"Are there covenants or minimum cash requirements?"
]
}
]open_questions.md
# Open Questions
- What is current unrestricted cash, and how does it reconcile to the reported 11 months of runway? (`financials/p_and_l.csv`)
- Is there an existing debt facility, undrawn revolver, covenant package, or minimum cash requirement? (`manifest.json`)
- What is management's financing plan, including target timing, amount, and contingency if markets are unavailable? (`manifest.json`)
- What burn reduction actions are available, and how many months of runway would each action add? (`financials/p_and_l.csv`)
- Which ARR figure is used in financing discussions: finance-controlled $36.9M ARR or board headline $43.0M ARR? (`financials/arr_bridge.csv`, `financials/revenue_recognition_notes.md`, `board_deck.md`)citations.json
[
{
"claim_id": "C-001",
"claim": "Finance reports FY2025 cash burn of $2.9M per month and 11 months of runway.",
"sources": [
"financials/p_and_l.csv"
]
},
{
"claim_id": "C-002",
"claim": "The December board packet repeats $2.9M monthly cash burn and 11 months of runway.",
"sources": [
"board_deck.md"
]
},
{
"claim_id": "C-003",
"claim": "Finance-controlled FY2025 ending ARR is $36.9M.",
"sources": [
"financials/arr_bridge.csv"
]
},
{
"claim_id": "C-004",
"claim": "The board deck reports $43.0M FY2025 ending ARR and 71% ARR growth.",
"sources": [
"board_deck.md"
]
},
{
"claim_id": "C-005",
"claim": "Finance states board ARR includes $2.8M of launch-stage commitments not live by 2025-12-31 and $1.1M of usage true-ups that finance does not classify as recurring ARR.",
"sources": [
"financials/revenue_recognition_notes.md"
]
},
{
"claim_id": "C-006",
"claim": "The dataroom does not provide a separate cash balance, debt schedule, facility availability, covenant package, or financing plan.",
"sources": [
"financials/p_and_l.csv",
"manifest.json"
]
}
]evidence_table.csv
claim_id,claim,sources
C-001,"Finance reports FY2025 cash burn of $2.9M per month and 11 months of runway.","financials/p_and_l.csv"
C-002,"The December board packet repeats $2.9M monthly cash burn and 11 months of runway.","board_deck.md"
C-003,"Finance-controlled FY2025 ending ARR is $36.9M.","financials/arr_bridge.csv"
C-004,"The board deck reports $43.0M FY2025 ending ARR and 71% ARR growth.","board_deck.md"
C-005,"Finance states board ARR includes $2.8M of launch-stage commitments not live by 2025-12-31 and $1.1M of usage true-ups that finance does not classify as recurring ARR.","financials/revenue_recognition_notes.md"
C-006,"The dataroom does not provide a separate cash balance, debt schedule, facility availability, covenant package, or financing plan.","financials/p_and_l.csv; manifest.json"第 4 步：生成示例性人工反馈和模型洞察
本节模拟了一位人类专家在智能体运行后审查轨迹的过程。在实际的尽职调查工作流程中，这可能是财务主管或其他了解决策细节的案例专家。在此示例中，审查员指出：母账户汇总比法律实体集中度更重要；未经核实的管理层 NRR 估算不应成为官方指标；且当证据仅支持 Type I 时，“SOC 2 已完成”的说法过于模糊。
模型生成的洞察保持独立。在完全自动化的路径中，LLM 会审查相同的轨迹并提出反复出现的问题或缺失的行为。那额外的处理步骤提高了覆盖范围，而主题专家评审则增加了扎根于工作本身的领域判断力。
def feedback_item(
trace: TraceRecord,
summary: str,
required: list[str],
prohibited: list[str],
theme: str,
) -> dict[str, Any]:
return {
"feedback_id": f"human-{trace.trace_label}",
"trace_id": trace.trace_id,
"trace_label": trace.trace_label,
"question": trace.question,
"source_type": "human_feedback",
"theme": theme,
"summary": summary,
"required_observations": required,
"prohibited_claims": prohibited,
}
def generate_mock_human_feedback(traces: list[TraceRecord]) -> list[dict[str, Any]]:
specs_by_question = {
"What do runway and burn tell us about near-term financing risk?": (
"State both the 11-month runway and rising burn as financing risk, not just a generic red flag.",
["Name the 11-month runway", "Tie burn to near-term financing pressure"],
["Do not imply the company has more than 12 months of runway"],
"financial_risk",
),
"How strong is revenue quality, and which ARR figure should we rely on?": (
"Use the controlled ARR bridge as the reliable figure and preserve the board-versus-finance contradiction.",
["Prefer finance ARR over board ARR", "Preserve the ARR contradiction"],
["Do not silently reconcile the ARR gap"],
"revenue_quality",
),
"What is the real customer concentration risk after parent-account rollups?": (
"Roll concentration up to Northstar Holdings. Legal-entity framing understates the real dependency.",
["Mention parent-account concentration", "Use account_hierarchy.csv"],
["Do not stop at legal-entity concentration"],
"customer_concentration",
),
"How ready is the company for enterprise security review?": (
"Be exact about certification status: Type I is complete; Type II is still in progress.",
["Distinguish Type I from Type II", "Treat sales FAQ as weaker evidence"],
["Do not say SOC 2 is simply complete"],
"security_readiness",
),
"What unsupported metrics should we refuse to infer from the dataroom?": (
"Refuse official NRR and CAC payback when the dataroom does not support them.",
["Mark official NRR unsupported", "Mark CAC payback unsupported"],
["Do not promote the management NRR estimate into an official metric"],
"unsupported_metrics",
),
}
return [feedback_item(trace, *specs_by_question[trace.question]) for trace in traces]
def extract_json(text: str) -> Any:
text = text.strip()
fenced = re.search(r"```(?:json)?\s*(.*?)```", text, flags=re.DOTALL)
candidate = fenced.group(1).strip() if fenced else text
return json.loads(candidate)
def generate_llm_feedback(traces: list[TraceRecord]) -> list[dict[str, Any]]:
payload = [asdict(trace) for trace in traces]
response = client.responses.create(
model=ANALYSIS_MODEL,
input=f"""
You are reviewing traces from a financial diligence analyst agent.
Return JSON only: a list of objects with keys `insight_id`, `trace_id`, `question`, `source_type`, and `observations`.
Use `source_type` = `llm_insight`.
For `trace_id`, copy the provided `trace_id` field exactly; do not use `sdk_trace_id` or `trace_label`.
For each trace, identify concise recurring-behavior observations that could help generate evals later.
Do not restate the whole answer. Do not invent unavailable evidence.
Traces:
{json.dumps(payload, indent=2)}
""".strip(),
)
parsed = extract_json(response.output_text)
if not isinstance(parsed, list):
raise ValueError("Expected a JSON list of LLM insights.")
trace_labels = {trace.trace_id: trace.trace_label for trace in traces}
for item in parsed:
try:
item["trace_label"] = trace_labels[item["trace_id"]]
except KeyError as exc:
raise ValueError(f"Unknown trace_id in LLM feedback: {item['trace_id']}") from exc
return parsed
feedback_started = time.perf_counter()
human_feedback = generate_mock_human_feedback(traces)
llm_feedback = generate_llm_feedback(traces)
print(f"Feedback generation completed in {format_duration(time.perf_counter() - feedback_started)}")
assert len(human_feedback) == TRACE_LIMIT
assert len(llm_feedback) == TRACE_LIMIT
print("Human feedback items:", len(human_feedback))
print("LLM insight items:", len(llm_feedback))
print("\nExample human feedback:")
print(json.dumps(human_feedback[0], indent=2))
print("\nExample LLM insight:")
print(json.dumps(llm_feedback[0], indent=2))Feedback generation completed in 13s
Human feedback items: 5
LLM insight items: 5
Example human feedback:
{
"feedback_id": "human-trace-01",
"trace_id": "43d9b03619a9d2ed4d2f3e3fd17c8bf4",
"trace_label": "trace-01",
"question": "What do runway and burn tell us about near-term financing risk?",
"source_type": "human_feedback",
"theme": "financial_risk",
"summary": "State both the 11-month runway and rising burn as financing risk, not just a generic red flag.",
"required_observations": [
"Name the 11-month runway",
"Tie burn to near-term financing pressure"
],
"prohibited_claims": [
"Do not imply the company has more than 12 months of runway"
]
}
Example LLM insight:
{
"insight_id": "llm_insight_01",
"trace_id": "43d9b03619a9d2ed4d2f3e3fd17c8bf4",
"question": "What do runway and burn tell us about near-term financing risk?",
"source_type": "llm_insight",
"observations": [
"Flags elevated financing risk when runway is under 12 months and monthly burn is cited from finance and board sources.",
"Prefers finance-controlled ARR over board headline ARR when ARR definitions conflict.",
"Explicitly identifies missing liquidity data such as cash balance, debt availability, covenants, and financing plan.",
"Includes source citations for key numeric claims and notes validation/artifact completion."
],
"trace_label": "trace-01"
}第 5 步：从轨迹和反馈中生成 Promptfoo 评估
评估套件由 LLM 根据迄今收集的证据动态生成：追踪的行为、人工反馈和模型生成的观察结果。这使得评估意见成为下一次框架修订稍后可以再次运行的测试。
Promptfoo 是一个用于评估和红队测试 LLM 应用程序的开源 CLI 和库。在本笔记本中，生成的行为成为 Promptfoo 测试用例：每一个都可以将文字断言与 LLM 评分准则结合起来，因此同一个门禁既可以检查确切的要求，也可以检查语义审查意图。
评估是值得主题专家和开发者投入手动工作的地方。完全自动化的路径可以快速提出有用的评估，但在成为长期测试套件的一部分之前，人们仍应检查评估是否准确、具有代表性，并衡量真正重要的行为。
def generate_feedback_derived_evals(
traces: list[TraceRecord],
human_feedback: list[dict[str, Any]],
llm_feedback: list[dict[str, Any]],
) -> list[dict[str, Any]]:
min_eval_count = min(5, max(2, len(traces)))
max_eval_count = min(7, max(min_eval_count, len(traces) + 2))
response = client.responses.create(
model=EVAL_GENERATION_MODEL,
input=f"""
You are designing an eval suite for an OpenAI Agents SDK-backed financial diligence analyst.
Use the traces, human feedback, and LLM insights below to generate {min_eval_count} to {max_eval_count} durable eval definitions.
Return JSON only: a list of objects with keys `eval_id`, `title`, `scoring_method`, `expected_behavior`, `source_trace_id`, `rubric`, `deterministic_assertions`, `suggested_pass_example`, and `suggested_fail_example`.
`scoring_method` must be one of `deterministic`, `llm_judge`, or `hybrid`.
`source_trace_id` must exactly match the provided `trace_id` field for the trace whose answer should be scored. Do not use `sdk_trace_id` or `trace_label` for this field; those are only for SDK transport and human-readable references.
`rubric` must be a concise pass/fail grading rubric suitable for Promptfoo `llm-rubric`.
`deterministic_assertions` must be a list of Promptfoo-style assertion objects and may use only `contains`, `icontains`, or `not-contains` when a literal check is clearly useful; otherwise return an empty list.
Prefer reusable behaviors over one-off trace restatements.
Traces:
{json.dumps([asdict(trace) for trace in traces], indent=2)}
Human feedback:
{json.dumps(human_feedback, indent=2)}
LLM insights:
{json.dumps(llm_feedback, indent=2)}
""".strip(),
)
parsed = extract_json(response.output_text)
if not isinstance(parsed, list):
raise ValueError("Expected a JSON list of eval definitions.")
trace_labels = {trace.trace_id: trace.trace_label for trace in traces}
for item in parsed:
try:
item["source_trace_label"] = trace_labels[item["source_trace_id"]]
except KeyError as exc:
raise ValueError(f"Unknown source_trace_id in generated eval: {item['source_trace_id']}") from exc
return parsed
eval_generation_started = time.perf_counter()
eval_suite = generate_feedback_derived_evals(traces, human_feedback, llm_feedback)
print(f"Eval generation completed in {format_duration(time.perf_counter() - eval_generation_started)}")
assert all({"title", "scoring_method", "suggested_pass_example", "suggested_fail_example", "expected_behavior", "source_trace_id", "rubric", "deterministic_assertions"} <= set(item) for item in eval_suite)
def markdown_table(rows: list[dict[str, Any]], columns: list[str]) -> str:
header = "| " + " | ".join(columns) + " |"
divider = "| " + " | ".join(["---"] * len(columns)) + " |"
body = ["| " + " | ".join(str(row[column]) for column in columns) + " |" for row in rows]
return "\n".join([header, divider, *body])
display(Markdown(markdown_table(eval_suite, ["title", "scoring_method", "expected_behavior"])))
for item in eval_suite:
print(f"\n{item['title']}")
print(" pass:", item["suggested_pass_example"])
print(" fail:", item["suggested_fail_example"])Eval generation completed in 52s| 标题 | 评分方法 | 预期行为 |
|---|---|---|
| 运营时间和现金消耗必须转化为近期融资风险 | 混合（hybrid） | 答案应明确指出，由于运营时间（runway）仅为 11 个月且月度现金消耗巨大/上升，融资风险较高，并将现金消耗与减少支出、提高现金转换或在 12 个月内运营时间耗尽前筹集资本的压力联系起来。不应暗示公司拥有超过 12 个月的运营时间。 |
| 收入质量评估必须优先考虑财务管控的 ARR，并保留 ARR 矛盾 | 混合（hybrid） | 答案应将收入质量描述为混合或中等，而不是干净的；在承销时应依赖财务管控的 2025 财年年末 ARR（约 3690 万美元）；并明确拒绝或限定 4300 万美元的董事会/头条 ARR 和 4080 万美元的预订调整后 ARR，认为它们不等于经常性 ARR。应保留矛盾之处，而不是默默调和差距。 |
| 客户集中度必须在母账户汇总后进行评估 | 混合（hybrid） | 答案应在评估集中度之前将法律实体汇总至母账户，特别承认 Northstar Holdings 是真正的母公司风险敞口。应使用财务管控的 ARR 作为分母，引用或参考账户层级证据，并避免停留在法律实体集中度层面。 |
| 企业安全准备情况必须区分 SOC 2 Type I 和 Type II | 混合（hybrid） | 答案应说明该公司仅为企业安全审查做好了部分准备，因为 SOC 2 Type I 已完成，但 SOC 2 Type II 仍在进行中，且尚未发布 Type II 报告。应将销售 FAQ 中的语言（如“SOC 2 已完成”）视为较弱或潜在的误导性证据，并将缺失的 Type II 证据与企业采购或客户摩擦联系起来。 |
| 必须拒绝而不是推断未经支持的指标 | 混合（hybrid） | 答案应拒绝推断缺失、冲突、部分或管理层估算的指标。特别是，必须将 CAC（获客成本）回收期标记为不支持/未提供，并将官方 NRR 标记为不支持，因为 122% 的 NRR 仅是未经核实的管理层估算。不应将部分或非官方指标提升为确定的尽职调查指标。 |
Runway and burn must be translated into near-term financing risk
pass: Near-term financing risk is elevated: the company has only 11 months of runway and meaningful monthly burn, creating pressure to reduce burn, improve cash conversion, or raise capital within a sub-12-month window.
fail: Financing risk appears manageable because the company has enough runway for the next year and should be able to continue operating without near-term funding pressure.
Revenue quality assessment must prefer finance-controlled ARR and preserve ARR contradictions
pass: Revenue quality is moderate, not clean. For underwriting, use the finance-controlled ARR bridge at $36.9M, while treating the $43.0M board ARR and $40.8M bookings-adjusted view as non-comparable or planning figures because they include items not classified as recurring ARR.
fail: Revenue quality is strong and the company has $43.0M of ARR; the board number can be used because it reconciles to the finance ARR bridge after normal adjustments.
Customer concentration must be assessed after parent-account rollups
pass: Concentration risk is high after parent rollups: Northstar Bank and Northstar Capital Markets roll up to Northstar Holdings at about $12.4M, roughly one-third of finance-controlled ARR. Looking only at legal entities understates the dependency.
fail: Customer concentration is acceptable because no single legal entity exceeds the threshold after reviewing the top-customer list.
Enterprise security readiness must distinguish SOC 2 Type I from Type II
pass: The company is partially ready, not frictionless: SOC 2 Type I is complete, but Type II fieldwork is still in progress and no Type II report is available. Sales materials saying 'SOC 2 complete' should be treated cautiously because enterprise buyers are waiting on Type II evidence.
fail: The company is ready for enterprise security review because SOC 2 is complete and the sales FAQ confirms there should be no security blocker.
Unsupported metrics must be refused rather than inferred
pass: Refuse to infer CAC payback because it is not provided. Also refuse to treat 122% NRR as official; it is an unvalidated management estimate and should not be used as a definitive retention metric.
fail: The company has 122% official NRR and CAC payback appears attractive based on its revenue growth, so both can be used in underwriting.第 6 步：使用 Promptfoo 验证当前框架
Promptfoo 将生成的测试针对当前的追踪输出进行运行。这为循环提供了框架当前表现良好之处以及哪些期望仍未达标的快照。Promptfoo 适合担任此角色，因为它能将文字要求的确定性检查与用于语义质量的 llm-rubric 评分结合起来。
在本笔记本中，Promptfoo 门禁对现有的追踪输出进行评分。要验证未来的框架修订，请用运行候选智能体的提供程序替换追踪输出提供程序。那些 Promptfoo 结果成为下文传递给 HALO 的优化输入的一部分。即使评估生成是自动化的，人类仍可以在让它们引导重复优化之前收紧薄弱的评估。
构建 Promptfoo 测试框架
提供程序将现有的追踪输出传回给 Promptfoo，测试构建器将生成的评估定义转化为可运行的 Promptfoo 用例。
PROMPTFOO_PROVIDER = r'''from __future__ import annotations
import json
from pathlib import Path
def call_api(prompt: str, options: dict, context: dict) -> dict:
config = options.get("config", {})
trace_outputs = json.loads(Path(config["trace_outputs_path"]).read_text(encoding="utf-8"))
trace_id = (context.get("vars") or {}).get("trace_id")
trace = trace_outputs[trace_id]
return {
"output": trace["answer"],
"metadata": {
"trace_id": trace_id,
"question": trace["question"],
},
}
'''
def trace_for_eval(item: dict[str, Any], traces: list[TraceRecord]) -> TraceRecord:
trace_by_id = {trace.trace_id: trace for trace in traces}
try:
return trace_by_id[item["source_trace_id"]]
except KeyError as exc:
raise ValueError(f"Unknown source_trace_id in generated eval: {item['source_trace_id']}") from exc
def promptfoo_test_from_eval(item: dict[str, Any], trace: TraceRecord) -> dict[str, Any]:
assertions = [
assertion
for assertion in item.get("deterministic_assertions") or []
if isinstance(assertion, dict)
and assertion.get("type") in {"contains", "icontains", "not-contains"}
and assertion.get("value")
]
assertions.append({
"type": "llm-rubric",
"provider": f"openai:{JUDGE_MODEL}",
"threshold": 0.8,
"value": item["rubric"],
})
return {
"description": item["title"],
"vars": {
"question": trace.question,
"trace_id": trace.trace_id,
"trace_label": trace.trace_label,
},
"metadata": {
"eval_id": item["eval_id"],
"scoring_method": item["scoring_method"],
},
"assert": assertions,
}
def write_promptfoo_artifacts(eval_suite: list[dict[str, Any]], traces: list[TraceRecord]) -> dict[str, Path]:
promptfoo_dir = ARTIFACT_DIR / "promptfoo"
promptfoo_dir.mkdir(parents=True, exist_ok=True)
provider_path = promptfoo_dir / "trace_output_provider.py"
trace_outputs_path = promptfoo_dir / "trace_outputs.json"
config_path = promptfoo_dir / "promptfooconfig.yaml"
output_path = promptfoo_dir / "promptfoo_results.json"
provider_path.write_text(PROMPTFOO_PROVIDER, encoding="utf-8")
trace_outputs_path.write_text(
json.dumps({trace.trace_id: asdict(trace) for trace in traces}, indent=2) + "\n",
encoding="utf-8",
)
tests = [promptfoo_test_from_eval(item, trace_for_eval(item, traces)) for item in eval_suite]
config = {
"description": "Feedback-derived diligence eval gate",
"prompts": ["{{question}}"],
"providers": [{
"id": "file://trace_output_provider.py",
"label": "current-trace-output",
"config": {"trace_outputs_path": str(trace_outputs_path)},
}],
"tests": tests,
}
# JSON is valid YAML, which keeps the generated config easy to inspect without
# adding another serialization dependency to the notebook.
config_path.write_text(json.dumps(config, indent=2) + "\n", encoding="utf-8")
return {
"dir": promptfoo_dir,
"provider": provider_path,
"trace_outputs": trace_outputs_path,
"config": config_path,
"output": output_path,
}
def promptfoo_summary(path: Path) -> dict[str, Any]:
data = json.loads(path.read_text(encoding="utf-8"))
results = (data.get("results") or {}).get("outputs") or (data.get("results") or {}).get("results") or []
rows = []
for result in results:
grading = result.get("gradingResult") or {}
components = grading.get("componentResults") or []
failing_component = next(
(
component
for component in components
if isinstance(component, dict) and component.get("pass") is False
),
None,
)
reason = str(grading.get("reason") or "")
if not reason and failing_component:
reason = str(failing_component.get("reason") or "")
if not reason and components and isinstance(components[0], dict):
reason = str(components[0].get("reason") or "")
test_case = result.get("testCase") or {}
test_vars = test_case.get("vars") or {}
rows.append({
"eval_id": (test_case.get("metadata") or {}).get("eval_id"),
"title": test_case.get("description") or "Untitled",
"trace_id": test_vars.get("trace_id"),
"trace_label": test_vars.get("trace_label"),
"passed": bool(result.get("success")),
"score": result.get("score"),
"explanation": reason,
})
return {
"backend": "promptfoo",
"total": len(rows),
"passed": sum(row["passed"] for row in rows),
"failed": sum(not row["passed"] for row in rows),
"rows": rows,
}运行 Promptfoo 门禁
执行生成的套件并总结当前框架的结果。
def run_promptfoo_feedback_eval_gate(eval_suite: list[dict[str, Any]], traces: list[TraceRecord]) -> dict[str, Any]:
artifacts = write_promptfoo_artifacts(eval_suite, traces)
command = [
"npx",
"--yes",
f"promptfoo@{PROMPTFOO_VERSION}",
"eval",
"--no-cache",
"--no-table",
"-c",
str(artifacts["config"]),
"-o",
str(artifacts["output"]),
]
env = os.environ.copy()
env["PROMPTFOO_PYTHON"] = sys.executable
env["PROMPTFOO_CONFIG_DIR"] = str(artifacts["dir"] / ".promptfoo")
env["PROMPTFOO_DISABLE_WAL_MODE"] = "true"
process = subprocess.run(
command,
cwd=artifacts["dir"],
env=env,
text=True,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
check=False,
)
if not artifacts["output"].exists():
raise RuntimeError(f"Promptfoo did not write results. Output:\n{process.stdout[-4000:]}")
summary = promptfoo_summary(artifacts["output"])
summary["command"] = command
summary["returncode"] = process.returncode
summary["result_path"] = str(artifacts["output"].relative_to(PROJECT_ROOT))
summary["log_tail"] = process.stdout[-4000:]
return summary
promptfoo_started = time.perf_counter()
gate_result = run_promptfoo_feedback_eval_gate(eval_suite, traces)
print(f"Promptfoo gate completed in {format_duration(time.perf_counter() - promptfoo_started)}")
display(Markdown(markdown_table(gate_result["rows"], ["title", "trace_label", "passed", "score", "explanation"])))
print({key: gate_result[key] for key in ["backend", "total", "passed", "failed", "result_path"]})Promptfoo gate completed in 9s| 标题 | 追踪标签 | 通过 | 分数 | 解释 |
|---|---|---|---|---|
| 运营时间和现金消耗必须转化为近期融资风险 | trace-01 | True | 1 | 所有断言通过 |
| 收入质量评估必须优先考虑财务管控的 ARR，并保留 ARR 矛盾 | trace-02 | True | 1 | 所有断言通过 |
| 客户集中度必须在母账户汇总后进行评估 | trace-03 | True | 1 | 所有断言通过 |
| 企业安全准备情况必须区分 SOC 2 Type I 和 Type II | trace-04 | True | 1 | 所有断言通过 |
| 必须拒绝而不是推断未经支持的指标 | trace-05 | True | 1 | 所有断言通过 |
{'backend': 'promptfoo', 'total': 5, 'passed': 5, 'failed': 0, 'result_path': 'examples/agents_sdk/agent_improvement_loop_artifacts/promptfoo/promptfoo_results.json'}第 7 步：运行 HALO 并编写移交方案
HALO（Hierarchical Agent Loop Optimization，分层智能体循环优化）是一种通过执行轨迹改进智能体框架的方法论和 Python 包。HALO 仓库描述了一个循环：收集轨迹、分析反复出现的框架级故障、将最终报告交给编码智能体，并在框架变更后重复此过程。
这是循环将积累的证据转化为建议的框架变更的节点。HALO 在审查当前框架的同时，会结合智能体追踪、人工反馈、模型反馈、生成的评估和 Promptfoo 结果。然后，它为下一次实施阶段产出一组排名后的变更。
HALO 在此处的价值在于它能同时对整个循环进行推理。它可以结合人类判断、运行时行为和评估结果，然后将结果打包为一个移交方案，Codex 可以利用该方案实施改进框架的代码变更。
收集 HALO 输入
构建一个将当前框架、追踪、反馈、评估和门禁结果保持在一起的上下文对象。
from datetime import datetime, timezone
def serialize_agent_config(config: AgentConfig) -> dict[str, Any]:
return {
"version": config.version,
"system_prompt": config.system_prompt,
"model_settings": asdict(config.model_settings),
"tool_policy": config.tool_policy,
"eval_metadata": config.eval_metadata,
}
def build_halo_context(
traces: list[TraceRecord],
human_feedback: list[dict[str, Any]],
llm_feedback: list[dict[str, Any]],
eval_suite: list[dict[str, Any]],
gate_result: dict[str, Any],
agent_config: AgentConfig,
) -> dict[str, Any]:
return {
"traces": [asdict(trace) for trace in traces],
"human_feedback": human_feedback,
"llm_feedback": llm_feedback,
"eval_suite": eval_suite,
"gate_result": gate_result,
"agent_config": serialize_agent_config(agent_config),
}
def synthetic_trace_id(value: str) -> str:
return hashlib.sha256(f"halo-context-{value}".encode("utf-8")).hexdigest()[:32]
def synthetic_span_id(value: str) -> str:
return hashlib.sha256(value.encode("utf-8")).hexdigest()[:16]
def synthetic_span(*, trace_id: str, span_id: str, name: str, observation_kind: str, attributes: dict[str, Any]) -> dict[str, Any]:
now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%f000Z")
return {
"trace_id": trace_id,
"span_id": span_id,
"parent_span_id": "",
"trace_state": "",
"name": name,
"kind": "SPAN_KIND_INTERNAL",
"start_time": now,
"end_time": now,
"status": {"code": "STATUS_CODE_OK", "message": ""},
"resource": {"attributes": {"service.name": "financial-diligence-analyst"}},
"scope": {"name": "halo-optimization-context", "version": "1"},
"attributes": {
"openinference.span.kind": observation_kind,
"inference.export.schema_version": 1,
"inference.project_id": "financial_diligence_analyst_optimization_context",
"inference.observation_kind": observation_kind,
**attributes,
},
}
def halo_input_summary(context: dict[str, Any]) -> str:
rows = [
("Current harness config", 1, "global config span", "system prompt, model settings, tool policy, eval metadata"),
("SDK execution traces", len(context["traces"]), "original runtime traces", "agent steps, tool calls, outputs"),
("Human feedback", len(context["human_feedback"]), "appended to the source trace", "reviewer summary, required observations, prohibited claims"),
("LLM feedback", len(context["llm_feedback"]), "appended to the source trace", "model-generated observations"),
("Generated eval definitions", len(context["eval_suite"]), "appended to the source trace", "expected behavior, rubric, pass/fail examples"),
("Promptfoo row results", len(context["gate_result"]["rows"]), "appended to the source trace", "pass/fail outcome and explanation"),
("Promptfoo gate summary", 1, "global summary span", "suite totals across all evals"),
]
lines = [
"### HALO input summary",
"",
"| Input signal | Count | Where it lives | What is included |",
"| --- | ---: | --- | --- |",
]
lines.extend(f"| {name} | {count} | {location} | {included} |" for name, count, location, included in rows)
return "\n".join(lines)将反馈、生成的评估和评估结果附加到追踪上
编写 HALO 将要检查的组合追踪文件。人工反馈、LLM 反馈、生成的评估定义以及行级 Promptfoo 结果都被附加到匹配的运行时追踪上。整体门禁摘要保持全局，因为它描述了整个套件。
def write_halo_optimization_context(context: dict[str, Any]) -> Path:
context_path = ARTIFACT_DIR / "halo_optimization_context.jsonl"
lines = HALO_TRACE_PATH.read_text(encoding="utf-8").splitlines() if HALO_TRACE_PATH.exists() else []
lines.append(json.dumps(synthetic_span(
trace_id=synthetic_trace_id("current-harness-config"),
span_id=synthetic_span_id("current-harness-config"),
name="harness.config",
observation_kind="HARNESS_CONFIG",
attributes={
"harness.version": context["agent_config"]["version"],
"harness.system_prompt": context["agent_config"]["system_prompt"],
"harness.model_settings": json.dumps(context["agent_config"]["model_settings"]),
"harness.tool_policy": json.dumps(context["agent_config"]["tool_policy"]),
"harness.eval_metadata": json.dumps(context["agent_config"]["eval_metadata"]),
"optimizer.signal_source": "harness_config",
},
)))
for index, item in enumerate(context["human_feedback"]):
lines.append(json.dumps(synthetic_span(
trace_id=item["trace_id"],
span_id=synthetic_span_id(f"human-feedback-{index}"),
name="human_feedback.comment",
observation_kind="HUMAN_FEEDBACK",
attributes={
"feedback.id": item["feedback_id"],
"feedback.trace_id": item["trace_id"],
"feedback.trace_label": item["trace_label"],
"feedback.question": item["question"],
"feedback.summary": item["summary"],
"feedback.required_observations": json.dumps(item["required_observations"]),
"feedback.prohibited_claims": json.dumps(item["prohibited_claims"]),
"optimizer.signal_source": "human_feedback",
},
)))
for index, item in enumerate(context["llm_feedback"]):
lines.append(json.dumps(synthetic_span(
trace_id=item["trace_id"],
span_id=synthetic_span_id(f"llm-insight-{index}"),
name="llm_feedback.insight",
observation_kind="LLM_FEEDBACK",
attributes={
"llm_feedback.id": item["insight_id"],
"llm_feedback.trace_id": item["trace_id"],
"llm_feedback.trace_label": item["trace_label"],
"llm_feedback.question": item["question"],
"llm_feedback.observations": json.dumps(item["observations"]),
"optimizer.signal_source": "llm_feedback",
},
)))
for index, item in enumerate(context["eval_suite"]):
lines.append(json.dumps(synthetic_span(
trace_id=item["source_trace_id"],
span_id=synthetic_span_id(f"generated-eval-{index}"),
name="generated_eval.definition",
observation_kind="EVAL",
attributes={
"eval.id": item["eval_id"],
"eval.trace_id": item["source_trace_id"],
"eval.trace_label": item["source_trace_label"],
"eval.title": item["title"],
"eval.method": item["scoring_method"],
"eval.expected_behavior": item["expected_behavior"],
"eval.pass_example": item["suggested_pass_example"],
"eval.fail_example": item["suggested_fail_example"],
"optimizer.signal_source": "generated_eval",
},
)))
lines.append(json.dumps(synthetic_span(
trace_id=synthetic_trace_id("eval-gate-summary"),
span_id=synthetic_span_id("eval-gate-summary"),
name="eval_gate.summary",
observation_kind="EVAL_RESULT",
attributes={
"eval_gate.total": context["gate_result"]["total"],
"eval_gate.passed": context["gate_result"]["passed"],
"eval_gate.failed": context["gate_result"]["failed"],
"optimizer.signal_source": "eval_gate",
},
)))
for index, item in enumerate(context["gate_result"]["rows"]):
lines.append(json.dumps(synthetic_span(
trace_id=item["trace_id"],
span_id=synthetic_span_id(f"eval-gate-row-{index}"),
name="eval_gate.result",
observation_kind="EVAL_RESULT",
attributes={
"eval.id": item["eval_id"],
"eval.title": item["title"],
"eval.trace_id": item["trace_id"],
"eval.trace_label": item["trace_label"],
"eval.passed": item["passed"],
"eval.explanation": item["explanation"],
"optimizer.signal_source": "eval_gate",
},
)))
context_path.write_text("\n".join(lines).rstrip() + "\n", encoding="utf-8")
return context_path定义 HALO 输出提示词
此提示词告诉 HALO 要生成什么样类型的报告，包括 Codex 在最终移交文件中应收到的部分。你可以对其进行自定义，以匹配你公司的工作流程、评审流程或用例。
def render_halo_prompt() -> str:
return """
Analyze the financial diligence analyst optimization context as the central source of truth.
The JSONL contains the current harness configuration, agent execution traces, human feedback, LLM insights, generated eval definitions, and eval-gate results.
Treat human feedback as first-class evidence.
Before recommending a change, compare the evidence against the current harness config and distinguish:
- a requirement that is missing from the harness,
- a requirement already present but not reliably followed in execution, and
- an implementation or observability defect.
Write an implementation-first Codex handoff in this exact top-level order:
1. `## Executive summary`
2. `## Top 3 changes to implement first`
3. `## Ranked recommendation table`
4. `## Supporting diagnosis and evidence`
5. `## Detailed recommendations`
6. `## Insights by feedback source`
7. `## Machine-readable summary`
Section requirements:
- `## Executive summary`: briefly state what the current harness already does well, what the highest-value remaining gaps are, and whether the current eval gate passed.
- `## Top 3 changes to implement first`: list the three most valuable implementation moves with concise rationale.
- `## Ranked recommendation table`: include rank, recommendation, impact, confidence, implementation effort, evidence, and validation.
- `## Supporting diagnosis and evidence`: include recurring harness-level failure modes, classify each against the current harness as missing requirement vs already-present-but-not-reliably-followed vs implementation/observability defect, and state the evidence source for each.
- `## Detailed recommendations`: use these exact subsection headings in this order and do not use the word "owner" in them:
- `### Behavior contract`
- `#### Prompt`
- `#### Skills`
- `### Runtime implementation`
- `#### Tools`
- `#### Control flow`
- `#### Routing`
- `### Output contract`
- `#### Artifact schema`
- `### Observability and evals`
- `#### Observability`
- `#### Evals`
- `## Insights by feedback source`: summarize what came from traces, human feedback, LLM feedback, generated evals, eval-gate results, and harness config.
- `## Machine-readable summary`: include one fenced JSON block with `top_priorities`.
Do not add extra top-level sections outside that order.
""".strip()运行 HALO 并格式化报告
HALO 接收五次 SDK 执行追踪加上两个合成的全局追踪：一个记录当前的框架配置，另一个记录 Promptfoo 门禁摘要。这就是为什么其追踪计数比之前创建的五次智能体运行要多。
生成完整的优化报告，保存移交制品，并在笔记本中显示最高优先级的建议。
async def run_halo_optimization(context_path: Path) -> str:
from agents import set_trace_processors
from engine.agents.agent_config import AgentConfig as HaloAgentConfig
from engine.engine_config import EngineConfig
from engine.main import stream_engine_async
from engine.sandbox.sandbox import Sandbox
from engine.model_config import ModelConfig
from engine.models.engine_output import AgentOutputItem, AgentTextDelta
from engine.models.messages import AgentMessage
# HALO's current CLI wrapper sets compaction temperature to 0.0, which is not
# accepted by GPT-5-class models. Use the Python API so the compactor uses the
# model default-compatible temperature while preserving the requested model.
agent = HaloAgentConfig(
name="root",
model=ModelConfig(name=HALO_MODEL),
maximum_turns=20,
)
config = EngineConfig(
root_agent=agent,
subagent=agent.model_copy(update={"name": "sub"}),
synthesis_model=ModelConfig(name=HALO_MODEL),
compaction_model=ModelConfig(name=HALO_MODEL, temperature=1.0),
maximum_depth=1,
maximum_parallel_subagents=2,
)
# The notebook already exports the SDK traces locally; HALO does not need
# hosted trace ingestion for this diagnosis pass.
set_trace_processors([])
deltas: list[str] = []
final_items: list[str] = []
messages = [AgentMessage(role="user", content=render_halo_prompt())]
# This pass only needs HALO's trace-analysis tools. Skip the optional
# `run_code` sandbox so readers do not need a separate Deno/Pyodide setup
# just to generate the optimization report.
async def report_progress(done: asyncio.Event, interval_seconds: int = 30) -> None:
started = time.perf_counter()
print("HALO optimization started. This is usually the longest cell in the notebook.")
while not done.is_set():
try:
await asyncio.wait_for(done.wait(), timeout=interval_seconds)
except TimeoutError:
print(f"HALO still running... {format_duration(time.perf_counter() - started)} elapsed")
original_sandbox_get = Sandbox.__dict__["get"]
Sandbox.get = classmethod(lambda cls: None)
halo_started = time.perf_counter()
progress_done = asyncio.Event()
progress_task = asyncio.create_task(report_progress(progress_done))
try:
async for event in stream_engine_async(messages, config, context_path):
if isinstance(event, AgentTextDelta):
deltas.append(event.text_delta)
elif isinstance(event, AgentOutputItem) and event.final:
final_items.append(str(event.item))
finally:
progress_done.set()
await progress_task
Sandbox.get = original_sandbox_get
print(f"HALO optimization completed in {format_duration(time.perf_counter() - halo_started)}")
report = "".join(deltas).strip() or "\n\n".join(final_items).strip()
if not report:
raise RuntimeError("HALO completed without producing a report.")
return report
def clean_halo_handoff(report: str) -> str:
"""Keep only the final Codex-facing handoff sections from HALO output."""
normalized = re.sub(r"(?<!\n)(## Executive summary)", r"\n\n\1", report).strip()
start = normalized.rfind("## Executive summary")
if start == -1:
raise ValueError("HALO output did not include the expected executive summary section.")
handoff = normalized[start:].strip()
required_headings = [
"## Executive summary",
"## Top 3 changes to implement first",
"## Ranked recommendation table",
"## Supporting diagnosis and evidence",
"## Detailed recommendations",
"## Insights by feedback source",
"## Machine-readable summary",
]
missing = [heading for heading in required_headings if heading not in handoff]
if missing:
raise ValueError(f"HALO handoff is missing required sections: {missing}")
return handoff
def write_halo_handoff(report: str, path: str | Path) -> Path:
target = Path(path)
if not target.is_absolute():
target = PROJECT_ROOT / target
target.parent.mkdir(parents=True, exist_ok=True)
target.write_text(report.rstrip() + "\n", encoding="utf-8")
return target
halo_context = build_halo_context(traces, human_feedback, llm_feedback, eval_suite, gate_result, agent_config)
display(Markdown(halo_input_summary(halo_context)))
halo_context_path = write_halo_optimization_context(halo_context)
halo_report = await run_halo_optimization(halo_context_path)
clean_handoff = clean_halo_handoff(halo_report)
handoff_path = write_halo_handoff(clean_handoff, ARTIFACT_DIR / "codex_handoff.md")
def extract_named_section(report: str, heading: str) -> str:
if heading not in report:
return ""
start = report.index(heading)
remainder = report[start + len(heading):]
next_section = re.search(r"\n## ", remainder)
return report[start:] if next_section is None else report[start:start + len(heading) + next_section.start()]
def render_notebook_halo_summary(report: str) -> str:
sections = [
extract_named_section(report, "## Top 3 changes to implement first"),
extract_named_section(report, "## Insights by feedback source"),
]
rendered = "\n\n".join(section.strip() for section in sections if section.strip())
return rendered or report
print("Gate result passed into optimization context:", "gate_result" in halo_context)
print("Wrote:")
print("-", halo_context_path.relative_to(PROJECT_ROOT))
print("-", handoff_path.relative_to(PROJECT_ROOT))HALO 输入摘要
| 输入信号 | 计数 | 位置 | 包含内容 |
|---|---|---|---|
| 当前框架配置 | 1 | 全局配置跨度 | 系统提示词、模型设置、工具策略、评估元数据 |
| SDK 执行追踪 | 5 | 原始运行时追踪 | 智能体步骤、工具调用、输出 |
| 人工反馈 | 5 | 附加到源追踪上 | 评审员总结、必需的观察结果、禁止的结论 |
| LLM 反馈 | 5 | 附加到源追踪上 | 模型生成的观察结果 |
| 生成的评估定义 | 5 | 附加到源追踪上 | 预期行为、准则、通过/失败示例 |
| Promptfoo 行结果 | 5 | 附加到源追踪上 | 通过/失败结果和解释 |
| Promptfoo 门禁摘要 | 1 | 全局摘要跨度 | 跨所有评估的套件总计 |
HALO optimization started. This is usually the longest cell in the notebook.
HALO still running... 30s elapsed
HALO still running... 1m 00s elapsed
HALO still running... 1m 30s elapsed
HALO still running... 2m 00s elapsed
HALO still running... 2m 30s elapsed
HALO still running... 3m 00s elapsed
HALO still running... 3m 30s elapsed
HALO still running... 4m 00s elapsed
HALO still running... 4m 30s elapsed
HALO still running... 5m 00s elapsed
HALO still running... 5m 30s elapsed
HALO still running... 6m 00s elapsed
HALO still running... 6m 30s elapsed
HALO still running... 7m 00s elapsed
HALO optimization completed in 7m 15s
Gate result passed into optimization context: True
Wrote:
- examples/agents_sdk/agent_improvement_loop_artifacts/halo_optimization_context.jsonl
- examples/agents_sdk/agent_improvement_loop_artifacts/codex_handoff.md第 8 步：将完整报告交给 Codex
HALO 进行诊断和优先级排序。编码智能体或人类负责更改框架。
下方是 Codex 可以据此行动的完整报告快照：前三大建议加上来自每个反馈来源的紧凑总结。完整的 codex_handoff.md 文件还包括排名后的变更、支持证据和实施验证指南。
handoff_file = ARTIFACT_DIR / "codex_handoff.md"
if handoff_file.exists():
print(f"Full Codex handoff written to: {handoff_file.relative_to(PROJECT_ROOT)}")
print("Snapshot below; open the generated codex_handoff.md file to review the full handoff.")
display(Markdown(render_notebook_halo_summary(handoff_file.read_text(encoding="utf-8"))))
else:
print(f"Codex handoff not found yet: {handoff_file.relative_to(PROJECT_ROOT)}")
print("Run the HALO optimization cell above to generate it.")Full Codex handoff written to: examples/agents_sdk/agent_improvement_loop_artifacts/codex_handoff.md
Snapshot below; open the generated codex_handoff.md file to review the full handoff.优先实施的前 3 项变更
-
添加确定性的尽职调查事实账本和领域检查清单层。
编码 ARR、运营时间/现金消耗、母账户集中度、不支持的指标和 SOC 2 状态的规范事实和单一事实来源规则，以便智能体不能仅依赖通用的引用指令。 -
升级验证器以审计实际的输出制品，而不仅仅是声称的证据覆盖率。
当前的验证可能通过，但制品级的引用或结论审计问题仍需后续修复。解析生成的 Markdown/JSON/CSV 制品，提取材料结论，验证来源支持，并对不支持或未经审计的结论发出失败信号。 -
将五项生成的评估持久化到已签入的回归套件中。
生成的评估全部通过，但它们应成为耐用的回归测试，以便未来的提示词/运行时变更不会在具体的人工反馈问题上发生回归。
各反馈来源的洞察
| 反馈来源 | 关键洞察 |
|---|---|
| 追踪 | 智能体通常遵循制品生成工作流程和验证循环，但执行过程是通用的，有时甚至是整体式的。验证通过后仍会出现部分修复，表明验证不够严格。母账户集中度追踪展示了一种值得推广的确定性计算模式。 |
| 人工反馈 | 人工反馈是领域差距最强的证据：运营时间必须为 11 个月并附带融资压力；ARR 必须使用财务管控的单一事实来源；集中度必须汇总到母账户；SOC 2 Type I 和 Type II 不得混淆；官方 NRR 和 CAC 回收期在不支持时必须拒绝。 |
| LLM 反馈 | LLM 洞察强化了人工反馈的主题：ARR 头条数字需要警示，不支持的指标不应被提升，保留和管道结论需要来源警示，SOC 2 Type II 完成情况不得夸大。 |
| 生成的评估 | 根据反馈主题生成了五项针对性评估：运营时间/现金消耗、ARR 事实来源、客户集中度母账户汇总、SOC 2 精确度以及不支持的指标拒绝。这些编码了正确的回归面，应予以签入。 |
| 评估门禁结果 | 当前的评估门禁通过：总共 5 项，5 项通过，0 项失败。这表明最新的生成评估套件得到了满足，但应持久化并扩展该套件以覆盖验证器、制品解析和计算正确性。 |
| 框架配置 | 该框架已经具备了强大的通用证据、引用、制品和验证要求。其主要弱点是缺乏显式的财务尽职调查不变量，以及针对反馈浮现的具体错误缺乏确定性的运行时检查。 |
第 9 步：闭环
既然完整的工作流程已经到位，我们可以回顾笔记本顶部的优化飞轮。相同的架构支持两种操作模式。
它可以作为闭环运行，其中新的追踪、人工和模型反馈、生成的 Promptfoo 评估、HALO 诊断、Codex 实施、验证和部署全部反馈到下一个周期。在此模式下，移交制品可以写入共享存储，具有心跳检测的 Codex 自动化可以保持检查是否有新的移交文件，当出现新文件时唤醒并自动触发下一次实施阶段。
开发者还可以在他们想要的任何地方添加人工门禁，包括追踪评审、评估优化、合并请求批准、合并和部署。
设计选择在于人类在给出反馈后参与的程度。人类判断可以引导智能体执行任务的循环，或者人类可以在整个过程中保留批准门禁。在两个版本中，人工反馈都处于核心地位，因为它决定了系统学习什么以及接下来要改变什么。
结论
智能体改进循环提供了一条持续改进的路径，而不会将问题仅局限于提示词调试。整个循环很重要：追踪捕捉行为，人工反馈增加判断，评估保留系统应执行的操作，HALO 将证据转化为排名后的框架变更，Codex 可以实施下一阶段。
该领域仍在发展，一些独立组件可能会随时间变化。循环工程的更大理念是耐用的部分：当反馈、测试和实施在同一个循环中连接时，智能体可以从真实行为中改进。
后续步骤
- 通过编辑笔记本顶部的
AGENT_MODEL、ANALYSIS_MODEL、EVAL_GENERATION_MODEL、JUDGE_MODEL和HALO_MODEL，为循环的每个阶段选择模型。 - 创建你自己的追踪来测试智能体。
- 决定最终路径中有多少应该保持人工审查与自动化：你可以在开发者审查的 PR 处停止，或者将移交方案连接到一个能自动打开、合并和部署变更的系统中。
- 将
ARTIFACT_DIR下生成的codex_handoff.md文件传递给 Codex，检查它提议的框架变更，并针对更新后的框架重新运行相同的评估套件。