レシピ¶

これらのレシピは、v0.1系列で実装されている動作だけを使用します。マニフェストを再現できるよう、すべてのレコードに安定したID、タイムスタンプ、出所を指定しています。

取得した文書から予算内で1件を選ぶ¶

import cognoxium as cx

frame = cx.CognitionFrame.from_records([
    {"id": "refund", "payload": "refund policy is 30 days", "sources": ["db://kb/refund"],
     "created_at": "2026-01-01T00:00:00Z"},
    {"id": "office", "payload": "office address is Tokyo", "sources": ["db://kb/office"],
     "created_at": "2026-01-01T00:01:00Z"},
])

pack = frame.rank("refund policy").pack(
    budget=8,
    token_profile=cx.profiles.approximate(),
    boundary=cx.Boundary.external("provider"),
)
print([item.id for item in pack.items])
print([(item["id"], item["reason"]) for item in pack.manifest.excluded])

['refund']
[('office', 'budget')]

組み込みランカーは字句一致を利用します。埋め込みまたはモデルベースのスコアを使う場合は、決定論的でバッチ対応のRankerを渡してください。Cognoxium自体はネットワークリクエストを行いません。

失敗したツール結果を隔離する¶

import cognoxium as cx

frame = cx.CognitionFrame.from_records([
    {"id": "ok", "payload": "tests passed", "role": "tool", "sources": ["tool://ci/1"],
     "metadata": {"exit_code": 0}, "created_at": "2026-01-01T00:00:00Z"},
    {"id": "failed", "payload": "unverified output", "role": "tool",
     "sources": ["tool://ci/2"], "metadata": {"exit_code": 1},
     "created_at": "2026-01-01T00:01:00Z"},
])

pack = frame.quarantine(
    lambda row: row["metadata"].get("exit_code") != 0
).pack(budget=100, boundary=cx.Boundary.external("provider"))
print([item.id for item in pack.items])
print(pack.manifest.excluded[0]["reason"])

['ok']
quarantined

隔離はアプリケーションの判断を記録する操作です。ツール結果が正しいかをCognoxiumが判定するわけではありません。

古い会話ターンを明示的に降格する¶

import cognoxium as cx

frame = cx.CognitionFrame.from_records([
    {"id": "policy", "payload": "Follow policy", "sources": ["app://policy"],
     "retention": "required", "trust": "trusted", "priority": 10,
     "created_at": "2026-01-01T00:00:00Z"},
    {"id": "old-turn", "payload": "Earlier discussion", "sources": ["chat://turn/1"],
     "retention": "required", "created_at": "2026-01-01T00:01:00Z"},
])

authorized = frame.demote(
    cx.col("id") == "old-turn",
    to=cx.Retention.OPTIONAL,
)
print([(item.id, item.retention.value) for item in authorized.collect()])

[('policy', 'required'), ('old-turn', 'optional')]

アプリケーションはdemote()を呼び出す前に承認を判断します。パッキングが必須コンテンツを自動的に降格することはありません。

監査可能なフレームを保存する¶

レビューしやすいfixtureにはwrite_jsonl()、高速なローカル交換にはwrite_ipc()、分析用ストレージにはwrite_parquet()を使用します。Arrow形式にはarrow extraが必要です。

frame.write_jsonl("context.jsonl")
restored = cx.CognitionFrame.read_jsonl("context.jsonl")
assert restored.collect() == frame.collect()

監査要件で許可される場合は、pack.manifest.to_dict()を別途保存してください。マニフェスト自体にも、機密性のある識別子、出所URI、競合値が含まれる場合があります。

予算超過からの復旧¶

BudgetExceededを捕捉し、構造化フィールドをログへ記録して、アプリケーション側で復旧方法を判断します。翻訳済みの例外文で分岐してはいけません。実行可能な復旧例は予算超過からの復旧を参照してください。