OCIレスポンスAPI圧縮チュートリアル

python3 -m venv .venv
source .venv/bin/activate
python -m pip install --upgrade 'openai>=2.45.0'

export OCI_BASE_URL='https://inference.generativeai.<REGION>.oci.oraclecloud.com/openai/v1'
export OCI_MODEL='<MODEL_ID>'
export OCI_OPENAI_PROJECT='<PROJECT_OCID>'

"""Compact a conversation with the OCI Responses API."""

import os
from getpass import getpass

from openai import OpenAI


def main() -> None:
    model = os.environ["OCI_MODEL"]
    api_key = os.environ.get("OCI_GENAI_API_KEY") or getpass(
        "OCI Generative AI API key: "
    )
    client = OpenAI(
        api_key=api_key,
        base_url=os.environ["OCI_BASE_URL"],
        project=os.environ["OCI_OPENAI_PROJECT"],
    )
    history = [{
        "role": "user",
        "content": (
            "Remember our deployment configuration: Frankfurt region, three "
            "application instances, PostgreSQL, daily backups with 30-day "
            "retention, and maintenance on Sunday at 02:00 UTC. "
            "Summarize the configuration."
        ),
    }]

    response = client.responses.create(model=model, input=history, store=False)
    print(response.output_text)
    history.extend(
        item.model_dump(mode="json", exclude_none=True)
        for item in response.output
    )

    compacted = client.responses.compact(model=model, input=history)
    print("Compaction observed:", any(
        item.type == "compaction" for item in compacted.output
    ))

    # Keep the complete compacted output, including any retained messages.
    history = [
        item.model_dump(mode="json", exclude_none=True)
        for item in compacted.output
    ]
    history.append({
        "role": "user",
        "content": "What are our backup retention and maintenance window?",
    })
    response = client.responses.create(model=model, input=history, store=False)
    print(response.output_text)
    client.close()


if __name__ == "__main__":
    main()

python standalone_compaction.py

"""Use automatic compaction with the OCI Responses API."""

import os
from getpass import getpass

from openai import OpenAI


def main() -> None:
    model = os.environ["OCI_MODEL"]
    api_key = os.environ.get("OCI_GENAI_API_KEY") or getpass(
        "OCI Generative AI API key: "
    )
    client = OpenAI(
        api_key=api_key,
        base_url=os.environ["OCI_BASE_URL"],
        project=os.environ["OCI_OPENAI_PROJECT"],
    )
    context_management = [{"type": "compaction", "compact_threshold": 1000}]
    history = [{
        "role": "user",
        "content": (
            "Remember our deployment configuration: Frankfurt region, three "
            "application instances, PostgreSQL, daily backups with 30-day "
            "retention, and maintenance on Sunday at 02:00 UTC."
        ),
    }]

    # Build synthetic history locally. These exchanges don't call the API.
    for day in range(1, 41):
        history.extend([
            {
                "role": "user",
                "content": (
                    f"Day {day}: health checks passed, database connections "
                    "are stable, and the daily backup completed. "
                    "Keep the deployment configuration unchanged."
                ),
            },
            {
                "role": "assistant",
                "content": [{
                    "type": "output_text",
                    "text": (
                        f"Day {day} status recorded. "
                        "The deployment configuration remains unchanged."
                    ),
                }],
            },
        ])
    history.append({
        "role": "user",
        "content": "Summarize our deployment configuration.",
    })

    response = client.responses.create(
        model=model,
        input=history,
        store=False,
        extra_body={"context_management": context_management},
    )
    print(response.output_text)
    output = [
        item.model_dump(mode="json", exclude_none=True)
        for item in response.output
    ]
    boundaries = [
        index for index, item in enumerate(output)
        if item["type"] == "compaction"
    ]
    print("Compaction observed:", bool(boundaries))

    # Keep the latest compaction item and everything after it.
    # Without a new compaction item, retain history and append all output.
    history = output[boundaries[-1]:] if boundaries else history + output
    history.append({
        "role": "user",
        "content": "What are our backup retention and maintenance window?",
    })
    response = client.responses.create(
        model=model,
        input=history,
        store=False,
        extra_body={"context_management": context_management},
    )
    print(response.output_text)
    client.close()


if __name__ == "__main__":
    main()

python server_compaction.py

export OCI_BASE_URL='https://inference.generativeai.<REGION>.oci.oraclecloud.com/openai/v1'
export OCI_MODEL='<INFERENCE_ENDPOINT_OCID>'
export OCI_OPENAI_PROJECT='<PROJECT_OCID>'

python standalone_compaction.py

python server_compaction.py