File size: 2,471 Bytes
5f25733 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 | import sys
from pathlib import Path
_EXAMPLES_ROOT = Path(__file__).resolve().parents[1]
if str(_EXAMPLES_ROOT) not in sys.path:
sys.path.insert(0, str(_EXAMPLES_ROOT))
from _bootstrap import ensure_local_sdk_src, runtime_config
ensure_local_sdk_src()
from openai_codex import (
Codex,
)
from openai_codex.types import (
ThreadTokenUsageUpdatedNotification,
TurnCompletedNotification,
)
print("Codex mini CLI. Type /exit to quit.")
def _format_usage(usage: object) -> str:
last = usage.last
total = usage.total
return (
"usage>\n"
f" last: input={last.input_tokens} output={last.output_tokens} reasoning={last.reasoning_output_tokens} total={last.total_tokens} cached={last.cached_input_tokens}\n"
f" total: input={total.input_tokens} output={total.output_tokens} reasoning={total.reasoning_output_tokens} total={total.total_tokens} cached={total.cached_input_tokens}"
)
with Codex(config=runtime_config()) as codex:
thread = codex.thread_start(model="gpt-5.4", config={"model_reasoning_effort": "high"})
print("Thread:", thread.id)
while True:
try:
user_input = input("you> ").strip()
except EOFError:
break
if not user_input:
continue
if user_input in {"/exit", "/quit"}:
break
turn = thread.turn(user_input)
usage = None
status = None
error = None
print("assistant> ", end="", flush=True)
for event in turn.stream():
payload = event.payload
if event.method == "item/agentMessage/delta":
delta = payload.delta
if delta:
print(delta, end="", flush=True)
continue
if isinstance(payload, ThreadTokenUsageUpdatedNotification):
usage = payload.token_usage
continue
if isinstance(payload, TurnCompletedNotification):
status = payload.turn.status
error = payload.turn.error
print()
if status is None:
raise RuntimeError("stream ended without turn/completed")
if usage is None:
raise RuntimeError("stream ended without token usage")
status_text = status.value
print(f"assistant.status> {status_text}")
if status_text == "failed":
print("assistant.error>", error)
print(_format_usage(usage))
|