Skip to content

feat: Enhanced Phoenix sync with OpenInference schema support #32

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
visahak merged 1 commit into from
Jan 24, 2026
Merged

feat: Enhanced Phoenix sync with OpenInference schema support #32

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion kaizen/llm/tips/tips.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -112,10 +112,11 @@ def parse_openai_agents_trajectory(messages: list[dict]) -> dict:

def generate_tips(messages: list[dict]) -> list[Tip]:
prompt_file = Path(__file__).parent / "prompts/generate_tips.jinja2"
supports_response_format = "response_format" in get_supported_openai_params(
supported_params = get_supported_openai_params(
model=llm_settings.tips_model,
custom_llm_provider=llm_settings.custom_llm_provider,
)
supports_response_format = supported_params and "response_format" in supported_params
response_schema_enabled = supports_response_schema(
model=llm_settings.tips_model,
custom_llm_provider=llm_settings.custom_llm_provider,
Expand Down
114 changes: 102 additions & 12 deletions kaizen/sync/phoenix_sync.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -114,9 +114,95 @@ def _parse_content(self, content: Any) -> Any:

def _extract_messages_from_span(self, span: dict) -> list[dict]:
"""Extract messages from a single span's attributes."""
attrs = span.get("attributes", {})
attrs = span.get("attributes") or {}
messages = []

# Try OpenInference schema first (input_messages/output_messages)
# Note: Phoenix API might return these as JSON strings or lists depending on version
input_msgs = attrs.get("llm.input_messages")
output_msgs = attrs.get("llm.output_messages")

# If input_messages is missing, try parsing input.value
if input_msgs is None:
input_val = attrs.get("input.value")
if input_val:
try:
parsed_input = self._parse_content(input_val)
if isinstance(parsed_input, dict) and "messages" in parsed_input:
input_msgs = parsed_input["messages"]
elif isinstance(parsed_input, list): # rare but possible
input_msgs = parsed_input
except:
pass
Comment on lines +135 to +136

@coderabbitai coderabbitai Bot Jan 23, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟡 Minor

Avoid bare except with silent pass.

Bare except catches all exceptions including KeyboardInterrupt and SystemExit. Silent pass makes debugging difficult when unexpected parsing failures occur.

Proposed fix 
-                except:
-                    pass
+                except (json.JSONDecodeError, ValueError, TypeError) as e:
+                    logger.debug(f"Failed to parse input.value: {e}")
🧰 Tools
🪛 Ruff (0.14.13)

135-135: Do not use bare except

(E722)

135-136: try-except-pass detected, consider logging the exception

(S110)

🤖 Prompt for AI Agents 
In `@kaizen/sync/phoenix_sync.py` around lines 135 - 136, Replace the bare
"except: pass" in phoenix_sync.py with a specific exception handler and logging:
catch the likely parsing/processing exceptions (e.g., ValueError, KeyError,
json.JSONDecodeError or Exception if uncertain) using "except <ExceptionType> as
e" and emit a logged error (e.g., logger.error or logger.exception with
contextual info about the record/item being parsed) instead of silently passing
so failures are visible while still allowing the loop to continue; update the
try/except block around the failing parse/processing code accordingly.


if input_msgs:
# Handle OpenInference format
# Ensure it's a list
if isinstance(input_msgs, str):
input_msgs = self._parse_content(input_msgs)

if isinstance(input_msgs, list):
for i, msg in enumerate(input_msgs):
# OpenInference often uses message.role / message.content keys in flattened export
# but via API it might be cleaner. Let's handle dict access safely.
role = msg.get("message.role") or msg.get("role")
content = msg.get("message.content") or msg.get("content")
tool_calls = msg.get("message.tool_calls") or msg.get("tool_calls")

if role:
mapped_msg = {
"index": i,
"type": "prompt",
"role": role,
"content": self._parse_content(content),
}
if tool_calls:
mapped_msg["tool_calls"] = tool_calls
messages.append(mapped_msg)

# Handle Output/Completion from OpenInference
if output_msgs is None:
output_val = attrs.get("output.value")
if output_val:
try:
parsed_output = self._parse_content(output_val)
# output.value is often just the string content or a list of choices
if isinstance(parsed_output, list) and len(parsed_output) > 0 and "message" in parsed_output[0]:
output_msgs = [c["message"] for c in parsed_output]
elif isinstance(parsed_output, dict) and "choices" in parsed_output: # OpenAI response format
output_msgs = [c["message"] for c in parsed_output["choices"]]
else:
# Fallback for simple string output
# output_msgs = [{"role": "assistant", "content": output_val}]
pass
except:
pass
Comment on lines +178 to +179

@coderabbitai coderabbitai Bot Jan 23, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟡 Minor

Same issue: bare except with silent pass.

Apply the same fix as suggested for input parsing.

Proposed fix 
-                except:
-                    pass
+                except (json.JSONDecodeError, ValueError, TypeError) as e:
+                    logger.debug(f"Failed to parse output.value: {e}")
🧰 Tools
🪛 Ruff (0.14.13)

178-178: Do not use bare except

(E722)

178-179: try-except-pass detected, consider logging the exception

(S110)

🤖 Prompt for AI Agents 
In `@kaizen/sync/phoenix_sync.py` around lines 178 - 179, Replace the bare
"except: pass" in phoenix_sync.py with a specific exception handler and proper
logging/handling: identify the surrounding function/block containing that bare
except (the except block shown in the diff), change it to catch the expected
exception types (e.g., ValueError, KeyError or more generally "except Exception
as e" if multiple types are possible), and record the error (e.g., logger.error
or logger.exception with the exception object) or take corrective action instead
of silently passing so failures are observable and debuggable.


if output_msgs:
if isinstance(output_msgs, str):
output_msgs = self._parse_content(output_msgs)

if isinstance(output_msgs, list):
for i, msg in enumerate(output_msgs):
role = msg.get("message.role") or msg.get("role")
content = msg.get("message.content") or msg.get("content")
tool_calls = msg.get("message.tool_calls") or msg.get("tool_calls")

if role:
mapped_msg = {
"index": i,
"type": "completion",
"role": role,
"content": self._parse_content(content),
}
if tool_calls:
mapped_msg["tool_calls"] = tool_calls
messages.append(mapped_msg)

if messages:
return messages

# Fallback to GenAI semantic conventions (original code)
# Extract prompt messages
prompt_indices = set()
for key in attrs:
Expand Down Expand Up @@ -231,7 +317,7 @@ def _convert_to_openai_format(self, content: Any, role: str) -> dict:

def _extract_trajectory(self, span: dict) -> dict:
"""Extract a complete trajectory from a span."""
attrs = span.get("attributes", {})
attrs = span.get("attributes") or {}
messages = self._extract_messages_from_span(span)

openai_messages = []
Expand Down Expand Up @@ -260,9 +346,9 @@ def _extract_trajectory(self, span: dict) -> dict:
"timestamp": span.get("start_time"),
"messages": openai_messages,
"usage": {
"prompt_tokens": attrs.get("gen_ai.usage.prompt_tokens"),
"completion_tokens": attrs.get("gen_ai.usage.completion_tokens"),
"total_tokens": attrs.get("llm.usage.total_tokens"),
"prompt_tokens": attrs.get("gen_ai.usage.prompt_tokens") or attrs.get("llm.token_count.prompt"),
"completion_tokens": attrs.get("gen_ai.usage.completion_tokens") or attrs.get("llm.token_count.completion"),
"total_tokens": attrs.get("llm.usage.total_tokens") or attrs.get("llm.token_count.total"),
},
}

Expand Down Expand Up @@ -303,7 +389,7 @@ def _process_trajectory(self, trajectory: dict) -> int:
if messages:
entity = Entity(
type="trajectory",
content=messages,
content=json.dumps(messages),
metadata={
"trace_id": trajectory["trace_id"],
"span_id": trajectory["span_id"],
Expand All @@ -313,6 +399,7 @@ def _process_trajectory(self, trajectory: dict) -> int:
"usage": trajectory.get("usage"),
},
)

self.client.update_entities(
namespace_id=self.namespace_id,
entities=[entity],
Expand Down Expand Up @@ -380,9 +467,9 @@ def sync(
errors = []

for span in spans:
# Filter to LLM request spans
if span.get("name") != "litellm_request":
continue
# Filter to LLM request spans - accept any span with prompt attributes
# if span.get("name") != "litellm_request":
# continue

# Filter errors if requested
if not include_errors and span.get("status_code") == "ERROR":
Expand All @@ -394,9 +481,12 @@ def sync(
skipped += 1
continue

# Only include spans with actual messages
attrs = span.get("attributes", {})
if not any(k.startswith("gen_ai.prompt.") for k in attrs):
# Only include spans with actual messages or GenAI/LLM prompt attributes
attrs = span.get("attributes") or {}
has_gen_ai = any(k.startswith("gen_ai.prompt.") for k in attrs)
has_llm_msgs = "llm.input_messages" in attrs or "input.value" in attrs

if not (has_gen_ai or has_llm_msgs):
continue

try:
Expand Down