From 992ec86f965a53c3a93b9ccf0a085ad5c577a085 Mon Sep 17 00:00:00 2001 From: Rajat Ahuja Date: Tue, 11 Nov 2025 16:41:00 -0500 Subject: [PATCH] fix: add local tracing --- src/deriver/deriver.py | 25 ++++++++- src/deriver/reasoners/deductive.py | 6 +-- src/deriver/reasoners/explicit.py | 6 +-- src/utils/logging.py | 87 +++++++++++++++++++++++++++++- 4 files changed, 115 insertions(+), 9 deletions(-) diff --git a/src/deriver/deriver.py b/src/deriver/deriver.py index c95105eb..5a4a6a47 100644 --- a/src/deriver/deriver.py +++ b/src/deriver/deriver.py @@ -317,7 +317,7 @@ class CertaintyReasoner: ) # Step 1: Explicit reasoning - explicit_response = await self.explicit_reasoner.reason( + explicit_response, explicit_prompt = await self.explicit_reasoner.reason( working_representation=working_representation, history=history, speaker_peer_card=speaker_peer_card, @@ -337,7 +337,7 @@ class CertaintyReasoner: obs.content for obs in explicit_observations.explicit ] + [obs.content for obs in explicit_observations.implicit] - deductive_response = await self.deductive_reasoner.reason( + deductive_response, deductive_prompt = await self.deductive_reasoner.reason( working_representation=working_representation, atomic_propositions=atomic_propositions, history=history, @@ -359,6 +359,27 @@ class CertaintyReasoner: deductive=deductive_observations.deductive, ) + # Save trace if local metrics collection is enabled + from src.utils.logging import save_reasoning_trace + + save_reasoning_trace( + provider=settings.DERIVER.PROVIDER, + model=settings.DERIVER.MODEL, + max_tokens=settings.DERIVER.MAX_OUTPUT_TOKENS + or settings.LLM.DEFAULT_MAX_TOKENS, + peer_id=self.observed, + peer_card=speaker_peer_card, + message_created_at=latest_message.created_at, + working_representation=working_representation, + history=history, + new_turns=new_turns, + explicit_prompt=explicit_prompt, + explicit_response=explicit_response.model_dump(), + deductive_prompt=deductive_prompt, + deductive_response=deductive_response.model_dump(), + atomic_propositions=atomic_propositions, + ) + analysis_duration_ms = (time.perf_counter() - analysis_start) * 1000 accumulate_metric( f"deriver_{latest_message.id}_{self.observer}", diff --git a/src/deriver/reasoners/deductive.py b/src/deriver/reasoners/deductive.py index d5632615..1012c23b 100644 --- a/src/deriver/reasoners/deductive.py +++ b/src/deriver/reasoners/deductive.py @@ -58,7 +58,7 @@ class DeductiveReasoner(BaseReasoner): atomic_propositions: list[str], history: str, speaker_peer_card: list[str] | None, - ) -> DeductiveResponse: + ) -> tuple[DeductiveResponse, str]: """Process input through deductive reasoning. Args: @@ -69,7 +69,7 @@ class DeductiveReasoner(BaseReasoner): speaker_peer_card: Peer card for the observed peer Returns: - DeductiveResponse containing only deductive observations + Tuple of (DeductiveResponse, prompt string) """ latest_message = self.ctx[-1] new_turns = [ @@ -109,7 +109,7 @@ class DeductiveReasoner(BaseReasoner): task_type="deductive_reasoning", ).inc(response.output_tokens + self.estimated_input_tokens) - return response.content + return response.content, prompt except Exception as e: raise exceptions.LLMError( speaker_peer_card=speaker_peer_card, diff --git a/src/deriver/reasoners/explicit.py b/src/deriver/reasoners/explicit.py index c2b12ecc..04bb6d34 100644 --- a/src/deriver/reasoners/explicit.py +++ b/src/deriver/reasoners/explicit.py @@ -57,7 +57,7 @@ class ExplicitReasoner(BaseReasoner): working_representation: Representation, history: str, speaker_peer_card: list[str] | None, - ) -> ExplicitResponse: + ) -> tuple[ExplicitResponse, str]: """Process input through explicit reasoning. Args: @@ -66,7 +66,7 @@ class ExplicitReasoner(BaseReasoner): speaker_peer_card: Peer card for the observed peer Returns: - ExplicitResponse containing only explicit observations + Tuple of (ExplicitResponse, prompt string) """ latest_message = self.ctx[-1] new_turns = [ @@ -105,7 +105,7 @@ class ExplicitReasoner(BaseReasoner): task_type="explicit_reasoning", ).inc(response.output_tokens + self.estimated_input_tokens) - return response.content + return response.content, prompt except Exception as e: raise exceptions.LLMError( speaker_peer_card=speaker_peer_card, diff --git a/src/utils/logging.py b/src/utils/logging.py index 2002b7bd..3d46f208 100644 --- a/src/utils/logging.py +++ b/src/utils/logging.py @@ -6,7 +6,7 @@ and a conditional observe decorator that only applies when Langfuse is configure import datetime from collections.abc import Callable -from typing import ParamSpec, TypeVar, overload +from typing import Any, ParamSpec, TypeVar, overload from fastapi import Request from langfuse import observe # pyright: ignore @@ -249,3 +249,88 @@ def get_route_template(request: Request) -> str: if route and getattr(route, "path", None): return normalize_template_path(route.path) return "unknown" + + +def save_reasoning_trace( + provider: str, + model: str, + max_tokens: int, + peer_id: str, + peer_card: list[str] | None, + message_created_at: datetime.datetime, + working_representation: Representation, + history: str, + new_turns: list[str], + explicit_prompt: str, + explicit_response: dict[str, Any], + deductive_prompt: str, + deductive_response: dict[str, Any], + atomic_propositions: list[str], +) -> None: + """ + Save the reasoning trace (explicit + deductive calls) to trace.jsonl file. + Uses JSONL format (one JSON object per line) for efficient appending. + Only writes if COLLECT_METRICS_LOCAL is enabled. + + Args: + provider: LLM provider name + model: Model name + max_tokens: Max tokens setting + peer_id: ID of peer being analyzed + peer_card: Peer card information + message_created_at: Timestamp of message + working_representation: Current representation context + history: Conversation history + new_turns: New conversation turns + explicit_prompt: Prompt for explicit reasoning + explicit_response: Response from explicit reasoning (as dict) + deductive_prompt: Prompt for deductive reasoning + deductive_response: Response from deductive reasoning (as dict) + atomic_propositions: Atomic propositions passed to deductive reasoner + """ + if not COLLECT_METRICS_LOCAL: + return + + import fcntl + import json + from pathlib import Path + + trace_file = Path("trace.jsonl") + + trace_data = { + "timestamp": datetime.datetime.now(datetime.timezone.utc).isoformat(), + "provider": provider, + "model": model, + "max_tokens": max_tokens, + "peer_id": peer_id, + "peer_card": peer_card, + "message_created_at": message_created_at.isoformat(), + "working_representation": { + "explicit": [obs.content for obs in working_representation.explicit], + "implicit": [obs.content for obs in working_representation.implicit], + "deductive": [ + { + "conclusion": obs.conclusion, + "premises": obs.premises, + } + for obs in working_representation.deductive + ], + }, + "history": history, + "new_turns": new_turns, + "explicit_call": { + "prompt": explicit_prompt, + "response": explicit_response, + }, + "deductive_call": { + "prompt": deductive_prompt, + "response": deductive_response, + "atomic_propositions": atomic_propositions, + }, + } + + # Use file locking to handle concurrent writes from multiple processes + with open(trace_file, "a") as f: + fcntl.flock(f.fileno(), fcntl.LOCK_EX) + f.write(json.dumps(trace_data) + "\n") + fcntl.flock(f.fileno(), fcntl.LOCK_UN)