parlot.instrumentation.langgraph

 22def parlotize(
 23    agent_id: str,
 24    *,
 25    endpoint: Optional[str] = None,
 26    api_key: Optional[str] = None,
 27    capture_genai_content: Optional[bool] = None,
 28    service_name: Optional[str] = None,
 29    tracer_provider: TracerProvider | None = None,
 30    version: Optional[str] = None,
 31    channel: Optional[str] = None,
 32    modality: Optional[str] = None,
 33    capture_logs: bool | list[str] | None = None,
 34    log_level: Optional[str] = None,
 35) -> ParlotContext:
 36    """Parlotize LangGraph instrumentation and OTLP export.
 37
 38    Registers a global LangChain callback handler (LangSmith-style) so
 39    ``invoke`` / ``ainvoke`` / ``astream`` emit GenAI spans without per-call
 40    callbacks. When LiveKit already owns the active session, contract spans
 41    are suppressed and GenAI ops nest under the current OTel context (LiveKit
 42    stamps channel/modality). For LangGraph-owned sessions, pass ``channel``
 43    (e.g. ``webchat``) and optionally ``modality`` so ingest does not assume
 44    voice.
 45
 46    Shared parameters (``agent_id``, plus keyword-only ``endpoint``, ``api_key``,
 47    ``version``, ``capture_genai_content``, ``capture_logs``, ``log_level``,
 48    ``service_name``, ``tracer_provider``) match every adapter — see
 49    ``parlot.core.ParlotizeProtocol``.
 50
 51    Args:
 52        agent_id: Shared — required canonical ``session.agent_id``. Must be
 53            non-empty after stripping whitespace.
 54        endpoint: Shared — Parlot OTLP base URL (or ``PARLOT_ENDPOINT``).
 55        api_key: Shared — org API key (or ``PARLOT_API_KEY``).
 56        capture_genai_content: Shared — GenAI payload capture override.
 57            Precedence: this kwarg → Settings → Generative AI → on.
 58        service_name: Shared — OTel ``service.name`` (defaults to ``agent_id``
 59            or ``"langgraph-agent"``).
 60        tracer_provider: Shared — existing ``TracerProvider``. If LiveKit
 61            already initialized one in-process, this call adopts it.
 62        version: Shared — ``gen_ai.agent.version``. Pass ``version=`` to set it;
 63            otherwise ``\"unknown\"``.
 64        channel: Communication channel for standalone LangGraph sessions
 65            (e.g. ``"webchat"``, ``"slack"``, ``"sms"``). Defaults to
 66            ``"text"``. Ignored when LiveKit owns the session.
 67        modality: Session modality (``"text"``, ``"voice"``, or
 68            ``"multimodal"``). Defaults to ``"text"`` for standalone sessions.
 69        capture_logs: Shared — session log capture (bool or globs).
 70        log_level: Shared — minimum level for session log capture.
 71
 72    Returns:
 73        The ``ParlotContext`` created for this process (or the prior one if
 74        already configured).
 75    """
 76    global _configured, _configured_agent_id, _configured_agent_version, _parlot_context
 77    if _configured:
 78        logger.debug("parlot-instrumentation.langgraph already configured — skipping")
 79        assert _parlot_context is not None
 80        return _parlot_context
 81
 82    from parlot.core import base_parlotize
 83    from parlot.core.genai_content_capture import should_capture_genai_content
 84    from parlot.core.sdk_version import resolve_parlot_sdk_version
 85
 86    _configured_agent_version = (version or "").strip() or "unknown"
 87
 88    res = base_parlotize(
 89        agent_id,
 90        endpoint=endpoint,
 91        api_key=api_key,
 92        capture_genai_content=capture_genai_content,
 93        tracer_provider=tracer_provider,
 94        version=_configured_agent_version,
 95        capture_logs=capture_logs,
 96        log_level=log_level,
 97    )
 98    _parlot_context = res.context
 99    _configured_agent_id = res.agent_id
100
101    runtime = res.context.runtime
102    capture = should_capture_genai_content(
103        _configured_agent_id,
104        capture_genai_content_config=res.capture_genai_content,
105        bootstrap_globs=list(runtime.capture_genai_content_globs) if runtime else None,
106        bootstrap_agents=runtime.capture_genai_content_agents_map() if runtime else None,
107        bootstrap_present=bool(runtime and runtime.capture_genai_content_policy_present),
108    )
109
110    tracer_provider = res.tracer_provider
111    if tracer_provider is None:
112        if not res.endpoint:
113            raise ValueError(
114                "No OTLP endpoint configured. Pass endpoint= or set the "
115                "PARLOT_ENDPOINT environment variable."
116            )
117        tracer_provider = _build_provider(
118            endpoint=res.endpoint,
119            api_key=res.api_key,
120            service_name=service_name,
121            service_version=_configured_agent_version or None,
122        )
123
124    from opentelemetry import trace
125
126    try:
127        trace.set_tracer_provider(tracer_provider)
128    except Exception:
129        logger.debug("Could not set global tracer provider", exc_info=True)
130
131    tracer = tracer_provider.get_tracer(
132        "parlot.instrumentation.langgraph",
133        resolve_parlot_sdk_version() or None,
134    )
135
136    from ._callbacks import ParlotLangGraphCallbackHandler
137    from ._hooks import install_parlotize_hook
138    from ._session import set_channel_modality, set_identity, set_tracer
139
140    set_identity(_configured_agent_id, _configured_agent_version)
141    set_channel_modality(channel=channel or "", modality=modality or "")
142    set_tracer(tracer)
143    handler = ParlotLangGraphCallbackHandler(tracer, capture_genai_content=capture)
144    install_parlotize_hook(handler)
145
146    _configured = True
147    logger.debug(
148        "parlot-instrumentation.langgraph configured (endpoint=%s)", res.endpoint
149    )
150    return res.context

177def close_session(thread_id: str, *, reason: str = "completed") -> None:
178    """Public helper to close a LangGraph-owned session by thread_id.
179
180    Emits ``parlot.session.close`` with accumulated usage and turn counts.
181    Optional — sessions are also flushed on process exit.
182
183    Args:
184        thread_id: LangGraph thread id from
185            ``config={"configurable": {"thread_id": "..."}}``.
186        reason: Close reason stamped on ``session.close_reason``. Defaults to
187            ``"completed"``.
188    """
189    from ._session import close_session as _close
190
191    _close(thread_id, reason=reason)

129class ParlotLangGraphCallbackHandler(BaseCallbackHandler):
130    """LangChain callback handler that emits Parlot GenAI spans.
131
132    Translates LangGraph / LangChain execution events (LLM starts/ends, tool
133    invocations, chain runs) into OpenTelemetry GenAI spans
134    (``invoke_agent``, ``invoke_workflow``, ``chat``, ``execute_tool``).
135
136    ``parlotize()`` registers this handler via LangChain configuration hooks,
137    so manual ``callbacks=[...]`` attachment is not required. When LiveKit
138    owns the active session, contract spans are suppressed and GenAI ops nest
139    under the current OTel context.
140    """
141
142    raise_error = False
143
144    def __init__(
145        self,
146        tracer: Tracer,
147        *,
148        capture_genai_content: bool = True,
149    ) -> None:
150        self._tracer = tracer
151        self._capture_genai_content = capture_genai_content
152        self._spans: dict[str, Span] = {}
153        self._root_runs: set[str] = set()
154
155    @property
156    def ignore_chain(self) -> bool:
157        return False
158
159    @property
160    def ignore_llm(self) -> bool:
161        return False
162
163    @property
164    def ignore_agent(self) -> bool:
165        return False
166
167    def _run_key(self, run_id: UUID | None) -> str:
168        return str(run_id) if run_id else ""
169
170    def _active_langgraph_state(self) -> _LangGraphSessionState | None:
171        state = get_active_session()
172        if isinstance(state, _LangGraphSessionState):
173            return state
174        return None
175
176    def _contract_attrs(
177        self, state: _LangGraphSessionState | None
178    ) -> dict[str, AttributeValue]:
179        """Session/turn keys required by collector operational ingest."""
180        if state is None:
181            return {}
182        turn_index = state.open_agent_turn_index or state.turn_count
183        attrs: dict[str, AttributeValue] = {
184            ATTR_SESSION_ID: state.session_id,
185            ATTR_SESSION_CONVERSATION_ID: state.conversation_id,
186            ATTR_GEN_AI_CONVERSATION_ID: state.conversation_id,
187            ATTR_LG_THREAD_ID: state.thread_id,
188        }
189        if turn_index > 0:
190            attrs[ATTR_TURN_INDEX] = turn_index
191        return attrs
192
193    def _start(
194        self,
195        name: str,
196        run_id: UUID | None,
197        *,
198        attributes: dict[str, AttributeValue] | None = None,
199        parent_run_id: UUID | None = None,
200    ) -> Span | None:
201        key = self._run_key(run_id)
202        if not key:
203            return None
204        parent_ctx = None
205        if parent_run_id is not None:
206            parent = self._spans.get(self._run_key(parent_run_id))
207            if parent is not None:
208                parent_ctx = trace.set_span_in_context(parent)
209        span = self._tracer.start_span(name, context=parent_ctx, attributes=attributes or {})
210        self._spans[key] = span
211        return span
212
213    def _end(self, run_id: UUID | None, *, error: BaseException | None = None) -> None:
214        key = self._run_key(run_id)
215        span = self._spans.pop(key, None)
216        if span is None:
217            return
218        if error is not None:
219            span.set_status(Status(StatusCode.ERROR, str(error)[:500]))
220            span.record_exception(error)
221        else:
222            span.set_status(Status(StatusCode.OK))
223        span.end()
224
225    def _thread_id(self, **kwargs: Any) -> str:
226        metadata = kwargs.get("metadata") or {}
227        tags = kwargs.get("tags")
228        tid = thread_id_from_metadata(metadata if isinstance(metadata, dict) else {})
229        if tid:
230            return tid
231        # LangGraph puts thread_id under metadata["thread_id"] or configurable
232        cfg = kwargs.get("config")
233        if isinstance(cfg, dict):
234            configurable = cfg.get("configurable") or {}
235            if isinstance(configurable, dict) and configurable.get("thread_id"):
236                return str(configurable["thread_id"])
237        if isinstance(tags, list):
238            for tag in tags:
239                if isinstance(tag, str) and tag.startswith("thread_id:"):
240                    return tag.split(":", 1)[1]
241        return ""
242
243    def on_chain_start(
244        self,
245        serialized: dict[str, Any] | None,
246        inputs: dict[str, Any],
247        *,
248        run_id: UUID,
249        parent_run_id: UUID | None = None,
250        tags: list[str] | None = None,
251        metadata: dict[str, Any] | None = None,
252        name: str | None = None,
253        **kwargs: Any,
254    ) -> Any:
255        serialized = serialized or {}
256        chain_name = name or serialized.get("name") or serialized.get("id", ["chain"])[-1]
257        thread_id = self._thread_id(metadata=metadata, tags=tags, **kwargs)
258        state = ensure_session(thread_id) if thread_id or not livekit_owns_session() else None
259        if state is None and not livekit_owns_session():
260            state = ensure_session(thread_id or "")
261
262        is_root = parent_run_id is None
263        span_name = SPAN_GEN_AI_INVOKE_AGENT if is_root else SPAN_GEN_AI_INVOKE_WORKFLOW
264        op = GEN_AI_OP_INVOKE_AGENT if is_root else GEN_AI_OP_INVOKE_WORKFLOW
265        attrs: dict[str, AttributeValue] = {
266            ATTR_AGENT_FRAMEWORK: "langgraph",
267            ATTR_GEN_AI_OP_NAME: op,
268            ATTR_AGENT_ROLE: "pipeline" if is_root else "pipeline",
269            ATTR_AGENT_STAGE: "node" if not is_root else "turn",
270            ATTR_LG_GRAPH_NAME: str(chain_name),
271            ATTR_LG_RUN_ID: str(run_id),
272        }
273        if not is_root:
274            attrs[ATTR_LG_NODE_NAME] = str(chain_name)
275        if is_root:
276            self._root_runs.add(str(run_id))
277            if state is not None and not livekit_owns_session():
278                user_text = _last_message_text(
279                    _messages_from_payload(inputs),
280                    roles={"human", "user"},
281                )
282                user_idx = emit_turn(
283                    state,
284                    role="user",
285                    utterance_text=user_text,
286                )
287                state.open_agent_turn_index = user_idx + 1
288        attrs.update(self._contract_attrs(state))
289
290        self._start(
291            span_name,
292            run_id,
293            attributes=attrs,
294            parent_run_id=parent_run_id,
295        )
296
297    def on_chain_end(
298        self,
299        outputs: dict[str, Any],
300        *,
301        run_id: UUID,
302        parent_run_id: UUID | None = None,
303        **kwargs: Any,
304    ) -> Any:
305        key = str(run_id)
306        if key in self._root_runs:
307            state = self._active_langgraph_state()
308            if state is not None and not livekit_owns_session():
309                agent_text = _last_message_text(
310                    _messages_from_payload(outputs),
311                    roles={"ai", "assistant"},
312                )
313                emit_turn(
314                    state,
315                    role="agent",
316                    turn_index=state.open_agent_turn_index,
317                    utterance_text=agent_text,
318                )
319                state.open_agent_turn_index = None
320        self._end(run_id)
321        self._root_runs.discard(key)
322
323    def on_chain_error(
324        self,
325        error: BaseException,
326        *,
327        run_id: UUID,
328        parent_run_id: UUID | None = None,
329        **kwargs: Any,
330    ) -> Any:
331        key = str(run_id)
332        if key in self._root_runs:
333            state = self._active_langgraph_state()
334            if state is not None:
335                state.open_agent_turn_index = None
336        self._end(run_id, error=error)
337        self._root_runs.discard(key)
338
339    def on_llm_start(
340        self,
341        serialized: dict[str, Any] | None,
342        prompts: list[str],
343        *,
344        run_id: UUID,
345        parent_run_id: UUID | None = None,
346        tags: list[str] | None = None,
347        metadata: dict[str, Any] | None = None,
348        **kwargs: Any,
349    ) -> Any:
350        serialized = serialized or {}
351        metadata = metadata or {}
352        model = str(
353            metadata.get("ls_model_name")
354            or serialized.get("name")
355            or metadata.get("model")
356            or ""
357        ).strip()
358        provider = str(metadata.get("ls_provider") or "").strip()
359        attrs: dict[str, AttributeValue] = {
360            ATTR_AGENT_FRAMEWORK: "langgraph",
361            ATTR_GEN_AI_OP_NAME: GEN_AI_OP_CHAT,
362            ATTR_AGENT_ROLE: "llm",
363            ATTR_AGENT_STAGE: "node",
364            ATTR_LG_RUN_ID: str(run_id),
365        }
366        if model:
367            attrs[ATTR_GEN_AI_MODEL] = model
368        if provider:
369            attrs[ATTR_GEN_AI_PROVIDER] = provider
370        attrs.update(self._contract_attrs(self._active_langgraph_state()))
371        span = self._start(
372            span_name_chat(model or None),
373            run_id,
374            attributes=attrs,
375            parent_run_id=parent_run_id,
376        )
377        if span is not None and self._capture_genai_content and prompts:
378            span.add_event(
379                EVENT_GEN_AI_USER_MESSAGE,
380                {"content": prompts[-1][:4000]},
381            )
382
383    def on_chat_model_start(
384        self,
385        serialized: dict[str, Any] | None,
386        messages: list[list[Any]],
387        *,
388        run_id: UUID,
389        parent_run_id: UUID | None = None,
390        tags: list[str] | None = None,
391        metadata: dict[str, Any] | None = None,
392        **kwargs: Any,
393    ) -> Any:
394        prompts: list[str] = []
395        for batch in messages or []:
396            for msg in batch:
397                content = getattr(msg, "content", None)
398                if isinstance(content, str) and content:
399                    prompts.append(content)
400        self.on_llm_start(
401            serialized,
402            prompts,
403            run_id=run_id,
404            parent_run_id=parent_run_id,
405            tags=tags,
406            metadata=metadata,
407            **kwargs,
408        )
409
410    def on_llm_end(
411        self,
412        response: Any,
413        *,
414        run_id: UUID,
415        parent_run_id: UUID | None = None,
416        **kwargs: Any,
417    ) -> Any:
418        span = self._spans.get(self._run_key(run_id))
419        if span is not None:
420            usage = _usage_from_llm_result(response)
421            if usage.get("input"):
422                span.set_attribute(ATTR_GEN_AI_IN_TOKENS, usage["input"])
423            if usage.get("output"):
424                span.set_attribute(ATTR_GEN_AI_OUT_TOKENS, usage["output"])
425            if self._capture_genai_content:
426                text = _text_from_llm_result(response)
427                if text:
428                    span.add_event(
429                        EVENT_GEN_AI_ASSISTANT_MESSAGE,
430                        {"content": text[:4000]},
431                    )
432        self._end(run_id)
433
434    def on_llm_error(
435        self,
436        error: BaseException,
437        *,
438        run_id: UUID,
439        parent_run_id: UUID | None = None,
440        **kwargs: Any,
441    ) -> Any:
442        self._end(run_id, error=error)
443
444    def on_tool_start(
445        self,
446        serialized: dict[str, Any] | None,
447        input_str: str,
448        *,
449        run_id: UUID,
450        parent_run_id: UUID | None = None,
451        tags: list[str] | None = None,
452        metadata: dict[str, Any] | None = None,
453        inputs: dict[str, Any] | None = None,
454        **kwargs: Any,
455    ) -> Any:
456        serialized = serialized or {}
457        tool_name = str(
458            serialized.get("name") or kwargs.get("name") or "tool"
459        ).strip() or "tool"
460        attrs: dict[str, AttributeValue] = {
461            ATTR_AGENT_FRAMEWORK: "langgraph",
462            ATTR_GEN_AI_OP_NAME: GEN_AI_OP_EXECUTE_TOOL,
463            ATTR_AGENT_ROLE: "tool",
464            ATTR_AGENT_STAGE: "call",
465            ATTR_AGENT_TOOL_NAME: tool_name,
466            ATTR_GEN_AI_TOOL_NAME: tool_name,
467            ATTR_LG_RUN_ID: str(run_id),
468        }
469        attrs.update(self._contract_attrs(self._active_langgraph_state()))
470        span = self._start(
471            span_name_execute_tool(tool_name),
472            run_id,
473            attributes=attrs,
474            parent_run_id=parent_run_id,
475        )
476        if span is not None and self._capture_genai_content:
477            payload = ""
478            if inputs is not None:
479                try:
480                    import json
481
482                    payload = json.dumps(inputs, default=str)
483                except Exception:
484                    payload = str(inputs)
485            elif input_str:
486                payload = str(input_str)
487            if payload:
488                trimmed = payload[:8192]
489                span.set_attribute(ATTR_TOOL_INPUT_PAYLOAD, trimmed)
490                span.set_attribute(
491                    ATTR_TOOL_INPUT_PAYLOAD_PREVIEW,
492                    trimmed[:512],
493                )
494                span.add_event(
495                    EVENT_GEN_AI_TOOL_MESSAGE,
496                    {"content": trimmed[:4000], "role": "tool_input"},
497                )
498
499    def on_tool_end(
500        self,
501        output: Any,
502        *,
503        run_id: UUID,
504        parent_run_id: UUID | None = None,
505        **kwargs: Any,
506    ) -> Any:
507        span = self._spans.get(self._run_key(run_id))
508        if span is not None and self._capture_genai_content and output is not None:
509            text = str(output)
510            if text:
511                preview = text[:512]
512                span.set_attribute(ATTR_TOOL_OUTPUT_PAYLOAD_PREVIEW, preview)
513                span.add_event(
514                    EVENT_GEN_AI_TOOL_MESSAGE,
515                    {"content": text[:4000], "role": "tool"},
516                )
517        self._end(run_id)
518
519    def on_tool_error(
520        self,
521        error: BaseException,
522        *,
523        run_id: UUID,
524        parent_run_id: UUID | None = None,
525        **kwargs: Any,
526    ) -> Any:
527        span = self._spans.get(self._run_key(run_id))
528        if span is not None:
529            span.set_attribute(ATTR_AGENT_TOOL_IS_ERROR, True)
530        self._end(run_id, error=error)