mirror of
https://github.com/microsoft/agent-framework.git
synced 2026-06-16 21:04:09 +08:00
westey
bad05a2bdc
* Add initial harness console for python * Add textual to project * Add planning and approval flows with list selector * Address PR comments * Fix list selection bug * Fix PR #6312 round 2 review comments - Escape untrusted agent text with rich.markup.escape() in observers (text_output, planning_output, reasoning_display) to prevent markup injection - Remove non-functional 'Always approve' choices from tool_approval.py (framework lacks CreateAlwaysApproveToolResponse support) - Remove textual from root pyproject.toml dev deps (sample-specific) - Add PEP 723 inline script metadata to harness_research.py - Narrow except Exception to except NoMatches in list_selection.py Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Fix build error * Fix build errors --------- Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
57 lines
2.0 KiB
Python
57 lines
2.0 KiB
Python
# Copyright (c) Microsoft. All rights reserved.
|
|
|
|
"""Usage display observer for token usage statistics."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
from .base import ConsoleObserver
|
|
|
|
if TYPE_CHECKING:
|
|
from agent_framework import Agent
|
|
|
|
from ..state_driver import IUXStateDriver
|
|
|
|
|
|
class UsageDisplayObserver(ConsoleObserver):
|
|
"""Displays token usage as a proportion of the context window.
|
|
|
|
Shows current token usage as reported by the API immediately when
|
|
usage information becomes available (via Content items or the final response).
|
|
The display shows input/output/total relative to configured budgets.
|
|
"""
|
|
|
|
async def on_content(
|
|
self,
|
|
ux: IUXStateDriver,
|
|
content: Any,
|
|
agent: Agent,
|
|
session: Any,
|
|
) -> None:
|
|
"""Update usage display immediately when usage content arrives.
|
|
|
|
Args:
|
|
ux: The UX state driver for UI updates.
|
|
content: A content item from the response.
|
|
agent: The AI agent.
|
|
session: The agent session.
|
|
"""
|
|
if getattr(content, "type", None) == "usage":
|
|
usage_details = getattr(content, "usage_details", None)
|
|
if isinstance(usage_details, dict):
|
|
# Pass through to state driver — the runner handles formatting
|
|
ux.set_usage_text(self._format_from_details(usage_details))
|
|
|
|
@staticmethod
|
|
def _format_from_details(usage: dict) -> str:
|
|
"""Format usage details dict into display text.
|
|
|
|
This is a fallback formatter for when usage arrives as Content
|
|
before the runner's final response processing.
|
|
"""
|
|
input_tokens = usage.get("input_token_count", 0) or 0
|
|
output_tokens = usage.get("output_token_count", 0) or 0
|
|
total_tokens = usage.get("total_token_count", 0) or input_tokens + output_tokens
|
|
return f"📊 Tokens — input: {input_tokens:,} | output: {output_tokens:,} | total: {total_tokens:,}"
|