fix(genai-instrumentors): cleanup code, align with OTel GenAI semconv, add missing attributes and fix deprecated usage (#706)

*Description of changes:*

Further aligns GenAI instrumentors with the OTel GenAI semantic
conventions spec.

- Use `SpanKind.CLIENT` for inference spans in LlamaIndex
instrumentation
- Fix `retrieval` operation span in LlamaIndex instrumentation
- Remove `gen_ai.system_instructions` from `invoke_agent` span
- Remove `on_agent_action`/`on_agent_finish` from LangChain as
on_chain_start will capture these spans
- add `gen_ai.response.finish_reasons`  from model responses 
- add LLM request params: `frequency_penalty`, `presence_penalty`,
`stop_sequences`, `top_p`, `top_k` across all instrumentors
- add `ToolCallRequestPart` and `ReasoningPart` in LlamaIndex output
messages
- fixed `error.message` to `error.type` in CrewAI per OTel spec
- changed `gen_ai.prompt` to `gen_ai.input.messages` in LangChain per
OTel spec
- changed `gen_ai.prompt` to `gen_ai.prompt.name` in MCP per OTel spec
- add `record_exception()` to LangChain error path

*Tests:*
- Contract tests: replace hardcoded strings with OTel semconv imports
- Contract tests: assert `gen_ai.system_instructions` on all chat spans
- add provider subtests for finish_reasons
- add OTel JSON schema validation for ToolCallRequestPart and
ReasoningPart


By submitting this pull request, I confirm that you can use, modify,
copy, and redistribute this contribution, under the terms of your
choice.

Author: liustve

CHANGELOG.md

@@ -12,6 +12,8 @@ If your change does not need a CHANGELOG entry, add the "skip changelog" label t
 
 ## Unreleased
 
+- fix(genai-instrumentors): cleanup code, align with OTel GenAI semconv, add missing attributes and fix deprecated usage
+  ([#706](https://github.com/aws-observability/aws-otel-python-instrumentation/pull/706))
 - feat(genai-instrumentation): add oldest/latest dependency testing and scheduled instrumentation tests for GenAI libraries
   ([#708](https://github.com/aws-observability/aws-otel-python-instrumentation/pull/708))
 - feat(instrumentors): add invoke_workflow span support for LlamaIndex AgentWorkflow and CrewAI

aws-opentelemetry-distro/src/amazon/opentelemetry/distro/instrumentation/crewai/_event_handler.py

@@ -14,7 +14,6 @@
     skip_instrumentation_if_suppressed,
 )
 from opentelemetry import context, trace
-from opentelemetry.semconv._incubating.attributes.error_attributes import ERROR_MESSAGE
 from opentelemetry.semconv._incubating.attributes.gen_ai_attributes import (
     GEN_AI_AGENT_DESCRIPTION,
     GEN_AI_AGENT_ID,
@@ -23,9 +22,14 @@
     GEN_AI_OPERATION_NAME,
     GEN_AI_OUTPUT_MESSAGES,
     GEN_AI_PROVIDER_NAME,
+    GEN_AI_REQUEST_FREQUENCY_PENALTY,
     GEN_AI_REQUEST_MAX_TOKENS,
     GEN_AI_REQUEST_MODEL,
+    GEN_AI_REQUEST_PRESENCE_PENALTY,
+    GEN_AI_REQUEST_STOP_SEQUENCES,
     GEN_AI_REQUEST_TEMPERATURE,
+    GEN_AI_REQUEST_TOP_P,
+    GEN_AI_RESPONSE_FINISH_REASONS,
     GEN_AI_RESPONSE_MODEL,
     GEN_AI_SYSTEM_INSTRUCTIONS,
     GEN_AI_TOOL_CALL_ARGUMENTS,
@@ -38,6 +42,7 @@
     GEN_AI_USAGE_OUTPUT_TOKENS,
     GenAiOperationNameValues,
 )
+from opentelemetry.semconv.attributes.error_attributes import ERROR_TYPE
 from opentelemetry.trace import SpanKind, Status, StatusCode
 
 if TYPE_CHECKING:
@@ -201,18 +206,18 @@ def _on_crew_start(self, source: "Crew", event: "CrewKickoffStartedEvent") -> No
         self._start_span(span_name, event.event_id, attributes, event.parent_event_id)
 
     def _on_crew_completed(
-        self, source: "Crew", event: "CrewKickoffCompletedEvent"  # pylint: disable=unused-argument
-    ) -> None:
+        self, source: "Crew", event: "CrewKickoffCompletedEvent"
+    ) -> None:  # pylint: disable=unused-argument
         self._end_span(event.started_event_id)
         self._event_id_to_span.clear()
 
     def _on_crew_failed(
-        self, source: "Crew", event: "CrewKickoffFailedEvent"  # pylint: disable=unused-argument
-    ) -> None:
+        self, source: "Crew", event: "CrewKickoffFailedEvent"
+    ) -> None:  # pylint: disable=unused-argument
         self._end_span(event.started_event_id, error=getattr(event, "error", None))
         self._event_id_to_span.clear()
 
-    def _on_agent_start(
+    def _on_agent_start(  # pylint: disable=too-many-branches
         self, source: "BaseAgent", event: "AgentExecutionStartedEvent"  # pylint: disable=unused-argument
     ) -> None:
         agent = getattr(event, "agent", None)
@@ -233,9 +238,6 @@ def _on_agent_start(
             goal = getattr(agent, "goal", None)
             if goal:
                 attributes[GEN_AI_AGENT_DESCRIPTION] = goal
-            backstory = getattr(agent, "backstory", None)
-            if backstory:
-                attributes[GEN_AI_SYSTEM_INSTRUCTIONS] = backstory
 
             llm = getattr(agent, "llm", None)
             provider, model = self._extract_provider_and_model(llm)
@@ -244,12 +246,18 @@ def _on_agent_start(
             if model:
                 attributes[GEN_AI_REQUEST_MODEL] = model
             if llm:
-                temperature = getattr(llm, "temperature", None)
-                if temperature is not None:
-                    attributes[GEN_AI_REQUEST_TEMPERATURE] = temperature
-                max_tokens = getattr(llm, "max_tokens", None)
-                if max_tokens is not None:
-                    attributes[GEN_AI_REQUEST_MAX_TOKENS] = max_tokens
+                if getattr(llm, "temperature", None) is not None:
+                    attributes[GEN_AI_REQUEST_TEMPERATURE] = llm.temperature
+                if getattr(llm, "max_tokens", None) is not None:
+                    attributes[GEN_AI_REQUEST_MAX_TOKENS] = llm.max_tokens
+                if getattr(llm, "top_p", None) is not None:
+                    attributes[GEN_AI_REQUEST_TOP_P] = llm.top_p
+                if getattr(llm, "frequency_penalty", None) is not None:
+                    attributes[GEN_AI_REQUEST_FREQUENCY_PENALTY] = llm.frequency_penalty
+                if getattr(llm, "presence_penalty", None) is not None:
+                    attributes[GEN_AI_REQUEST_PRESENCE_PENALTY] = llm.presence_penalty
+                if getattr(llm, "stop", None):
+                    attributes[GEN_AI_REQUEST_STOP_SEQUENCES] = llm.stop
 
         self._start_span(span_name, event.event_id, attributes, event.parent_event_id)
 
@@ -259,8 +267,8 @@ def _on_agent_completed(
         self._end_span(event.started_event_id)
 
     def _on_agent_failed(
-        self, source: "BaseAgent", event: "AgentExecutionErrorEvent"  # pylint: disable=unused-argument
-    ) -> None:
+        self, source: "BaseAgent", event: "AgentExecutionErrorEvent"
+    ) -> None:  # pylint: disable=unused-argument
         self._end_span(event.started_event_id, error=getattr(event, "error", None))
 
     def _on_tool_start(self, source: "ToolUsage", event: "ToolUsageStartedEvent") -> None:
@@ -296,8 +304,8 @@ def _on_tool_start(self, source: "ToolUsage", event: "ToolUsageStartedEvent") ->
         self._start_span(span_name, event.event_id, attributes, event.parent_event_id)
 
     def _on_tool_finished(
-        self, source: "ToolUsage", event: "ToolUsageFinishedEvent"  # pylint: disable=unused-argument
-    ) -> None:
+        self, source: "ToolUsage", event: "ToolUsageFinishedEvent"
+    ) -> None:  # pylint: disable=unused-argument
         attrs: Dict[str, Any] = {}
         output = getattr(event, "output", None)
         if output is not None:
@@ -344,6 +352,7 @@ def _on_llm_completed(self, source: "LLM", event: "LLMCallCompletedEvent") -> No
         from crewai.events.types.llm_events import LLMCallType  # pylint: disable=import-outside-toplevel
 
         finish_reason = "tool_call" if event.call_type == LLMCallType.TOOL_CALL else "stop"
+        attrs[GEN_AI_RESPONSE_FINISH_REASONS] = [finish_reason]
 
         response = event.response
         if response:
@@ -412,7 +421,7 @@ def _end_span(
                         entry.span.set_attribute(key, value)
             if error:
                 entry.span.set_status(Status(StatusCode.ERROR, error))
-                entry.span.set_attribute(ERROR_MESSAGE, error)
+                entry.span.set_attribute(ERROR_TYPE, error)
             else:
                 entry.span.set_status(Status(StatusCode.OK))
             entry.span.end()

aws-opentelemetry-distro/src/amazon/opentelemetry/distro/instrumentation/langchain/callback_handler.py

@@ -23,12 +23,16 @@
     GEN_AI_INPUT_MESSAGES,
     GEN_AI_OPERATION_NAME,
     GEN_AI_OUTPUT_MESSAGES,
-    GEN_AI_PROMPT,
     GEN_AI_PROVIDER_NAME,
+    GEN_AI_REQUEST_FREQUENCY_PENALTY,
     GEN_AI_REQUEST_MAX_TOKENS,
     GEN_AI_REQUEST_MODEL,
+    GEN_AI_REQUEST_PRESENCE_PENALTY,
+    GEN_AI_REQUEST_STOP_SEQUENCES,
     GEN_AI_REQUEST_TEMPERATURE,
+    GEN_AI_REQUEST_TOP_K,
     GEN_AI_REQUEST_TOP_P,
+    GEN_AI_RESPONSE_FINISH_REASONS,
     GEN_AI_RESPONSE_ID,
     GEN_AI_RESPONSE_MODEL,
     GEN_AI_SYSTEM_INSTRUCTIONS,
@@ -49,7 +53,6 @@
 from opentelemetry.util.types import AttributeValue
 
 if TYPE_CHECKING:
-    from langchain_core.agents import AgentAction, AgentFinish
     from langchain_core.messages import BaseMessage
     from langchain_core.outputs import LLMResult
 
@@ -152,7 +155,11 @@ def on_llm_start(
         self._set_langgraph_span_attributes(span, metadata)
         self._set_span_attribute(span, GEN_AI_PROVIDER_NAME, provider)
         self._set_span_attribute(span, GEN_AI_OPERATION_NAME, GenAiOperationNameValues.TEXT_COMPLETION.value)
-        self._set_span_attribute(span, GEN_AI_PROMPT, serialize_to_json_string(prompts))
+        self._set_span_attribute(
+            span,
+            GEN_AI_INPUT_MESSAGES,
+            serialize_to_json_string([{"role": "user", "parts": [{"type": "text", "content": p}]} for p in prompts]),
+        )
         self._set_llm_request_span_attributes(span, kwargs, serialized=serialized_kwargs, model_name=model_name)
 
     @skip_instrumentation_if_suppressed
@@ -174,9 +181,11 @@ def on_llm_end(self, response: LLMResult, *, run_id: UUID, **kwargs: Any) -> Non
         )
 
         if response.generations:
-            self._set_span_attribute(
-                span, GEN_AI_OUTPUT_MESSAGES, serialize_to_json_string(self._format_lc_llm_output(response.generations))
-            )
+            output_messages = self._format_lc_llm_output(response.generations)
+            self._set_span_attribute(span, GEN_AI_OUTPUT_MESSAGES, serialize_to_json_string(output_messages))
+            finish_reasons = self._extract_finish_reasons(response.generations)
+            if finish_reasons is not None:
+                self._set_span_attribute(span, GEN_AI_RESPONSE_FINISH_REASONS, finish_reasons)
 
         self._set_span_attribute(span, GEN_AI_RESPONSE_MODEL, model)
         self._set_span_attribute(span, GEN_AI_RESPONSE_ID, response_id)
@@ -283,32 +292,6 @@ def on_tool_end(self, output: Any, *, run_id: UUID, **kwargs: Any) -> None:
     def on_tool_error(self, error: BaseException, *, run_id: UUID, **kwargs: Any) -> None:
         self._handle_error(error, run_id, **kwargs)
 
-    @skip_instrumentation_if_suppressed
-    def on_agent_action(self, action: AgentAction, *, run_id: UUID, **kwargs: Any) -> None:
-
-        entry = self._safe_get_span(run_id)
-        if not entry:
-            return
-        span, _ = entry
-        self._set_span_attribute(
-            span, GEN_AI_TOOL_CALL_ARGUMENTS, serialize_to_json_string(getattr(action, "tool_input", None))
-        )
-        self._set_span_attribute(span, GEN_AI_TOOL_NAME, getattr(action, "tool", None))
-        self._set_span_attribute(span, GEN_AI_OPERATION_NAME, GenAiOperationNameValues.INVOKE_AGENT.value)
-
-    @skip_instrumentation_if_suppressed
-    def on_agent_finish(self, finish: AgentFinish, *, run_id: UUID, **kwargs: Any) -> None:
-
-        entry = self._safe_get_span(run_id)
-        if not entry:
-            return
-        span, _ = entry
-        self._set_span_attribute(
-            span,
-            GEN_AI_TOOL_CALL_RESULT,
-            serialize_to_json_string(finish.return_values.get("output")),
-        )
-
     def on_agent_error(self, error: BaseException, *, run_id: UUID, **kwargs: Any) -> None:
         self._handle_error(error, run_id, **kwargs)
 
@@ -470,6 +453,18 @@ def _set_llm_request_span_attributes(
             span, GEN_AI_REQUEST_TEMPERATURE, params.get("temperature") or config.get("temperature")
         )
         self._set_span_attribute(span, GEN_AI_REQUEST_TOP_P, params.get("top_p") or config.get("top_p"))
+        self._set_span_attribute(span, GEN_AI_REQUEST_TOP_K, params.get("top_k") or config.get("top_k"))
+        self._set_span_attribute(
+            span,
+            GEN_AI_REQUEST_FREQUENCY_PENALTY,
+            params.get("frequency_penalty") or config.get("frequency_penalty"),
+        )
+        self._set_span_attribute(
+            span, GEN_AI_REQUEST_PRESENCE_PENALTY, params.get("presence_penalty") or config.get("presence_penalty")
+        )
+        stop = params.get("stop") or config.get("stop")
+        if stop:
+            self._set_span_attribute(span, GEN_AI_REQUEST_STOP_SEQUENCES, stop)
 
     def _should_skip_chain(
         self, serialized: dict[str, Any], name: Optional[str], metadata: Optional[dict] = None
@@ -512,6 +507,7 @@ def _handle_error(self, error: BaseException, run_id: UUID, **kwargs: Any) -> No
         if not entry:
             return
         span, _ = entry
+        span.record_exception(error)
         span.set_status(Status(StatusCode.ERROR, str(error)))
         span.set_attribute(ERROR_TYPE, type(error).__qualname__)
         self._end_span(run_id)
@@ -553,6 +549,18 @@ def _get_name_from_callback(serialized: dict[str, Any], **kwargs: Any) -> Option
                 return ids[-1]
         return kwargs.get("name")
 
+    @staticmethod
+    def _extract_finish_reasons(generations: list) -> list[str] | None:
+        reasons: list[str] = []
+        for batch in generations:
+            for gen in batch:
+                raw = (getattr(gen, "generation_info", None) or {}).get("finish_reason")
+                if not raw and (msg := getattr(gen, "message", None)):
+                    raw = (getattr(msg, "response_metadata", None) or {}).get("stop_reason")
+                if raw:
+                    reasons.append(raw)
+        return reasons or None
+
     @staticmethod
     def _extract_llm_model_id(kwargs: dict, serialized: Optional[dict] = None) -> Optional[str]:
         config = serialized or {}

aws-opentelemetry-distro/src/amazon/opentelemetry/distro/instrumentation/llama_index/_handler.py

@@ -150,11 +150,13 @@ def new_span(
                 parent_id=parent_span_id,
             )
 
+        kind = SpanKind.CLIENT if isinstance(instance, (BaseLLM, MultiModalLLM, BaseEmbedding)) else SpanKind.INTERNAL
+
         otel_span = self._otel_tracer.start_span(
             name="llama_index.operation",  # generic operation name, updated in span.end()
             start_time=time_ns(),
             attributes={},
-            kind=SpanKind.INTERNAL,
+            kind=kind,
             context=(
                 parent.context
                 if parent