Cost Tracking

run_with_response() and arun_with_response()

Instead of returning a plain string, these methods return a RunResult that includes the text reply, why the run stopped, every tool call made, and token/cost totals summed over every LiteLLM completion in the run.

from cyclops import Agent, AgentConfig

config = AgentConfig(model="gpt-4o-mini")
agent = Agent(config)

result = agent.run_with_response("Summarise the theory of relativity in two sentences.")

print(result.content)
print(f"Model      : {result.model}")
print(f"Prompt     : {result.prompt_tokens} tokens")
print(f"Completion : {result.completion_tokens} tokens")
print(f"Total      : {result.tokens_used} tokens")
print(f"Cost       : ${result.cost:.6f}")

The async version works identically:

import asyncio

async def main():
    result = await agent.arun_with_response("Explain neural networks briefly.")
    print(result.content)
    print(f"Cost: ${result.cost:.6f}")

asyncio.run(main())

RunResult fields

Field	Type	Description
content	str	The final answer. Empty unless the run completed.
tool_calls	List[ToolCall]	Every tool call made during the run, in order.
denied_tool_calls	List[DeniedToolCall]	Tool calls that middleware denied, each with id (None in naive tool mode), name, arguments and the reason the model saw.
model	str	The model identifier used (from AgentConfig.model).
run_id	str	Unique id of the run, also given to tool middleware and guards.
stop_reason	str	"completed" for a final answer, "max_tokens" when the output limit cut the last completion, "content_filter" when the provider filtered or refused it (output limit and filtering), "max_steps" or "budget" when the run reached a limit, "guard" when a guard still rejected the answer after max_guard_rounds or with no step or budget left for another round, "output_validation" when the answer still failed response_model validation after output_retries (retries), "cancelled" when a run handle was cancelled, "error" when it failed.
finish_reason	str | None	The last completion’s finish reason as LiteLLM normalizes it: "stop", "length", "tool_calls" or "content_filter". None if the run made none.
steps	int	LLM completions made in the run.
usage	Usage | None	Tokens and cost summed over every completion in the run, including those of subagents it called. None if any completion reported no usage.
usage_by_model	Dict[str, Usage | None]	usage split by the model, as configured, that made each completion: fallback models, subagents on other models and a compaction model each get their own entry. A model’s entry is None if one of its completions reported no usage.
duration	float	Wall-clock seconds the run took, tool calls and waits included.
error	Exception | None	The exception that ended the run with "error". Left out of model_dump().
tokens_used	int | None	Total tokens (prompt + completion), from usage.
prompt_tokens	int | None	Tokens in the prompt/context sent to the model, from usage.
completion_tokens	int | None	Tokens in the generated response, from usage.
cached_tokens	int | None	Prompt tokens the provider served from its cache, from usage.
cost	float | None	Estimated USD cost LiteLLM computed for each completion, with the pricing of the deployment that served it. None if LiteLLM can’t price a completion of the run.

tokens_used, prompt_tokens, completion_tokens, cached_tokens and cost are read-only properties over usage.

Usage fields

RunResult.usage, each entry of usage_by_model, StepFinished.usage and session.usage are Usage objects, and two of them add up with +. The token counts are those LiteLLM normalizes from each provider’s response.

Field	Type	Description
prompt_tokens	int	Input tokens, the cached and cache-written ones included.
completion_tokens	int	Output tokens, the reasoning ones included.
cached_tokens	int	Prompt tokens read from the provider’s cache.
cache_write_tokens	int	Prompt tokens written to the provider’s cache. Anthropic and Bedrock report them; OpenAI and Gemini cache without reporting writes.
reasoning_tokens	int	Completion tokens the model spent reasoning, for models that report them.
total_tokens	int	Prompt plus completion tokens.
cost	float | None	Estimated USD cost. None if LiteLLM can’t price a completion.
api_duration	float	Seconds from each completion’s request to its whole response, retries included. A paced stream (stream(), astream(), stream_events() or astream_events()) reads only a few chunks ahead of the caller, so the time the caller takes to read it counts too. Completions of subagents run in parallel add up, so this can exceed the run’s duration.

result = agent.run_with_response("Summarise the report.")
for model, usage in result.usage_by_model.items():
    print(f"{model}: {usage.total_tokens} tokens, ${usage.cost:.6f}")
print(f"{result.duration:.1f}s in all, {result.usage.api_duration:.1f}s in completions")

Each ToolCall in tool_calls has:

Field	Type	Description
id	str	Unique call identifier assigned by the LLM.
name	str	Name of the tool that was called.
arguments	dict	Arguments the model passed to the tool.
result	Any | None	The value returned by the tool.

Failed runs

run_with_response() and arun_with_response() don’t raise when the run fails. They return a RunResult with stop_reason="error", the exception in error, and the usage of the steps that finished, so a failed run’s cost is still known. run() and arun() raise the exception.

result = agent.run_with_response("Summarise this report.")
if result.stop_reason == "error":
    print(f"Failed after {result.steps} steps: {result.error}, cost ${result.cost}")

Logging costs over time

A common production pattern is to accumulate cost and token data across many agent runs:

import asyncio
from dataclasses import dataclass, field
from typing import List
from cyclops import Agent, AgentConfig
from cyclops import RunResult


@dataclass
class UsageTracker:
    total_cost: float = 0.0
    total_tokens: int = 0
    runs: List[RunResult] = field(default_factory=list)

    def record(self, response: RunResult) -> None:
        self.runs.append(response)
        self.total_cost += response.cost or 0.0
        self.total_tokens += response.tokens_used or 0

    def report(self) -> None:
        print(f"Runs       : {len(self.runs)}")
        print(f"Total cost : ${self.total_cost:.4f}")
        print(f"Total tokens: {self.total_tokens:,}")


tracker = UsageTracker()
config = AgentConfig(model="gpt-4o-mini")
agent = Agent(config)

questions = [
    "What is photosynthesis?",
    "Explain DNS in one sentence.",
    "What is the boiling point of water in Celsius?",
]

for q in questions:
    result = agent.run_with_response(q)
    tracker.record(result)
    print(f"Q: {q}")
    print(f"A: {result.content}")
    print(f"   cost={result.cost:.6f}  tokens={result.tokens_used}\n")

tracker.report()

Tool calls in the response

When tools are used, each tool invocation is recorded in result.tool_calls:

from cyclops.toolkit import tool


@tool
def lookup(symbol: str) -> str:
    """Look up a stock price."""
    return "189.50"


config = AgentConfig(model="gpt-4o-mini")
agent = Agent(config, tools=[lookup])

result = agent.run_with_response("What is the current price of AAPL?")

print(result.content)
for tc in result.tool_calls:
    print(f"  Called: {tc.name}({tc.arguments}) → {tc.result}")

Note

Cost tracking depends on LiteLLM’s internal pricing table. A router deployment’s own prices (input_cost_per_token, output_cost_per_token) or its model_info.base_model take precedence, as does a base_model in completion_kwargs. Ollama and other self-hosted models will have cost=None since no billing data is available. Token counts are still reported if the server returns usage information.

Full example

examples/cost_tracking.py

"""Cost and token tracking example: run_with_response() / arun_with_response()

Demonstrates:
  1. Inspecting RunResult metadata (tokens_used, cost, prompt_tokens,
     completion_tokens) from a single synchronous run
  2. Async variant with arun_with_response()
  3. Accumulating cost and tokens across multiple calls to track a session budget

Cost data is powered by LiteLLM's built-in pricing table.  The numbers are
accurate for cloud APIs (OpenAI, Anthropic, etc.) but will be None for
local/free models like Ollama since they have no billing cost.

Default model is gpt-4o-mini so the cost fields are populated with real values.
Swap MODEL below if you only have a local Ollama instance available.
"""

import asyncio

from cyclops import Agent, AgentConfig, RunResult, Session

# ---------------------------------------------------------------------------
# Model configuration
# ---------------------------------------------------------------------------

# gpt-4o-mini gives real cost data (requires OPENAI_API_KEY)
MODEL = "gpt-4o-mini"

# For a free/local alternative, swap to:
#   MODEL = "ollama/qwen3:4b"
# Note: cost will be None for Ollama: token counts still work.

# Other cloud alternatives with cost data:
#   MODEL = "claude-3-haiku-20240307"    # ANTHROPIC_API_KEY
#   MODEL = "groq/llama-3.1-8b-instant"  # GROQ_API_KEY (cost is ~$0 but tracked)


# ---------------------------------------------------------------------------
# Helper: pretty-print a RunResult
# ---------------------------------------------------------------------------


def print_response(label: str, response: RunResult) -> None:
    print(f"\n--- {label} ---")
    print(
        f"Content          : {response.content[:120]}{'...' if len(response.content) > 120 else ''}"
    )
    print(f"Model            : {response.model}")
    print(f"Prompt tokens    : {response.prompt_tokens}")
    print(f"Completion tokens: {response.completion_tokens}")
    print(f"Total tokens     : {response.tokens_used}")
    if response.cost is not None:
        print(f"Cost             : ${response.cost:.6f}")
    else:
        print("Cost             : N/A (local model or no pricing data)")
    if response.tool_calls:
        print(f"Tool calls       : {[tc.name for tc in response.tool_calls]}")


# ---------------------------------------------------------------------------
# 1. Single synchronous call
# ---------------------------------------------------------------------------


def demo_single_run() -> RunResult:
    print("=" * 60)
    print("1. Single run_with_response() call")
    print("=" * 60)

    config = AgentConfig(
        model=MODEL,
        system_prompt="You are a concise assistant. Keep answers brief.",
        temperature=0.3,
    )
    agent = Agent(config)

    response = agent.run_with_response("What is the difference between RAM and ROM?")
    print_response("run_with_response", response)
    return response


# ---------------------------------------------------------------------------
# 2. Single async call
# ---------------------------------------------------------------------------


async def demo_async_run() -> RunResult:
    print("\n" + "=" * 60)
    print("2. Single arun_with_response() call (async)")
    print("=" * 60)

    config = AgentConfig(
        model=MODEL,
        system_prompt="You are a concise assistant. Keep answers brief.",
        temperature=0.3,
    )
    agent = Agent(config)

    response = await agent.arun_with_response(
        "Explain TCP vs UDP in one sentence each."
    )
    print_response("arun_with_response", response)
    return response


# ---------------------------------------------------------------------------
# 3. Accumulate cost across multiple calls in a session
# ---------------------------------------------------------------------------


def demo_cost_accumulation() -> None:
    print("\n" + "=" * 60)
    print("3. Accumulating cost across multiple runs")
    print("=" * 60)

    config = AgentConfig(
        model=MODEL,
        system_prompt="You are a helpful coding assistant. Be concise.",
        temperature=0.2,
    )
    agent = Agent(config)
    session = Session()

    questions = [
        "What does the 'yield' keyword do in Python?",
        "When should I use a list comprehension vs a generator expression?",
        "Give me a one-line example of a generator expression.",
    ]

    total_prompt_tokens = 0
    total_completion_tokens = 0
    total_tokens = 0
    total_cost = 0.0
    cost_available = False

    print(
        f"\nAsking {len(questions)} questions to the same agent (one session, so its transcript grows):\n"
    )

    for i, question in enumerate(questions, start=1):
        response = agent.run_with_response(question, session=session)
        print_response(f"Q{i}: {question[:50]}...", response)

        # Accumulate: guard against None (local models)
        if response.prompt_tokens is not None:
            total_prompt_tokens += response.prompt_tokens
        if response.completion_tokens is not None:
            total_completion_tokens += response.completion_tokens
        if response.tokens_used is not None:
            total_tokens += response.tokens_used
        if response.cost is not None:
            total_cost += response.cost
            cost_available = True

    print("\n" + "-" * 40)
    print("SESSION TOTALS")
    print("-" * 40)
    print(f"Prompt tokens     : {total_prompt_tokens}")
    print(f"Completion tokens : {total_completion_tokens}")
    print(f"Total tokens      : {total_tokens}")
    if cost_available:
        print(f"Total cost        : ${total_cost:.6f}")
        print(f"Average per call  : ${total_cost / len(questions):.6f}")
    else:
        print("Total cost        : N/A (local model)")


# ---------------------------------------------------------------------------
# 4. Demonstrate accessing the raw RunResult fields directly
# ---------------------------------------------------------------------------


def demo_field_access() -> None:
    print("\n" + "=" * 60)
    print("4. Direct field access on RunResult")
    print("=" * 60)

    config = AgentConfig(model=MODEL, temperature=0.1)
    agent = Agent(config)

    response: RunResult = agent.run_with_response("Name three sorting algorithms.")

    # RunResult is a Pydantic model: all fields are typed attributes
    assert isinstance(response.content, str)
    assert response.model == MODEL

    # Token fields are Optional[int]: check before using
    if response.tokens_used is not None:
        print(f"Used {response.tokens_used} tokens total.")

    # Cost is Optional[float]: may be None for local/unknown models
    if response.cost is not None:
        budget = 0.01  # $0.01 example budget
        remaining = budget - response.cost
        print(
            f"Cost ${response.cost:.6f} of ${budget:.2f} budget. Remaining: ${remaining:.4f}"
        )
    else:
        print("Cost not available for this model.")

    print(f"\nFull response content:\n{response.content}")


# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------

if __name__ == "__main__":
    demo_single_run()
    asyncio.run(demo_async_run())
    demo_cost_accumulation()
    demo_field_access()