diff --git a/.env.template b/.env.template index e3be457d..a91ede16 100644 --- a/.env.template +++ b/.env.template @@ -140,3 +140,9 @@ LLM_ANTHROPIC_API_KEY=your-anthropic-api-key-here # SENTRY_ENVIRONMENT=development # SENTRY_TRACES_SAMPLE_RATE=0.1 # SENTRY_PROFILES_SAMPLE_RATE=0.1 + +# ============================================================================= +# Metrics (Optional) +# ============================================================================= +# ENABLED=false +# NAMESPACE=honcho diff --git a/config.toml.example b/config.toml.example index a26d4672..50f618ba 100644 --- a/config.toml.example +++ b/config.toml.example @@ -106,3 +106,8 @@ THINKING_BUDGET_TOKENS = 512 [webhook] SECRET = "" MAX_WORKSPACE_LIMIT = 10 + +# Metrics settings +[metrics] +ENABLED = false +NAMESPACE = "honcho" diff --git a/fly.toml b/fly.toml index 09c90275..26824fe3 100644 --- a/fly.toml +++ b/fly.toml @@ -28,3 +28,13 @@ kill_timeout = '5s' cpu_kind = 'shared' cpus = 1 processes = ['api', 'deriver'] + +[[metrics]] +port = 8000 +path = "/metrics" +processes = ["api"] + +[[metrics]] +port = 9090 +path = "/metrics" +processes = ["deriver"] diff --git a/pyproject.toml b/pyproject.toml index 329e5604..95475114 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -31,6 +31,7 @@ dependencies = [ "google-genai>=1.32.0", "pdfplumber>=0.11.7", "typing-extensions>=4.11.0", + "prometheus-client>=0.20.0", "json-repair>=0.49.0", ] [tool.uv] diff --git a/src/config.py b/src/config.py index d50ce340..6458717e 100644 --- a/src/config.py +++ b/src/config.py @@ -286,6 +286,13 @@ class WebhookSettings(HonchoSettings): MAX_WORKSPACE_LIMIT: int = 10 +class MetricsSettings(HonchoSettings): + model_config = SettingsConfigDict(env_prefix="METRICS_", extra="ignore") # pyright: ignore + + ENABLED: bool = False + NAMESPACE: str = "honcho" + + class DreamSettings(HonchoSettings): model_config = SettingsConfigDict(env_prefix="DREAM_", extra="ignore") # pyright: ignore @@ -337,6 +344,7 @@ class AppSettings(HonchoSettings): PEER_CARD: PeerCardSettings = Field(default_factory=PeerCardSettings) SUMMARY: SummarySettings = Field(default_factory=SummarySettings) WEBHOOK: WebhookSettings = Field(default_factory=WebhookSettings) + METRICS: MetricsSettings = Field(default_factory=MetricsSettings) DREAM: DreamSettings = Field(default_factory=DreamSettings) @field_validator("LOG_LEVEL") diff --git a/src/deriver/__main__.py b/src/deriver/__main__.py index ef171972..6b54d5a4 100644 --- a/src/deriver/__main__.py +++ b/src/deriver/__main__.py @@ -3,6 +3,7 @@ import logging import os import uvloop +from prometheus_client import start_http_server from src.config import settings @@ -43,6 +44,16 @@ def setup_logging(): logging.getLogger("groq._base_client").setLevel(logging.WARNING) +def start_metrics_server() -> None: + """Start Prometheus metrics HTTP server on port 9090.""" + try: + # Uses default REGISTRY from prometheus_client + start_http_server(9090, addr="0.0.0.0") # nosec B104 + print("[DERIVER] Starting Prometheus metrics server on port 9090") + except Exception as e: + print(f"[DERIVER] Failed to start Prometheus metrics server: {str(e)}") + + if __name__ == "__main__": print("[DERIVER] Starting deriver queue processor") @@ -52,6 +63,8 @@ if __name__ == "__main__": asyncio.set_event_loop_policy(uvloop.EventLoopPolicy()) try: print("[DERIVER] Running main loop") + if settings.METRICS.ENABLED: + start_metrics_server() asyncio.run(main()) except KeyboardInterrupt: print("[DERIVER] Shutdown initiated via KeyboardInterrupt") diff --git a/src/deriver/deriver.py b/src/deriver/deriver.py index 365586f2..4be32527 100644 --- a/src/deriver/deriver.py +++ b/src/deriver/deriver.py @@ -4,7 +4,7 @@ import time import sentry_sdk -from src import crud, exceptions +from src import crud, exceptions, prometheus from src.config import settings from src.crud.representation import RepresentationManager from src.dependencies import tracked_db @@ -43,6 +43,7 @@ async def critical_analysis_call( working_representation: Representation, history: str, new_turns: list[str], + estimated_input_tokens: int, ) -> PromptRepresentation: prompt = critical_analysis_prompt( peer_id=peer_id, @@ -68,6 +69,10 @@ async def critical_analysis_call( retry_attempts=3, ) + prometheus.DERIVER_TOKENS_PROCESSED.labels( + task_type="representation", + ).inc(response.output_tokens + estimated_input_tokens) + return response.content @@ -212,6 +217,8 @@ async def process_representation_tasks_batch( include_summary=True, ) + session_context_tokens = estimate_tokens(formatted_history) + # got working representation and peer card, log timing context_prep_duration = (time.perf_counter() - context_prep_start) * 1000 accumulate_metric( @@ -239,12 +246,12 @@ async def process_representation_tasks_batch( observed=observed, ) - # Create reasoner instance reasoner = CertaintyReasoner( representation_manager=representation_manager, ctx=messages, observed=observed, observer=observer, + estimated_input_tokens=estimated_input_tokens + session_context_tokens, ) # Run single-pass reasoning @@ -298,11 +305,13 @@ class CertaintyReasoner: *, observed: str, observer: str, + estimated_input_tokens: int, ) -> None: self.representation_manager = representation_manager self.ctx = ctx self.observed = observed self.observer = observer + self.estimated_input_tokens: int = estimated_input_tokens @conditional_observe @sentry_sdk.trace @@ -343,6 +352,7 @@ class CertaintyReasoner: working_representation=working_representation, history=history, new_turns=new_turns, + estimated_input_tokens=self.estimated_input_tokens, ) except Exception as e: raise exceptions.LLMError( diff --git a/src/deriver/queue_manager.py b/src/deriver/queue_manager.py index 02779b47..1b488b61 100644 --- a/src/deriver/queue_manager.py +++ b/src/deriver/queue_manager.py @@ -15,7 +15,7 @@ from sqlalchemy.dialects.postgresql import insert from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.sql import func -from src import models +from src import models, prometheus from src.config import settings from src.dependencies import tracked_db from src.deriver.consumer import ( @@ -518,7 +518,7 @@ class QueueManager: try: if work_unit.task_type in ["representation", "summary"]: logger.debug( - f"Publishing queue.empty event for {work_unit_key}" + f"Publishing queue.empty event for {work_unit_key} in workspace {work_unit.workspace_name}" ) await publish_webhook_event( QueueEmptyEvent( @@ -713,6 +713,7 @@ class QueueManager: if not items: return async with tracked_db("process_queue_item_batch") as db: + work_unit = parse_work_unit_key(work_unit_key) item_ids = [item.id for item in items] await db.execute( update(models.QueueItem) @@ -727,6 +728,12 @@ class QueueManager: ) await db.commit() + if work_unit.task_type in ["representation", "summary"]: + prometheus.DERIVER_QUEUE_ITEMS_PROCESSED.labels( + workspace_name=work_unit.workspace_name, + task_type=work_unit.task_type, + ).inc(len(items)) + async def mark_queue_item_as_errored( self, item: QueueItem, work_unit_key: str, error: str ) -> None: diff --git a/src/main.py b/src/main.py index a881ca21..c7f4c53d 100644 --- a/src/main.py +++ b/src/main.py @@ -11,6 +11,9 @@ from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse from fastapi_pagination import add_pagination +from src import prometheus +from src.utils.logging import get_route_template + if TYPE_CHECKING: from sentry_sdk._types import Event, Hint from sentry_sdk.integrations.fastapi import FastApiIntegration @@ -144,6 +147,7 @@ app.add_middleware( allow_headers=["*"], ) + add_pagination(app) app.include_router(workspaces.router, prefix="/v2") @@ -153,12 +157,23 @@ app.include_router(messages.router, prefix="/v2") app.include_router(keys.router, prefix="/v2") app.include_router(webhooks.router, prefix="/v2") +app.add_api_route("/metrics", prometheus.metrics, methods=["GET"]) + # Global exception handlers @app.exception_handler(HonchoException) -async def honcho_exception_handler(_request: Request, exc: HonchoException): +async def honcho_exception_handler(request: Request, exc: HonchoException): """Handle all Honcho-specific exceptions.""" logger.error(f"{exc.__class__.__name__}: {exc.detail}", exc_info=exc) + + if prometheus.METRICS_ENABLED and request.url.path != "/metrics": + template = get_route_template(request) + prometheus.API_REQUESTS.labels( + method=request.method, + endpoint=template, + status_code=str(exc.status_code), + ).inc() + return JSONResponse( status_code=exc.status_code, content={"detail": exc.detail}, @@ -166,9 +181,18 @@ async def honcho_exception_handler(_request: Request, exc: HonchoException): @app.exception_handler(Exception) -async def global_exception_handler(_request: Request, exc: Exception): +async def global_exception_handler(request: Request, exc: Exception): """Handle all unhandled exceptions.""" logger.error(f"Unhandled exception: {str(exc)}", exc_info=True) + + if prometheus.METRICS_ENABLED and request.url.path != "/metrics": + template = get_route_template(request) + prometheus.API_REQUESTS.labels( + method=request.method, + endpoint=template, + status_code="500", + ).inc() + if SENTRY_ENABLED: sentry_sdk.capture_exception(exc) return JSONResponse( @@ -190,6 +214,17 @@ async def track_request( token = request_context.set(f"api:{request_id}") try: - return await call_next(request) + response = await call_next(request) + + # Track Prometheus metrics if enabled + if prometheus.METRICS_ENABLED and request.url.path != "/metrics": + template = get_route_template(request) + prometheus.API_REQUESTS.labels( + method=request.method, + endpoint=template, + status_code=str(response.status_code), + ).inc() + + return response finally: request_context.reset(token) diff --git a/src/prometheus.py b/src/prometheus.py new file mode 100644 index 00000000..85bda9ec --- /dev/null +++ b/src/prometheus.py @@ -0,0 +1,119 @@ +""" +Prometheus metrics for Honcho. + +This module defines all Prometheus metrics for all Honcho processes and exposes them via the /metrics endpoint. +""" + +import logging + +from prometheus_client import ( + CONTENT_TYPE_LATEST, + REGISTRY, + Counter, + disable_created_metrics, + generate_latest, +) +from starlette.responses import Response as StarletteResponse + +from src.config import settings + +METRICS_ENABLED = settings.METRICS.ENABLED +disable_created_metrics() # Disables _created metrics on counters, histograms, and summaries + +logger = logging.getLogger(__name__) + + +class NamespacedCounter(Counter): + """Counter that automatically includes namespace label""" + + def labels(self, **kwargs: str) -> "NamespacedCounter": + """Override labels to automatically appends namespace label""" + kwargs["namespace"] = settings.METRICS.NAMESPACE + return super().labels(**kwargs) + + +# Tracks all requests to the Honcho API. +# +# Incremented in: src/main.py middleware for every request +# Labels: +# - method: HTTP method (GET, POST, PUT, DELETE, etc.) +# - endpoint: FastAPI route template (e.g., "/v2/workspaces/{workspace_id}/sessions/{session_id}/messages") +# - status_code: HTTP response status code (200, 404, 500, etc.) +API_REQUESTS = NamespacedCounter( + "api_requests_total", + "Total API requests", + [ + "namespace", + "method", + "endpoint", + "status_code", + ], +) + +# Tracks the total number of honcho messages created. +# +# Incremented in: src/routers/messages.py when messages are successfully created +# Labels: +# - workspace_name: The workspace where messages were created +MESSAGES_CREATED = NamespacedCounter( + "messages_created_total", + "Total messages created", + [ + "namespace", + "workspace_name", + ], +) + + +# Tracks the total number of dialectic calls made. +# +# Incremented in: src/routers/peers.py when successful dialectic calls are made +# Labels: +# - workspace_name: The workspace where the dialectic call was made +DIALECTIC_CALLS = NamespacedCounter( + "dialectic_calls_total", + "Total dialectic calls", + [ + "namespace", + "workspace_name", + ], +) + +# Tracks the total number of queue items successfully processed by the deriver. +# +# Incremented in: src/deriver/queue_manager.py when queue items are processed +# Labels: +# - workspace_name: The workspace where items were processed +# - task_type: The type of task processed (e.g., "representation", "summary") +DERIVER_QUEUE_ITEMS_PROCESSED = NamespacedCounter( + "deriver_queue_items_processed_total", + "Total deriver queue items processed", + ["namespace", "workspace_name", "task_type"], +) + +# Tracks the total number of input and output tokens processed by the deriver. +# +# Incremented in: src/deriver/deriver.py after the critical analysis call is made +# Labels: +# - task_type: The type of task that processed the tokens (e.g., "representation") +DERIVER_TOKENS_PROCESSED = NamespacedCounter( + "tokens_processed_total", + "Total tokens processed", + [ + "namespace", + "task_type", + ], +) + + +async def metrics() -> StarletteResponse: + """Prometheus metrics endpoint""" + if not settings.METRICS.ENABLED: + return StarletteResponse("Metrics are disabled", status_code=404) + try: + return StarletteResponse( + generate_latest(REGISTRY), media_type=CONTENT_TYPE_LATEST + ) + except Exception as e: + logger.error(f"Failed to generate metrics: {e}", exc_info=True) + return StarletteResponse("Failed to generate metrics", status_code=500) diff --git a/src/routers/messages.py b/src/routers/messages.py index 2ae174c1..36f32b7a 100644 --- a/src/routers/messages.py +++ b/src/routers/messages.py @@ -16,7 +16,7 @@ from fastapi_pagination.ext.sqlalchemy import apaginate from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm.attributes import flag_modified -from src import crud, schemas +from src import crud, prometheus, schemas from src.config import settings from src.dependencies import db from src.deriver import enqueue @@ -57,6 +57,10 @@ async def create_messages_for_session( session_name=session_id, ) + prometheus.MESSAGES_CREATED.labels( + workspace_name=workspace_id, + ).inc(len(created_messages)) + # Enqueue for processing (existing logic) payloads = [ { @@ -73,6 +77,7 @@ async def create_messages_for_session( # Enqueue all messages in one call background_tasks.add_task(enqueue, payloads) + return created_messages except ValueError as e: logger.warning(f"Failed to create messages for session {session_id}: {str(e)}") @@ -137,6 +142,9 @@ async def create_messages_with_file( logger.info( f"Batch of {len(created_messages)} messages created from file uploads and queued for processing" ) + prometheus.MESSAGES_CREATED.labels( + workspace_name=workspace_id, + ).inc(len(created_messages)) return created_messages diff --git a/src/routers/peers.py b/src/routers/peers.py index a6ecb10e..f3879bf3 100644 --- a/src/routers/peers.py +++ b/src/routers/peers.py @@ -8,7 +8,7 @@ from fastapi_pagination import Page from fastapi_pagination.ext.sqlalchemy import apaginate from sqlalchemy.ext.asyncio import AsyncSession -from src import crud, schemas +from src import crud, prometheus, schemas from src.dependencies import db, tracked_db from src.dialectic import chat as dialectic_chat from src.exceptions import AuthenticationException, ResourceNotFoundException @@ -178,6 +178,12 @@ async def chat( # and it's answered from the omniscient Honcho perspective observed=options.target if options.target is not None else peer_id, ) + + if prometheus.METRICS_ENABLED: + prometheus.DIALECTIC_CALLS.labels( + workspace_name=workspace_id, + ).inc() + return schemas.DialecticResponse(content=str(response)) async def parse_stream() -> AsyncGenerator[str, None]: @@ -190,6 +196,12 @@ async def chat( observer=peer_id, observed=options.target if options.target is not None else peer_id, ) + + if prometheus.METRICS_ENABLED: + prometheus.DIALECTIC_CALLS.labels( + workspace_name=workspace_id, + ).inc() + if isinstance(stream, AsyncIterator): async for chunk in stream: if chunk.content: diff --git a/src/utils/logging.py b/src/utils/logging.py index dc9d6ad6..8d28e1db 100644 --- a/src/utils/logging.py +++ b/src/utils/logging.py @@ -8,6 +8,7 @@ import datetime from collections.abc import Callable from typing import Any +from fastapi import Request from rich import box from rich.console import Console from rich.table import Table @@ -179,3 +180,16 @@ def log_performance_metrics( if metrics: console.print(table) console.print() + + +def normalize_template_path(path: str) -> str: + if path != "/" and path.endswith("/"): + return path.rstrip("/") + return path + + +def get_route_template(request: Request) -> str: + route = request.scope.get("route") + if route and getattr(route, "path", None): + return normalize_template_path(route.path) + return "unknown" diff --git a/tests/test_llm_mock.py b/tests/test_llm_mock.py index 3cb877ab..92292655 100644 --- a/tests/test_llm_mock.py +++ b/tests/test_llm_mock.py @@ -44,6 +44,7 @@ async def test_generic_honcho_llm_call_mock(): ), history="test history", new_turns=["test new turn"], + estimated_input_tokens=100, ) # Verify that we get a mock result, not an actual LLM call diff --git a/uv.lock b/uv.lock index f2396942..a8615c45 100644 --- a/uv.lock +++ b/uv.lock @@ -689,6 +689,7 @@ dependencies = [ { name = "openai" }, { name = "pdfplumber" }, { name = "pgvector" }, + { name = "prometheus-client" }, { name = "psycopg", extra = ["binary"] }, { name = "pydantic" }, { name = "pydantic-settings" }, @@ -732,6 +733,7 @@ requires-dist = [ { name = "openai", specifier = ">=1.99.7" }, { name = "pdfplumber", specifier = ">=0.11.7" }, { name = "pgvector", specifier = ">=0.2.5" }, + { name = "prometheus-client", specifier = ">=0.20.0" }, { name = "psycopg", extras = ["binary"], specifier = ">=3.1.19" }, { name = "pydantic", specifier = ">=2.11.7" }, { name = "pydantic-settings", specifier = ">=2.10.1" }, @@ -1553,6 +1555,27 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/88/74/a88bf1b1efeae488a0c0b7bdf71429c313722d1fc0f377537fbe554e6180/pre_commit-4.2.0-py2.py3-none-any.whl", hash = "sha256:a009ca7205f1eb497d10b845e52c838a98b6cdd2102a6c8e4540e94ee75c58bd", size = 220707, upload-time = "2025-03-18T21:35:19.343Z" }, ] +[[package]] +name = "prometheus-client" +version = "0.23.1" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/23/53/3edb5d68ecf6b38fcbcc1ad28391117d2a322d9a1a3eff04bfdb184d8c3b/prometheus_client-0.23.1.tar.gz", hash = "sha256:6ae8f9081eaaaf153a2e959d2e6c4f4fb57b12ef76c8c7980202f1e57b48b2ce", size = 80481, upload-time = "2025-09-18T20:47:25.043Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/b8/db/14bafcb4af2139e046d03fd00dea7873e48eafe18b7d2797e73d6681f210/prometheus_client-0.23.1-py3-none-any.whl", hash = "sha256:dd1913e6e76b59cfe44e7a4b83e01afc9873c1bdfd2ed8739f1e76aeca115f99", size = 61145, upload-time = "2025-09-18T20:47:23.875Z" }, +] + +[[package]] +name = "proto-plus" +version = "1.26.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "protobuf" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/f4/ac/87285f15f7cce6d4a008f33f1757fb5a13611ea8914eb58c3d0d26243468/proto_plus-1.26.1.tar.gz", hash = "sha256:21a515a4c4c0088a773899e23c7bbade3d18f9c66c73edd4c7ee3816bc96a012", size = 56142, upload-time = "2025-03-10T15:54:38.843Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/4e/6d/280c4c2ce28b1593a19ad5239c8b826871fc6ec275c21afc8e1820108039/proto_plus-1.26.1-py3-none-any.whl", hash = "sha256:13285478c2dcf2abb829db158e1047e2f1e8d63a077d94263c2b88b043c75a66", size = 50163, upload-time = "2025-03-10T15:54:37.335Z" }, +] + [[package]] name = "protobuf" version = "5.29.5"