| 1 | from helpers.api import ApiHandler, Input, Output, Request |
| 2 | from plugins._context_window.helpers.usage import ( |
| 3 | latest_provider_usage, |
| 4 | usage_snapshot, |
| 5 | ) |
| 6 | from plugins._model_config.helpers.model_config import get_chat_model_config |
| 7 | |
| 8 | |
| 9 | class ContextWindow(ApiHandler): |
| 10 | async def process(self, input: Input, request: Request) -> Output: |
| 11 | context = self.use_context(str(input.get("context") or "")) |
| 12 | agent = context.streaming_agent or context.agent0 |
| 13 | window = agent.get_data(agent.DATA_NAME_CTX_WINDOW) |
| 14 | window = window if isinstance(window, dict) else {} |
| 15 | config = get_chat_model_config(agent) |
| 16 | |
| 17 | return { |
| 18 | "tokens": max(int(window.get("tokens") or 0), 0), |
| 19 | "context_window": max(int(config.get("ctx_length") or 0), 0), |
| 20 | "usage": usage_snapshot(window.get("usage")), |
| 21 | "provider_usage": latest_provider_usage(agent), |
| 22 | } |