feat: Enhanced Phoenix sync with OpenInference schema support - #32
Conversation
📝 WalkthroughWalkthroughThe PR enhances message extraction in Phoenix sync to support OpenInference formats (llm.input_messages, llm.output_messages, input.value, output.value) with unified message structure, changes trajectory storage from list to JSON string, updates token usage mapping fallbacks, and improves span filtering and None handling safety. Changes
Sequence Diagram(s)sequenceDiagram
participant Span as Phoenix Span
participant Filter as Span Filter
participant Extract as Message Extractor
participant Format as OpenAI Formatter
participant Store as Trajectory Storage
Span->>Filter: Check span eligibility (name or GenAI/llm attributes)
alt Span matches new broader criteria
Filter->>Extract: Process span
Extract->>Extract: Check llm.input_messages/output_messages
alt OpenInference format found
Extract->>Extract: Parse input.value/output.value
Extract->>Extract: Build unified messages (index/type/role/content)
else
Extract->>Extract: Fallback to GenAI prompt/completion
end
Extract->>Format: Unified message structure
Format->>Format: Convert to OpenAI format
Format->>Store: Stringified trajectory (json.dumps)
else
Filter-->>Span: Skip span
end
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~25 minutes Possibly related PRs
Suggested reviewers
Poem
🚥 Pre-merge checks | ✅ 3✅ Passed checks (3 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing touches
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
kaizen/sync/phoenix_sync.py (1)
390-402: Trajectory content is stored as JSON string, but deserialization is inconsistent between backends.The
json.dumps(messages)storage change requires careful handling: Milvus backend automatically deserializes viaparse_milvus_entity(), so consumers get a list back. However, Filesystem backend returns the raw JSON string without deserialization, creating inconsistent behavior. Code using Filesystem backend that expectsentity.contentto be a list will receive a JSON string instead. Either add deserialization to Filesystem backend'ssearch_entities(similar to Milvus'sparse_milvus_entity), or document this backend-specific behavior.
🤖 Fix all issues with AI agents
In `@kaizen/sync/phoenix_sync.py`:
- Around line 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.
- Around line 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.
🧹 Nitpick comments (3)
kaizen/sync/phoenix_sync.py (3)
144-161: Add type guard before accessing dict methods.If
input_msgscontains non-dict items (e.g., strings), callingmsg.get()will raiseAttributeError.Proposed fix
if isinstance(input_msgs, list): for i, msg in enumerate(input_msgs): + if not isinstance(msg, dict): + continue # OpenInference often uses message.role / message.content keys in flattened export
185-200: Same type guard needed for output messages.Apply the same type check before calling
msg.get().Proposed fix
if isinstance(output_msgs, list): for i, msg in enumerate(output_msgs): + if not isinstance(msg, dict): + continue role = msg.get("message.role") or msg.get("role")
470-472: Remove commented-out code instead of leaving it.Dead code should be deleted rather than commented out. Version control preserves history if needed later.
Proposed fix
for span in spans: - # Filter to LLM request spans - accept any span with prompt attributes - # if span.get("name") != "litellm_request": - # continue + # Filter to LLM request spans - accept any span with prompt attributes
| except: | ||
| pass |
There was a problem hiding this comment.
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.
| except: | ||
| pass |
There was a problem hiding this comment.
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.
2 participants
Description
Summary
Enhanced the Phoenix sync module to support multiple OpenTelemetry/Phoenix schema formats, improving compatibility with different tracing implementations and Phoenix API versions.
Changes Made
Enhanced Message Extraction (
_extract_messages_from_span)llm.input_messages,llm.output_messages)input.valueandoutput.valueattributesmessage.role,message.content) and nested message formatsImproved Token Usage Tracking
llm.token_count.prompt(fallback forgen_ai.usage.prompt_tokens)llm.token_count.completion(fallback forgen_ai.usage.completion_tokens)llm.token_count.total(fallback forllm.usage.total_tokens)Enhanced Span Filtering
litellm_requestcheck)Improved Null Safety
span.get("attributes", {})tospan.get("attributes") or {}throughoutTrajectory Storage Update
json.dumps()before storage for consistencyTesting
Benefits
Summary by CodeRabbit
Release Notes
New Features
Improvements
✏️ Tip: You can customize this high-level summary in your review settings.