jpmschweitzer/tatlock
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 5 Wiki Activity

feat: integrate tracing throughout request pipeline

Browse Source 
Instrument the full request flow with trace spans for debugging:

- Wrap expert delegations (librarian/biographer/housekeeper) in spans
- Add orchestrate and synthesize spans to TatlockAgent
- Trace Steward analysis in preprocessing
- Start/end traces in response service with context management
- Simplify router by moving context handling to service layer
- Include tracing router in debug mode
- Remove benchmark recording from tool_tracking and steward service

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
jpmschweitzer
2025-12-22 10:26:37 +01:00
co-authored by Claude Opus 4.5
parent 2a9449bc81
commit 87f2926db2
8 changed files with 502 additions and 369 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/responses/router.py
+5 -73
View File
@@ -10,7 +10,6 @@ from sse_starlette.sse import EventSourceResponse
from src.responses import service
from src.responses.schemas import ResponseRequest, Response
from src.core.exceptions import ModelNotFoundError, AppException
from src.core.context import current_user, current_conversation, get_default_user
from src.core.logging_config import get_logger
logger = get_logger(__name__)
@@ -37,77 +36,16 @@ async def create_response(
Returns:
Response object or SSE stream
Example non-streaming request:
POST /v1/responses
{
"model": "lorem-tester",
"input": [{"role": "user", "content": "Hello"}],
"reasoning": {"effort": "medium", "summary": "auto"},
"stream": false
}
Example streaming request:
POST /v1/responses
{
"model": "lorem-tester",
"input": [{"role": "user", "content": "Hello"}],
"stream": true
}
Response format (non-streaming):
{
"id": "resp_...",
"object": "response",
"created_at": 1733529600,
"model": "lorem-tester",
"status": "completed",
"output": [
{
"type": "reasoning",
"id": "rs_...",
"summary": ["Analyzing...", "Considering..."]
},
{
"type": "message",
"id": "msg_...",
"role": "assistant",
"content": [{"type": "output_text", "text": "Lorem ipsum..."}]
}
],
"usage": {
"input_tokens": 10,
"output_tokens": 50,
"reasoning_tokens": 20,
"total_tokens": 80
}
}
Streaming format (SSE):
event: response.reasoning_summary_text.delta
data: {"delta": "Analyzing..."}
event: response.output_text.delta
data: {"delta": "Lorem"}
event: response.done
data: {"response": {...}}
"""
# Set request context (propagates through all async calls)
effective_user = request.user or get_default_user()
user_token = current_user.set(effective_user)
conv_id = request.metadata.get("conversation_id") if request.metadata else None
conv_token = current_conversation.set(conv_id)
logger.info(
"response_request_received",
model=request.model,
user=effective_user,
conversation_id=conv_id,
user=request.user,
streaming=request.stream,
)
try:
# Check if this is a Tatlock request - use Steward preprocessing (Phase 2)
# Check if this is a Tatlock request - use Steward preprocessing
model_id = request.model
if "." in model_id:
model_id = model_id.split(".", 1)[1]
@@ -116,21 +54,20 @@ async def create_response(
if request.stream:
logger.info("Streaming response requested")
if use_steward:
logger.info("Streaming with Steward preprocessing for Tatlock request")
# Use Steward + Tatlock streaming (Milestone 3.5)
from src.responses.streaming import StreamingCoordinator
coordinator = StreamingCoordinator()
return EventSourceResponse(
coordinator.stream_response_with_steward(request)
)
else:
# Regular streaming for non-Tatlock models
return EventSourceResponse(
service.create_response_stream(request)
)
# Use appropriate service method
# Non-streaming response
if use_steward:
logger.info("Using Steward preprocessing for Tatlock request")
return await service.create_response_with_steward(request)
@@ -148,8 +85,3 @@ async def create_response(
except Exception as e:
logger.error(f"Unexpected error: {e}", exc_info=True)
raise HTTPException(status_code=500, detail="Internal server error")
finally:
# Reset context (important for connection reuse)
current_user.reset(user_token)
current_conversation.reset(conv_token)
src/responses/service.py
+251 -140
View File
@@ -26,6 +26,8 @@ from src.responses.context import ContextWindow
from src.core.preprocessing import preprocess_request
from src.core.tool_tracking import ToolCallTracker
from src.core.logging_config import get_logger
from src.core.tracing import start_trace, end_trace, start_span, SpanType
from src.core.context import current_user, current_conversation, get_default_user
from src.agents.steward.schemas import StewardRecommendation
import re
@@ -34,6 +36,29 @@ import asyncio
logger = get_logger(__name__)
def _extract_user_input(input_data) -> str:
"""Extract user input text from request input for tracing."""
if isinstance(input_data, str):
return input_data
elif isinstance(input_data, list) and input_data:
last_msg = input_data[-1]
if isinstance(last_msg, dict):
return last_msg.get("content", str(last_msg))
return str(last_msg)
return ""
def _extract_response_preview(response: Response) -> str:
"""Extract response preview text for tracing."""
if response.output:
for item in response.output:
if hasattr(item, 'content'):
for content in item.content:
if hasattr(content, 'text'):
return content.text[:200]
return ""
async def _execute_single_delegation(
agent_name: str,
task: str,
@@ -405,45 +430,88 @@ async def create_response(request: ResponseRequest) -> Response:
# Get or generate conversation ID
conversation_id = await _conversation_history.get_conversation_id(request)
# Strip pipeline prefix if present (e.g., "pipeline.model" -> "model")
model_id = request.model
if "." in model_id:
model_id = model_id.split(".", 1)[1]
# Set context for tracing
effective_user = request.user or get_default_user()
current_user.set(effective_user)
current_conversation.set(conversation_id)
# Get agent for model
agent = ModelRegistry.get_agent(model_id)
# Extract user input for tracing
user_input = _extract_user_input(request.input)
# Collect all output items from agent
output_items = []
async for item in agent.generate_response(
messages=request.input,
reasoning=request.reasoning,
tools=request.tools,
temperature=request.temperature,
max_tokens=request.max_output_tokens,
stop=request.stop,
):
output_items.append(item)
# Convert agent OutputItems to schema OutputItems
converted_items = _convert_output_items(output_items)
# Calculate token usage
usage = _calculate_usage(request.input, output_items)
response = Response(
id=f"resp_{generate_id()}",
created_at=int(time.time()),
model=request.model,
status="completed",
output=converted_items,
usage=usage
# Start trace
trace = start_trace(
conversation_id=conversation_id,
user=effective_user,
request={
"model": request.model,
"input_preview": user_input[:200] if user_input else "",
"full_input": request.input,
"streaming": False,
},
)
# Track conversation history (for analytics and future vector memory)
await _conversation_history.add_response(conversation_id, response)
# Start service span
service_span = start_span(
"create_response",
SpanType.ROUTER,
metadata={"model": request.model, "user": effective_user},
)
return response
try:
# Strip pipeline prefix if present (e.g., "pipeline.model" -> "model")
model_id = request.model
if "." in model_id:
model_id = model_id.split(".", 1)[1]
# Get agent for model
agent = ModelRegistry.get_agent(model_id)
# Collect all output items from agent
output_items = []
async for item in agent.generate_response(
messages=request.input,
reasoning=request.reasoning,
tools=request.tools,
temperature=request.temperature,
max_tokens=request.max_output_tokens,
stop=request.stop,
):
output_items.append(item)
# Convert agent OutputItems to schema OutputItems
converted_items = _convert_output_items(output_items)
# Calculate token usage
usage = _calculate_usage(request.input, output_items)
response = Response(
id=f"resp_{generate_id()}",
created_at=int(time.time()),
model=request.model,
status="completed",
output=converted_items,
usage=usage
)
# Track conversation history (for analytics and future vector memory)
await _conversation_history.add_response(conversation_id, response)
# End trace with response info
response_preview = _extract_response_preview(response)
end_trace(
response={
"output_preview": response_preview,
"output_count": len(response.output) if response.output else 0,
"status": response.status,
},
status="completed",
)
return response
except Exception as e:
end_trace(status="error")
raise
async def create_response_with_steward(request: ResponseRequest) -> Response:
@@ -454,7 +522,7 @@ async def create_response_with_steward(request: ResponseRequest) -> Response:
1. Steward analyzes the request and recommends capabilities
2. Phase 1: Tatlock orchestrates tool calls and expert delegations
3. Phase 2: Tatlock synthesizes butler-toned response from results
4. Tool usage is tracked for benchmarking
4. Tool usage is tracked for analysis
Args:
request: Response request
@@ -473,133 +541,176 @@ async def create_response_with_steward(request: ResponseRequest) -> Response:
# Get or generate conversation ID
conversation_id = await _conversation_history.get_conversation_id(request)
# Extract user message and conversation history
user_message = ""
for msg in reversed(request.input):
if msg.get("role") == "user":
user_message = msg.get("content", "")
break
# Set context for tracing
effective_user = request.user or get_default_user()
current_user.set(effective_user)
current_conversation.set(conversation_id)
# Conversation history is all messages except the current one
conversation_history = request.input[:-1] if len(request.input) > 1 else []
# Extract user input for tracing
user_input = _extract_user_input(request.input)
logger.info(
"creating_response_with_steward",
user_message_preview=user_message[:100],
history_length=len(conversation_history),
# Start trace
trace = start_trace(
conversation_id=conversation_id,
user=effective_user,
request={
"model": request.model,
"input_preview": user_input[:200] if user_input else "",
"full_input": request.input,
"streaming": False,
},
)
# Steward preprocessing
enriched = await preprocess_request(
user_message,
conversation_history=conversation_history,
conversation_id=conversation_id,
# Start service span
service_span = start_span(
"create_response_with_steward",
SpanType.ROUTER,
metadata={"model": request.model, "user": effective_user},
)
# Initialize tool tracker
tracker = ToolCallTracker(
recommended_capabilities=enriched.recommendation.recommended_capabilities,
conversation_id=conversation_id,
)
try:
# Extract user message and conversation history
user_message = ""
for msg in reversed(request.input):
if msg.get("role") == "user":
user_message = msg.get("content", "")
break
# Check if direct delegation is recommended
# If Steward recommends ONLY delegation agents (biographer/librarian/housekeeper),
# we still use two-phase but delegate directly in Phase 1
delegation_agents = {"biographer", "librarian", "housekeeper"}
delegation_only = all(
cap in delegation_agents
for cap in enriched.recommendation.recommended_capabilities
) and enriched.recommendation.recommended_capabilities
# Conversation history is all messages except the current one
conversation_history = request.input[:-1] if len(request.input) > 1 else []
from src.agents.tatlock import TatlockAgent
tatlock = TatlockAgent()
# Use enriched query (with location/timezone context) if available
effective_query = enriched.recommendation.enriched_query or user_message
if delegation_only:
# Direct delegation path - collect results then synthesize
orchestration_results = await _direct_delegation_with_results(
effective_query, enriched.recommendation, tracker, conversation_id
)
else:
# Phase 1: Orchestrate tool calls
orchestration_results = await tatlock.orchestrate_tool_calls(
user_message=effective_query,
steward_note=enriched.steward_note,
scoped_tools=enriched.scoped_tools,
message_history=conversation_history,
tool_tracker=tracker,
logger.info(
"creating_response_with_steward",
user_message_preview=user_message[:100],
history_length=len(conversation_history),
conversation_id=conversation_id,
)
# Handle text-based delegation fallback if present
if "[DELEGATE:" in orchestration_results.get("raw_output", ""):
text_delegation_results = await _handle_text_delegation(
orchestration_results["raw_output"], tracker, conversation_id
# Steward preprocessing
enriched = await preprocess_request(
user_message,
conversation_history=conversation_history,
conversation_id=conversation_id,
)
# Initialize tool tracker
tracker = ToolCallTracker(
recommended_capabilities=enriched.recommendation.recommended_capabilities,
conversation_id=conversation_id,
)
# Check if direct delegation is recommended
# If Steward recommends ONLY delegation agents (biographer/librarian/housekeeper),
# we still use two-phase but delegate directly in Phase 1
delegation_agents = {"biographer", "librarian", "housekeeper"}
delegation_only = all(
cap in delegation_agents
for cap in enriched.recommendation.recommended_capabilities
) and enriched.recommendation.recommended_capabilities
from src.agents.tatlock import TatlockAgent
tatlock = TatlockAgent()
# Use enriched query (with location/timezone context) if available
effective_query = enriched.recommendation.enriched_query or user_message
if delegation_only:
# Direct delegation path - collect results then synthesize
orchestration_results = await _direct_delegation_with_results(
effective_query, enriched.recommendation, tracker, conversation_id
)
else:
# Phase 1: Orchestrate tool calls
orchestration_results = await tatlock.orchestrate_tool_calls(
user_message=effective_query,
steward_note=enriched.steward_note,
scoped_tools=enriched.scoped_tools,
message_history=conversation_history,
tool_tracker=tracker,
)
# Add text delegation results to expert_results
if text_delegation_results != orchestration_results["raw_output"]:
orchestration_results["expert_results"]["text_delegation"] = text_delegation_results
# Phase 2: Synthesize butler-toned response from all results
tatlock_response = await tatlock.synthesize_from_results(
user_message=user_message,
orchestration_results=orchestration_results,
message_history=conversation_history,
)
# Handle text-based delegation fallback if present
if "[DELEGATE:" in orchestration_results.get("raw_output", ""):
text_delegation_results = await _handle_text_delegation(
orchestration_results["raw_output"], tracker, conversation_id
)
# Add text delegation results to expert_results
if text_delegation_results != orchestration_results["raw_output"]:
orchestration_results["expert_results"]["text_delegation"] = text_delegation_results
# Finalize tool tracking
await tracker.finalize()
# Phase 2: Synthesize butler-toned response from all results
tatlock_response = await tatlock.synthesize_from_results(
user_message=user_message,
orchestration_results=orchestration_results,
message_history=conversation_history,
)
# Build response output items
output_items = []
# Finalize tool tracking
await tracker.finalize()
# Add Steward reasoning as a reasoning output item
output_items.append(ReasoningOutputItem(
id=f"reasoning_{generate_id()}",
summary=[
"🎩 Steward's Analysis:",
enriched.steward_reasoning,
],
status="completed"
))
# Build response output items
output_items = []
# Add Tatlock's message
output_items.append(MessageOutputItem(
id=f"msg_{generate_id()}",
role="assistant",
content=[OutputTextContent(
type="output_text",
text=tatlock_response,
annotations=[]
)],
status="completed"
))
# Add Steward reasoning as a reasoning output item
output_items.append(ReasoningOutputItem(
id=f"reasoning_{generate_id()}",
summary=[
"🎩 Steward's Analysis:",
enriched.steward_reasoning,
],
status="completed"
))
# Calculate usage (approximate)
usage = _calculate_usage(request.input, output_items)
# Add Tatlock's message
output_items.append(MessageOutputItem(
id=f"msg_{generate_id()}",
role="assistant",
content=[OutputTextContent(
type="output_text",
text=tatlock_response,
annotations=[]
)],
status="completed"
))
response = Response(
id=f"resp_{generate_id()}",
created_at=int(time.time()),
model=request.model,
status="completed",
output=output_items,
usage=usage
)
# Calculate usage (approximate)
usage = _calculate_usage(request.input, output_items)
# Track conversation history
await _conversation_history.add_response(conversation_id, response)
response = Response(
id=f"resp_{generate_id()}",
created_at=int(time.time()),
model=request.model,
status="completed",
output=output_items,
usage=usage
)
logger.info(
"response_with_steward_complete",
response_id=response.id,
recommended_capabilities=enriched.recommendation.recommended_capabilities,
tool_summary=tracker.get_summary(),
)
# Track conversation history
await _conversation_history.add_response(conversation_id, response)
return response
logger.info(
"response_with_steward_complete",
response_id=response.id,
recommended_capabilities=enriched.recommendation.recommended_capabilities,
tool_summary=tracker.get_summary(),
)
# End trace with response info
response_preview = _extract_response_preview(response)
end_trace(
response={
"output_preview": response_preview,
"output_count": len(response.output) if response.output else 0,
"status": response.status,
},
status="completed",
)
return response
except Exception as e:
end_trace(status="error")
raise
async def create_response_stream(