diff --git a/docs/instrumentation-genai/crewai.rst b/docs/instrumentation-genai/crewai.rst new file mode 100644 index 0000000000..c87a625ffd --- /dev/null +++ b/docs/instrumentation-genai/crewai.rst @@ -0,0 +1,7 @@ +OpenTelemetry Python - CrewAI Instrumentation +============================================== + +.. automodule:: opentelemetry.instrumentation.crewai + :members: + :undoc-members: + :show-inheritance: diff --git a/instrumentation-genai/opentelemetry-instrumentation-crewai/.changelog/4860.added b/instrumentation-genai/opentelemetry-instrumentation-crewai/.changelog/4860.added new file mode 100644 index 0000000000..b0484e4b58 --- /dev/null +++ b/instrumentation-genai/opentelemetry-instrumentation-crewai/.changelog/4860.added @@ -0,0 +1 @@ +`opentelemetry-instrumentation-crewai`: add initial instrumentation for CrewAI diff --git a/instrumentation-genai/opentelemetry-instrumentation-crewai/CHANGELOG.md b/instrumentation-genai/opentelemetry-instrumentation-crewai/CHANGELOG.md new file mode 100644 index 0000000000..a6dc2b8c20 --- /dev/null +++ b/instrumentation-genai/opentelemetry-instrumentation-crewai/CHANGELOG.md @@ -0,0 +1,16 @@ +# Changelog + +All notable changes to this project will be documented in this file. + + + +The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), +and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). + + diff --git a/instrumentation-genai/opentelemetry-instrumentation-crewai/LICENSE b/instrumentation-genai/opentelemetry-instrumentation-crewai/LICENSE new file mode 100644 index 0000000000..261eeb9e9f --- /dev/null +++ b/instrumentation-genai/opentelemetry-instrumentation-crewai/LICENSE @@ -0,0 +1,201 @@ + Apache License + Version 2.0, January 2004 + http://www.apache.org/licenses/ + + TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION + + 1. Definitions. + + "License" shall mean the terms and conditions for use, reproduction, + and distribution as defined by Sections 1 through 9 of this document. + + "Licensor" shall mean the copyright owner or entity authorized by + the copyright owner that is granting the License. + + "Legal Entity" shall mean the union of the acting entity and all + other entities that control, are controlled by, or are under common + control with that entity. For the purposes of this definition, + "control" means (i) the power, direct or indirect, to cause the + direction or management of such entity, whether by contract or + otherwise, or (ii) ownership of fifty percent (50%) or more of the + outstanding shares, or (iii) beneficial ownership of such entity. + + "You" (or "Your") shall mean an individual or Legal Entity + exercising permissions granted by this License. + + "Source" form shall mean the preferred form for making modifications, + including but not limited to software source code, documentation + source, and configuration files. + + "Object" form shall mean any form resulting from mechanical + transformation or translation of a Source form, including but + not limited to compiled object code, generated documentation, + and conversions to other media types. + + "Work" shall mean the work of authorship, whether in Source or + Object form, made available under the License, as indicated by a + copyright notice that is included in or attached to the work + (an example is provided in the Appendix below). + + "Derivative Works" shall mean any work, whether in Source or Object + form, that is based on (or derived from) the Work and for which the + editorial revisions, annotations, elaborations, or other modifications + represent, as a whole, an original work of authorship. For the purposes + of this License, Derivative Works shall not include works that remain + separable from, or merely link (or bind by name) to the interfaces of, + the Work and Derivative Works thereof. + + "Contribution" shall mean any work of authorship, including + the original version of the Work and any modifications or additions + to that Work or Derivative Works thereof, that is intentionally + submitted to Licensor for inclusion in the Work by the copyright owner + or by an individual or Legal Entity authorized to submit on behalf of + the copyright owner. For the purposes of this definition, "submitted" + means any form of electronic, verbal, or written communication sent + to the Licensor or its representatives, including but not limited to + communication on electronic mailing lists, source code control systems, + and issue tracking systems that are managed by, or on behalf of, the + Licensor for the purpose of discussing and improving the Work, but + excluding communication that is conspicuously marked or otherwise + designated in writing by the copyright owner as "Not a Contribution." + + "Contributor" shall mean Licensor and any individual or Legal Entity + on behalf of whom a Contribution has been received by Licensor and + subsequently incorporated within the Work. + + 2. Grant of Copyright License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + copyright license to reproduce, prepare Derivative Works of, + publicly display, publicly perform, sublicense, and distribute the + Work and such Derivative Works in Source or Object form. + + 3. Grant of Patent License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + (except as stated in this section) patent license to make, have made, + use, offer to sell, sell, import, and otherwise transfer the Work, + where such license applies only to those patent claims licensable + by such Contributor that are necessarily infringed by their + Contribution(s) alone or by combination of their Contribution(s) + with the Work to which such Contribution(s) was submitted. If You + institute patent litigation against any entity (including a + cross-claim or counterclaim in a lawsuit) alleging that the Work + or a Contribution incorporated within the Work constitutes direct + or contributory patent infringement, then any patent licenses + granted to You under this License for that Work shall terminate + as of the date such litigation is filed. + + 4. Redistribution. You may reproduce and distribute copies of the + Work or Derivative Works thereof in any medium, with or without + modifications, and in Source or Object form, provided that You + meet the following conditions: + + (a) You must give any other recipients of the Work or + Derivative Works a copy of this License; and + + (b) You must cause any modified files to carry prominent notices + stating that You changed the files; and + + (c) You must retain, in the Source form of any Derivative Works + that You distribute, all copyright, patent, trademark, and + attribution notices from the Source form of the Work, + excluding those notices that do not pertain to any part of + the Derivative Works; and + + (d) If the Work includes a "NOTICE" text file as part of its + distribution, then any Derivative Works that You distribute must + include a readable copy of the attribution notices contained + within such NOTICE file, excluding those notices that do not + pertain to any part of the Derivative Works, in at least one + of the following places: within a NOTICE text file distributed + as part of the Derivative Works; within the Source form or + documentation, if provided along with the Derivative Works; or, + within a display generated by the Derivative Works, if and + wherever such third-party notices normally appear. The contents + of the NOTICE file are for informational purposes only and + do not modify the License. You may add Your own attribution + notices within Derivative Works that You distribute, alongside + or as an addendum to the NOTICE text from the Work, provided + that such additional attribution notices cannot be construed + as modifying the License. + + You may add Your own copyright statement to Your modifications and + may provide additional or different license terms and conditions + for use, reproduction, or distribution of Your modifications, or + for any such Derivative Works as a whole, provided Your use, + reproduction, and distribution of the Work otherwise complies with + the conditions stated in this License. + + 5. Submission of Contributions. Unless You explicitly state otherwise, + any Contribution intentionally submitted for inclusion in the Work + by You to the Licensor shall be under the terms and conditions of + this License, without any additional terms or conditions. + Notwithstanding the above, nothing herein shall supersede or modify + the terms of any separate license agreement you may have executed + with Licensor regarding such Contributions. + + 6. Trademarks. This License does not grant permission to use the trade + names, trademarks, service marks, or product names of the Licensor, + except as required for reasonable and customary use in describing the + origin of the Work and reproducing the content of the NOTICE file. + + 7. Disclaimer of Warranty. Unless required by applicable law or + agreed to in writing, Licensor provides the Work (and each + Contributor provides its Contributions) on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + implied, including, without limitation, any warranties or conditions + of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A + PARTICULAR PURPOSE. You are solely responsible for determining the + appropriateness of using or redistributing the Work and assume any + risks associated with Your exercise of permissions under this License. + + 8. Limitation of Liability. In no event and under no legal theory, + whether in tort (including negligence), contract, or otherwise, + unless required by applicable law (such as deliberate and grossly + negligent acts) or agreed to in writing, shall any Contributor be + liable to You for damages, including any direct, indirect, special, + incidental, or consequential damages of any character arising as a + result of this License or out of the use or inability to use the + Work (including but not limited to damages for loss of goodwill, + work stoppage, computer failure or malfunction, or any and all + other commercial damages or losses), even if such Contributor + has been advised of the possibility of such damages. + + 9. Accepting Warranty or Additional Liability. While redistributing + the Work or Derivative Works thereof, You may choose to offer, + and charge a fee for, acceptance of support, warranty, indemnity, + or other liability obligations and/or rights consistent with this + License. However, in accepting such obligations, You may act only + on Your own behalf and on Your sole responsibility, not on behalf + of any other Contributor, and only if You agree to indemnify, + defend, and hold each Contributor harmless for any liability + incurred by, or claims asserted against, such Contributor by reason + of your accepting any such warranty or additional liability. + + END OF TERMS AND CONDITIONS + + APPENDIX: How to apply the Apache License to your work. + + To apply the Apache License to your work, attach the following + boilerplate notice, with the fields enclosed by brackets "[]" + replaced with your own identifying information. (Don't include + the brackets!) The text should be enclosed in the appropriate + comment syntax for the file format. We also recommend that a + file or class name and description of purpose be included on the + same "printed page" as the copyright notice for easier + identification within third-party archives. + + Copyright [yyyy] [name of copyright owner] + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. diff --git a/instrumentation-genai/opentelemetry-instrumentation-crewai/README.rst b/instrumentation-genai/opentelemetry-instrumentation-crewai/README.rst new file mode 100644 index 0000000000..f0f8c5ad02 --- /dev/null +++ b/instrumentation-genai/opentelemetry-instrumentation-crewai/README.rst @@ -0,0 +1,102 @@ +OpenTelemetry CrewAI Instrumentation +===================================== + +|pypi| + +.. |pypi| image:: https://badge.fury.io/py/opentelemetry-instrumentation-crewai.svg + :target: https://pypi.org/project/opentelemetry-instrumentation-crewai/ + +This library instruments `CrewAI `_, a +framework for orchestrating multi-agent LLM workflows. It produces one +``invoke_workflow`` span per crew run, one ``invoke_agent`` span per +agent/task pair, and ``chat``/``execute_tool`` spans for every LLM call and +tool call an agent makes along the way. + +When one agent's task output feeds into another agent's task, that handoff +is represented as an OpenTelemetry span **link** between the two agent +invocations, not as a parent-child relationship -- the two agents ran one +after another, not one inside the other, and nesting them would say +something that isn't true about the timeline. + +Installation +------------ + +If your application is already instrumented with OpenTelemetry, add this +package to your requirements. +:: + + pip install opentelemetry-instrumentation-crewai + +If you don't have a CrewAI application yet, try our `examples `_, +which only need a valid LLM provider API key. + +Check out `zero-code example `_ for a quick start. + +Usage +----- + +This section describes how to set up CrewAI instrumentation if you're +setting OpenTelemetry up manually. Check out the +`manual example `_ for more details. + +.. code-block:: python + + from crewai import Agent, Crew, Task + from opentelemetry.instrumentation.crewai import CrewAIInstrumentor + + CrewAIInstrumentor().instrument() + + crew = Crew(agents=[...], tasks=[...]) + crew.kickoff() + +Enabling message content +************************* + +Prompt text, response text, and tool call arguments/results are not +captured by default -- only shapes and numbers (model name, token counts, +tool name, whether it errored). To capture message content, two settings +have to be enabled together: + +- ``OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT`` set to + ``span_only``, ``event_only``, or ``span_and_event``. +- ``OTEL_SEMCONV_STABILITY_OPT_IN`` set to ``gen_ai_latest_experimental`` + (see below). Without it, message content is not captured regardless of + the setting above -- this instrumentation stays on the stable v1.30.0 + semantic conventions, which do not carry message content. + +Tool call arguments and results are the one exception: they're gated only +by ``OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT`` (any truthy +value, including plain ``true``), independent of the experimental opt-in. + +See the `opentelemetry-util-genai README +`_ +for the full list of GenAI configuration variables. + +Enabling the latest experimental features +******************************************* + +To enable the latest experimental features, including message content +capture, set the environment variable ``OTEL_SEMCONV_STABILITY_OPT_IN`` to +``gen_ai_latest_experimental``. + +.. note:: Generative AI semantic conventions are still evolving. The latest experimental features will introduce breaking changes in future releases. + +Uninstrument +************ + +To uninstrument, call the uninstrument method: + +.. code-block:: python + + from opentelemetry.instrumentation.crewai import CrewAIInstrumentor + + CrewAIInstrumentor().instrument() + # ... + + CrewAIInstrumentor().uninstrument() + +References +---------- +* `signoz#11663 `_ -- the request this instrumentation was built against +* `OpenTelemetry Project `_ +* `OpenTelemetry Python Examples `_ diff --git a/instrumentation-genai/opentelemetry-instrumentation-crewai/examples/manual/.env.example b/instrumentation-genai/opentelemetry-instrumentation-crewai/examples/manual/.env.example new file mode 100644 index 0000000000..4badcefdb2 --- /dev/null +++ b/instrumentation-genai/opentelemetry-instrumentation-crewai/examples/manual/.env.example @@ -0,0 +1,14 @@ +# Update this with your real OpenAI API key +OPENAI_API_KEY=sk-YOUR_API_KEY + +# Uncomment and change to your OTLP endpoint +# OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4317 +# OTEL_EXPORTER_OTLP_PROTOCOL=grpc + +OTEL_SERVICE_NAME=opentelemetry-python-crewai + +# Remove to hide prompt and completion content +OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT=span_only + +# Required for the capture-content setting above to actually take effect. +OTEL_SEMCONV_STABILITY_OPT_IN=gen_ai_latest_experimental diff --git a/instrumentation-genai/opentelemetry-instrumentation-crewai/examples/manual/README.rst b/instrumentation-genai/opentelemetry-instrumentation-crewai/examples/manual/README.rst new file mode 100644 index 0000000000..f422abc653 --- /dev/null +++ b/instrumentation-genai/opentelemetry-instrumentation-crewai/examples/manual/README.rst @@ -0,0 +1,48 @@ +OpenTelemetry CrewAI Instrumentation Example +============================================= + +This is an example of how to instrument CrewAI when configuring the +OpenTelemetry SDK and this instrumentation manually. + +When `main.py `_ is run, it builds a two-agent crew (a researcher +with a word-count tool, and a writer who summarizes the researcher's +output) and exports traces to an OTLP compatible endpoint. The trace +includes one ``invoke_workflow`` span for the whole run, one +``invoke_agent`` span per agent, and ``chat``/``execute_tool`` spans +underneath. The writer's ``invoke_agent`` span carries a **link** back to +the researcher's, representing the handoff between them. + +Note: `.env.example <.env.example>`_ configures additional environment +variables: + +- ``OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT=span_only`` captures prompt/response content on span attributes. +- ``OTEL_SEMCONV_STABILITY_OPT_IN=gen_ai_latest_experimental`` enables latest experimental features -- required for the message content above to actually appear. + +Setup +----- + +Copy `.env.example <.env.example>`_ to ``.env`` and update it with your +``OPENAI_API_KEY``. An OTLP compatible endpoint should be listening for +traces on http://localhost:4317. If not, update +``OTEL_EXPORTER_OTLP_ENDPOINT`` as well. + +Next, set up a virtual environment like this: + +:: + + python3 -m venv .venv + source .venv/bin/activate + pip install "python-dotenv[cli]" + pip install -r requirements.txt + +Run +--- + +Run the example like this: + +:: + + dotenv run -- python main.py + +You should see a two-sentence summary printed, while the trace exports to +your configured observability tool. diff --git a/instrumentation-genai/opentelemetry-instrumentation-crewai/examples/manual/main.py b/instrumentation-genai/opentelemetry-instrumentation-crewai/examples/manual/main.py new file mode 100644 index 0000000000..c34d8b9115 --- /dev/null +++ b/instrumentation-genai/opentelemetry-instrumentation-crewai/examples/manual/main.py @@ -0,0 +1,92 @@ +# Copyright The OpenTelemetry Authors +# SPDX-License-Identifier: Apache-2.0 + +# pylint: skip-file +import os + +from crewai import Agent, Crew, Process, Task +from crewai.tools import BaseTool +from pydantic import BaseModel, Field + +from opentelemetry import trace +from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import ( + OTLPSpanExporter, +) +from opentelemetry.instrumentation.crewai import CrewAIInstrumentor +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import BatchSpanProcessor + +# configure tracing +trace.set_tracer_provider(TracerProvider()) +trace.get_tracer_provider().add_span_processor( + BatchSpanProcessor(OTLPSpanExporter()) +) + +# instrument CrewAI +CrewAIInstrumentor().instrument() + + +class WordCountInput(BaseModel): + text: str = Field(description="The text to count words in.") + + +class WordCountTool(BaseTool): + name: str = "word_count" + description: str = "Counts the number of words in a piece of text." + args_schema: type[BaseModel] = WordCountInput + + def _run(self, text: str) -> str: + return f"{len(text.split())} words" + + +def build_crew() -> Crew: + researcher = Agent( + role="Researcher", + goal="Produce a short, factual note about the given topic.", + backstory="A concise research analyst who writes brief, factual notes with no filler.", + tools=[WordCountTool()], + ) + + writer = Agent( + role="Writer", + goal="Turn a research note into a two-sentence summary.", + backstory="A summarizer who writes exactly two plain sentences.", + ) + + research_task = Task( + description=( + "Write a 3-sentence factual note about {topic}. Then use the " + "word_count tool to report how many words your note has." + ), + expected_output="A 3-sentence note about {topic}, plus its word count.", + agent=researcher, + ) + + writing_task = Task( + description=( + "Using the research note above, write exactly two plain " + "sentences summarizing {topic} for a general audience." + ), + expected_output="A two-sentence plain-language summary of {topic}.", + agent=writer, + context=[research_task], + ) + + return Crew( + agents=[researcher, writer], + tasks=[research_task, writing_task], + process=Process.sequential, + ) + + +def main(): + os.environ.setdefault( + "OPENAI_MODEL_NAME", os.environ.get("CHAT_MODEL", "gpt-4o-mini") + ) + crew = build_crew() + result = crew.kickoff(inputs={"topic": "the history of the paperclip"}) + print(result) + + +if __name__ == "__main__": + main() diff --git a/instrumentation-genai/opentelemetry-instrumentation-crewai/examples/manual/requirements.txt b/instrumentation-genai/opentelemetry-instrumentation-crewai/examples/manual/requirements.txt new file mode 100644 index 0000000000..7e8ea6dd14 --- /dev/null +++ b/instrumentation-genai/opentelemetry-instrumentation-crewai/examples/manual/requirements.txt @@ -0,0 +1,5 @@ +crewai>=1.15.5 + +opentelemetry-sdk~=1.36.0 +opentelemetry-exporter-otlp-proto-grpc~=1.36.0 +opentelemetry-instrumentation-crewai diff --git a/instrumentation-genai/opentelemetry-instrumentation-crewai/examples/zero-code/.env.example b/instrumentation-genai/opentelemetry-instrumentation-crewai/examples/zero-code/.env.example new file mode 100644 index 0000000000..4badcefdb2 --- /dev/null +++ b/instrumentation-genai/opentelemetry-instrumentation-crewai/examples/zero-code/.env.example @@ -0,0 +1,14 @@ +# Update this with your real OpenAI API key +OPENAI_API_KEY=sk-YOUR_API_KEY + +# Uncomment and change to your OTLP endpoint +# OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4317 +# OTEL_EXPORTER_OTLP_PROTOCOL=grpc + +OTEL_SERVICE_NAME=opentelemetry-python-crewai + +# Remove to hide prompt and completion content +OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT=span_only + +# Required for the capture-content setting above to actually take effect. +OTEL_SEMCONV_STABILITY_OPT_IN=gen_ai_latest_experimental diff --git a/instrumentation-genai/opentelemetry-instrumentation-crewai/examples/zero-code/README.rst b/instrumentation-genai/opentelemetry-instrumentation-crewai/examples/zero-code/README.rst new file mode 100644 index 0000000000..0bf6a69845 --- /dev/null +++ b/instrumentation-genai/opentelemetry-instrumentation-crewai/examples/zero-code/README.rst @@ -0,0 +1,47 @@ +OpenTelemetry CrewAI Zero-Code Instrumentation Example +======================================================== + +This is an example of how to instrument CrewAI with zero code changes, +using ``opentelemetry-instrument``. + +When `main.py `_ is run, it builds a two-agent crew (a researcher +with a word-count tool, and a writer who summarizes the researcher's +output) and exports the trace to an OTLP compatible endpoint, without a +single line of OpenTelemetry code in ``main.py`` itself -- instrumentation +is discovered and wired up automatically at process startup via +``opentelemetry-instrument``. + +Note: `.env.example <.env.example>`_ configures additional environment +variables: + +- ``OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT=span_only`` captures prompt/response content on span attributes. +- ``OTEL_SEMCONV_STABILITY_OPT_IN=gen_ai_latest_experimental`` enables latest experimental features -- required for the message content above to actually appear. + +Setup +----- + +Copy `.env.example <.env.example>`_ to ``.env`` and update it with your +``OPENAI_API_KEY``. An OTLP compatible endpoint should be listening for +traces on http://localhost:4317. If not, update +``OTEL_EXPORTER_OTLP_ENDPOINT`` as well. + +Next, set up a virtual environment like this: + +:: + + python3 -m venv .venv + source .venv/bin/activate + pip install "python-dotenv[cli]" + pip install -r requirements.txt + +Run +--- + +Run the example like this: + +:: + + dotenv run -- opentelemetry-instrument python main.py + +You should see a two-sentence summary printed, while the trace exports to +your configured observability tool. diff --git a/instrumentation-genai/opentelemetry-instrumentation-crewai/examples/zero-code/main.py b/instrumentation-genai/opentelemetry-instrumentation-crewai/examples/zero-code/main.py new file mode 100644 index 0000000000..d3160591bb --- /dev/null +++ b/instrumentation-genai/opentelemetry-instrumentation-crewai/examples/zero-code/main.py @@ -0,0 +1,74 @@ +# Copyright The OpenTelemetry Authors +# SPDX-License-Identifier: Apache-2.0 + +import os + +from crewai import Agent, Crew, Process, Task +from crewai.tools import BaseTool +from pydantic import BaseModel, Field + + +class WordCountInput(BaseModel): + text: str = Field(description="The text to count words in.") + + +class WordCountTool(BaseTool): + name: str = "word_count" + description: str = "Counts the number of words in a piece of text." + args_schema: type[BaseModel] = WordCountInput + + def _run(self, text: str) -> str: + return f"{len(text.split())} words" + + +def build_crew() -> Crew: + researcher = Agent( + role="Researcher", + goal="Produce a short, factual note about the given topic.", + backstory="A concise research analyst who writes brief, factual notes with no filler.", + tools=[WordCountTool()], + ) + + writer = Agent( + role="Writer", + goal="Turn a research note into a two-sentence summary.", + backstory="A summarizer who writes exactly two plain sentences.", + ) + + research_task = Task( + description=( + "Write a 3-sentence factual note about {topic}. Then use the " + "word_count tool to report how many words your note has." + ), + expected_output="A 3-sentence note about {topic}, plus its word count.", + agent=researcher, + ) + + writing_task = Task( + description=( + "Using the research note above, write exactly two plain " + "sentences summarizing {topic} for a general audience." + ), + expected_output="A two-sentence plain-language summary of {topic}.", + agent=writer, + context=[research_task], + ) + + return Crew( + agents=[researcher, writer], + tasks=[research_task, writing_task], + process=Process.sequential, + ) + + +def main(): + os.environ.setdefault( + "OPENAI_MODEL_NAME", os.environ.get("CHAT_MODEL", "gpt-4o-mini") + ) + crew = build_crew() + result = crew.kickoff(inputs={"topic": "the history of the paperclip"}) + print(result) + + +if __name__ == "__main__": + main() diff --git a/instrumentation-genai/opentelemetry-instrumentation-crewai/examples/zero-code/requirements.txt b/instrumentation-genai/opentelemetry-instrumentation-crewai/examples/zero-code/requirements.txt new file mode 100644 index 0000000000..ddd8f343da --- /dev/null +++ b/instrumentation-genai/opentelemetry-instrumentation-crewai/examples/zero-code/requirements.txt @@ -0,0 +1,6 @@ +crewai>=1.15.5 + +opentelemetry-sdk~=1.36.0 +opentelemetry-exporter-otlp-proto-grpc~=1.36.0 +opentelemetry-distro~=0.57b0 +opentelemetry-instrumentation-crewai diff --git a/instrumentation-genai/opentelemetry-instrumentation-crewai/pyproject.toml b/instrumentation-genai/opentelemetry-instrumentation-crewai/pyproject.toml new file mode 100644 index 0000000000..bde6563263 --- /dev/null +++ b/instrumentation-genai/opentelemetry-instrumentation-crewai/pyproject.toml @@ -0,0 +1,56 @@ +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[project] +name = "opentelemetry-instrumentation-crewai" +dynamic = ["version"] +description = "OpenTelemetry CrewAI instrumentation" +readme = "README.rst" +license = "Apache-2.0" +requires-python = ">=3.10" +authors = [ + { name = "OpenTelemetry Authors", email = "cncf-opentelemetry-contributors@lists.cncf.io" }, +] +classifiers = [ + "Development Status :: 4 - Beta", + "Intended Audience :: Developers", + "License :: OSI Approved :: Apache Software License", + "Programming Language :: Python", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Programming Language :: Python :: 3.13", + "Programming Language :: Python :: 3.14", +] +dependencies = [ + "opentelemetry-api >= 1.40", + "opentelemetry-instrumentation >= 0.61b0", + "opentelemetry-semantic-conventions >= 0.61b0", + "opentelemetry-util-genai >= 0.4b0", +] + +[project.optional-dependencies] +instruments = [ + "crewai >= 1.15.5", +] + +[project.entry-points.opentelemetry_instrumentor] +crewai = "opentelemetry.instrumentation.crewai:CrewAIInstrumentor" + +[project.urls] +Homepage = "https://github.com/open-telemetry/opentelemetry-python-contrib/tree/main/instrumentation-genai/opentelemetry-instrumentation-crewai" +Repository = "https://github.com/open-telemetry/opentelemetry-python-contrib" +Issue = "https://github.com/SigNoz/signoz/issues/11663" + +[tool.hatch.version] +path = "src/opentelemetry/instrumentation/crewai/version.py" + +[tool.hatch.build.targets.sdist] +include = [ + "/src", +] + +[tool.hatch.build.targets.wheel] +packages = ["src/opentelemetry"] diff --git a/instrumentation-genai/opentelemetry-instrumentation-crewai/src/opentelemetry/instrumentation/crewai/__init__.py b/instrumentation-genai/opentelemetry-instrumentation-crewai/src/opentelemetry/instrumentation/crewai/__init__.py new file mode 100644 index 0000000000..a9866d1d46 --- /dev/null +++ b/instrumentation-genai/opentelemetry-instrumentation-crewai/src/opentelemetry/instrumentation/crewai/__init__.py @@ -0,0 +1,152 @@ +# Copyright The OpenTelemetry Authors +# SPDX-License-Identifier: Apache-2.0 + +""" +CrewAI instrumentation supporting `crewai`_, instrumented by +``CrewAIInstrumentor``. + +.. _crewai: https://pypi.org/project/crewai/ + +Usage +----- + +.. code:: python + + from crewai import Agent, Crew, Task + from opentelemetry.instrumentation.crewai import CrewAIInstrumentor + + CrewAIInstrumentor().instrument() + + crew = Crew(agents=[...], tasks=[...]) + crew.kickoff() + +This produces one ``invoke_workflow`` root span per crew run, with a nested +``invoke_agent`` span per agent/task pair, ``chat``/``execute_tool`` spans +nested under those, and a span **link** (not a parent span) between agents +when one agent's task output feeds another. + +Configuration +------------- + +Telemetry emission, content capture, and related environment variables are +owned by ``opentelemetry-util-genai``. See the `opentelemetry-util-genai +README +`_ +for the full list. + +API +--- +""" + +from __future__ import annotations + +from collections.abc import Collection +from importlib import import_module +from typing import Any + +from crewai.events.event_bus import crewai_event_bus +from crewai.events.types.llm_events import LLMCallCompletedEvent +from wrapt import wrap_function_wrapper + +from opentelemetry.instrumentation.crewai._wrappers import ( + on_llm_call_completed, + wrap_agent_execute_task, + wrap_crew_kickoff, + wrap_llm_call, + wrap_tool_run, +) +from opentelemetry.instrumentation.crewai.package import _instruments +from opentelemetry.instrumentation.instrumentor import BaseInstrumentor +from opentelemetry.instrumentation.utils import unwrap +from opentelemetry.util.genai.completion_hook import load_completion_hook +from opentelemetry.util.genai.handler import TelemetryHandler + +# ``LLM.call`` is abstract on CrewAI's own BaseLLM -- every concrete +# provider implementation (the generic litellm-backed crewai.LLM, plus each +# native provider CrewAI ships) defines its own ``call``, so there is no +# single class whose method covers every provider. These are every +# concrete implementation as of crewai 1.15.5 (confirmed by reading the +# installed source, not assumed); provider modules for SDKs that aren't +# installed (e.g. boto3 for Bedrock) raise ImportError at import time, so +# each entry is patched best-effort and skipped if unavailable. +_LLM_CALL_CLASSES: tuple[tuple[str, str], ...] = ( + ("crewai", "LLM"), + ("crewai.llms.providers.openai.completion", "OpenAICompletion"), + ("crewai.llms.providers.anthropic.completion", "AnthropicCompletion"), + ("crewai.llms.providers.gemini.completion", "GeminiCompletion"), + ("crewai.llms.providers.azure.completion", "AzureCompletion"), + ("crewai.llms.providers.bedrock.completion", "BedrockCompletion"), +) + + +class CrewAIInstrumentor(BaseInstrumentor): + """An :class:`~opentelemetry.instrumentation.instrumentor.BaseInstrumentor` + for CrewAI. + + Patches ``Crew.kickoff``, ``Agent.execute_task``, every concrete LLM + provider's ``call`` (see ``_LLM_CALL_CLASSES``), and ``BaseTool.run`` at + runtime to emit GenAI telemetry through ``opentelemetry-util-genai``; + never modifies CrewAI's installed source. ``instrument()``/ + ``uninstrument()`` are idempotent (inherited from ``BaseInstrumentor``): + calling either twice in a row is a no-op on the second call. + """ + + def __init__(self) -> None: + super().__init__() + self._patched_llm_modules: list[tuple[str, str]] = [] + + def instrumentation_dependencies(self) -> Collection[str]: + """Return the ``crewai`` version range this instrumentor supports.""" + return _instruments + + def _instrument(self, **kwargs: Any) -> None: + """Patch CrewAI's entry points and start listening for LLM call events.""" + handler = TelemetryHandler( + tracer_provider=kwargs.get("tracer_provider"), + meter_provider=kwargs.get("meter_provider"), + logger_provider=kwargs.get("logger_provider"), + completion_hook=kwargs.get("completion_hook") + or load_completion_hook(), + ) + + crewai_event_bus.register_handler( + LLMCallCompletedEvent, on_llm_call_completed + ) + + wrap_function_wrapper( + "crewai", "Crew.kickoff", wrap_crew_kickoff(handler) + ) + wrap_function_wrapper( + "crewai", "Agent.execute_task", wrap_agent_execute_task(handler) + ) + wrap_function_wrapper( + "crewai.tools", "BaseTool.run", wrap_tool_run(handler) + ) + + llm_call_wrapper = wrap_llm_call(handler) + self._patched_llm_modules = [] + for module_path, class_name in _LLM_CALL_CLASSES: + try: + import_module(module_path) + except ImportError: + continue # optional provider SDK not installed + wrap_function_wrapper( + module_path, f"{class_name}.call", llm_call_wrapper + ) + self._patched_llm_modules.append((module_path, class_name)) + + def _uninstrument(self, **kwargs: Any) -> None: + """Undo every patch applied by ``_instrument``.""" + import crewai # noqa: PLC0415 + import crewai.tools # noqa: PLC0415 + + unwrap(crewai.Crew, "kickoff") + unwrap(crewai.Agent, "execute_task") + unwrap(crewai.tools.BaseTool, "run") + + for module_path, class_name in self._patched_llm_modules: + module = import_module(module_path) + unwrap(getattr(module, class_name), "call") + self._patched_llm_modules = [] + + crewai_event_bus.off(LLMCallCompletedEvent, on_llm_call_completed) diff --git a/instrumentation-genai/opentelemetry-instrumentation-crewai/src/opentelemetry/instrumentation/crewai/_wrappers.py b/instrumentation-genai/opentelemetry-instrumentation-crewai/src/opentelemetry/instrumentation/crewai/_wrappers.py new file mode 100644 index 0000000000..209ca81531 --- /dev/null +++ b/instrumentation-genai/opentelemetry-instrumentation-crewai/src/opentelemetry/instrumentation/crewai/_wrappers.py @@ -0,0 +1,428 @@ +# Copyright The OpenTelemetry Authors +# SPDX-License-Identifier: Apache-2.0 + +"""Method wrappers that turn CrewAI execution into GenAI telemetry. + +Each public ``wrap_*`` function here is bound with a closure to a +``TelemetryHandler`` in :mod:`opentelemetry.instrumentation.crewai` and +installed with ``wrapt.wrap_function_wrapper``. They all follow the same +shape: start an invocation via the handler, set what fields are available +before calling the wrapped method, call it, set the remaining fields from +the result, and on exception fail the invocation and re-raise unmodified. + +Telemetry emission itself (spans, metrics, events, content-capture gating) +is owned entirely by ``opentelemetry-util-genai``'s ``TelemetryHandler`` -- +this module never touches a ``Tracer``, ``Meter``, or ``Logger`` directly, +per the GenAI instrumentation review rules. +""" + +from __future__ import annotations + +import threading +from collections import deque +from typing import TYPE_CHECKING, Any, Callable + +from opentelemetry.trace import SpanContext +from opentelemetry.util.genai.handler import TelemetryHandler +from opentelemetry.util.genai.invocation import ( + AgentInvocation, + Error, + InferenceInvocation, +) +from opentelemetry.util.genai.types import InputMessage, OutputMessage, Text + +if TYPE_CHECKING: + from crewai import Agent, Crew, Task + from crewai.tools import BaseTool + +# --------------------------------------------------------------------------- +# crew.kickoff -> WorkflowInvocation +# --------------------------------------------------------------------------- + + +def wrap_crew_kickoff( + handler: TelemetryHandler, +) -> Callable[ + [Callable[..., Any], Crew, tuple[Any, ...], dict[str, Any]], Any +]: + """Build the wrapper for ``Crew.kickoff``. + + Only the synchronous ``kickoff`` is patched. ``Crew.kickoff_async`` + delegates to ``self.kickoff`` via ``asyncio.to_thread`` (which copies the + current ``contextvars.Context`` into the worker thread), so patching + both would double-count every async run as two nested workflow spans. + Patching only the sync method covers both entry points correctly. + """ + + def _wrap_crew_kickoff( + wrapped: Callable[..., Any], + instance: Crew, + args: tuple[Any, ...], + kwargs: dict[str, Any], + ) -> Any: + agents: list[Any] = getattr(instance, "agents", None) or [] + tasks: list[Any] = getattr(instance, "tasks", None) or [] + process = getattr(instance, "process", None) + + invocation = handler.start_workflow(name="crew.kickoff") + invocation.attributes["crewai.crew.id"] = str( + getattr(instance, "id", "") + ) + invocation.attributes["crewai.crew.process"] = ( + getattr(process, "value", str(process)) if process else "" + ) + invocation.attributes["crewai.crew.agent_count"] = len(agents) + invocation.attributes["crewai.crew.task_count"] = len(tasks) + + _reset_handoff_state(instance) + try: + result = wrapped(*args, **kwargs) + except Exception as exc: + invocation.fail(Error(type=type(exc), message=str(exc))) + raise + finally: + _clear_handoff_state(instance) + + invocation.stop() + return result + + return _wrap_crew_kickoff + + +# --------------------------------------------------------------------------- +# agent.execute_task -> AgentInvocation (local, INTERNAL span kind) +# + opt-in prompt capture, + handoff links +# --------------------------------------------------------------------------- + +# Per-crew-run handoff tracking: id(crew) -> {id(task): SpanContext of the +# agent invocation that produced that task's output}. Kept on the +# instrumentation module rather than on the crewai objects themselves, +# because Crew/Task/Agent are pydantic models that reject arbitrary +# attribute assignment. +_handoff_spans: dict[int, dict[int, SpanContext]] = {} +_handoff_lock = threading.Lock() + + +def _reset_handoff_state(crew: Crew) -> None: + with _handoff_lock: + _handoff_spans[id(crew)] = {} + + +def _clear_handoff_state(crew: Crew) -> None: + with _handoff_lock: + _handoff_spans.pop(id(crew), None) + + +def _record_handoff_span( + crew: Crew | None, task: Task, span_context: SpanContext +) -> None: + if crew is None or isinstance(crew, str): + return + with _handoff_lock: + crew_spans = _handoff_spans.get(id(crew)) + if crew_spans is not None: + crew_spans[id(task)] = span_context + + +def _resolve_handoff_span_contexts( + crew: Crew | None, task: Task +) -> list[SpanContext]: + """Resolve the span contexts representing a handoff into ``task``. + + Mirrors ``Crew._get_context``: an explicit ``task.context`` list links + only to those specific prior tasks; the default "not specified" sentinel + links to every task executed so far in this crew run (matching CrewAI's + own default of aggregating every prior task output); an explicit + ``None``/empty list means no handoff at all. + """ + if crew is None or isinstance(crew, str): + return [] + with _handoff_lock: + crew_spans = _handoff_spans.get(id(crew)) + if not crew_spans: + return [] + + task_context = getattr(task, "context", None) + if not task_context: + return [] + + try: + from crewai.utilities.constants import ( # noqa: PLC0415 + NOT_SPECIFIED, + ) + + context_is_not_specified = task_context is NOT_SPECIFIED + except ImportError: + context_is_not_specified = False # pragma: no cover - defensive + + if context_is_not_specified: + source_task_ids = list(crew_spans.keys()) + else: + source_task_ids = [id(t) for t in task_context] + + return [ + crew_spans[task_id] + for task_id in source_task_ids + if task_id in crew_spans + ] + + +def wrap_agent_execute_task( + handler: TelemetryHandler, +) -> Callable[ + [Callable[..., Any], Agent, tuple[Any, ...], dict[str, Any]], Any +]: + """Build the wrapper for ``Agent.execute_task``.""" + capture_content = handler.should_capture_content() + + def _wrap_agent_execute_task( + wrapped: Callable[..., Any], + instance: Agent, + args: tuple[Any, ...], + kwargs: dict[str, Any], + ) -> Any: + task: Task | None = args[0] if args else kwargs.get("task") + crew = getattr(instance, "crew", None) + role = getattr(instance, "role", "") or "" + + invocation: AgentInvocation = handler.start_invoke_local_agent( + "crewai", agent_name=role or None + ) + invocation.attributes["crewai.agent.role"] = role + if task is not None: + invocation.attributes["crewai.task.description"] = str( + getattr(task, "description", "") + ) + for link_context in _resolve_handoff_span_contexts(crew, task): + invocation.span.add_link(link_context) + + if capture_content: + task_context = ( + args[1] if len(args) > 1 else kwargs.get("context") + ) + if task_context: + invocation.input_messages = [ + InputMessage( + role="user", + parts=[Text(content=str(task_context))], + ) + ] + + try: + result = wrapped(*args, **kwargs) + except Exception as exc: + invocation.fail(Error(type=type(exc), message=str(exc))) + raise + + if capture_content: + invocation.output_messages = [ + OutputMessage( + role="assistant", + parts=[Text(content=str(result))], + finish_reason="stop", + ) + ] + + invocation.stop() + + if task is not None: + _record_handoff_span( + crew, task, invocation.span.get_span_context() + ) + return result + + return _wrap_agent_execute_task + + +# --------------------------------------------------------------------------- +# llm.call -> InferenceInvocation +# --------------------------------------------------------------------------- + +# LLM.call() only returns text (or a tool result) -- usage, finish_reason, +# and the resolved model aren't in that return value, they're only on the +# LLMCallCompletedEvent CrewAI emits from inside the call. That emit() is +# fire-and-forget (crewai_event_bus dispatches sync handlers on a +# ThreadPoolExecutor and does not wait for them -- see +# CrewAIEventsBus.emit()), so there is no guarantee the event has already +# been handled by the time the wrapped call() returns. We correlate by +# id(instance) (the handler's ``source`` argument is the same LLM instance +# our wrapper is closed over) and wait briefly with a threading.Event, +# rather than relying on contextvars, which would not propagate a value set +# inside a different worker thread back to the caller's context. +# +# Each instance maps to a FIFO queue of waiters, not a single slot: CrewAI +# gives each agent its own LLM instance by default, so in the common case +# there's one waiter per instance and the queue never holds more than one +# entry. But if a user shares one LLM object across agents/tasks that run +# concurrently, two calls on the *same* instance can be in flight at once; +# a single overwritable slot would let the second call's registration clobber +# the first's, delivering the wrong usage data to the wrong span. A queue +# means overlapping calls each get their own waiter instead of colliding. +# This still assumes completion events arrive in the same order calls were +# dispatched, which holds unless two concurrent calls on one shared instance +# finish out of order -- a narrow edge case that's a documented limitation, +# not a crash: attributes just end up on the wrong (but still valid) +# invocation. +_pending_llm_calls: dict[int, deque[tuple[threading.Event, list[Any]]]] = {} +_pending_lock = threading.Lock() +_LLM_EVENT_WAIT_SECONDS = 2.0 + + +def on_llm_call_completed(source: Any, event: Any) -> None: + """crewai_event_bus listener: hand a completed call's usage/finish data + to the oldest still-waiting ``wrap_llm_call`` invocation for this + instance.""" + with _pending_lock: + queue = _pending_llm_calls.get(id(source)) + entry = queue.popleft() if queue else None + if entry is not None: + wait_event, holder = entry + holder.append(event) + wait_event.set() + + +def wrap_llm_call( + handler: TelemetryHandler, +) -> Callable[[Callable[..., Any], Any, tuple[Any, ...], dict[str, Any]], Any]: + """Build the wrapper for ``LLM.call``.""" + capture_content = handler.should_capture_content() + + def _wrap_llm_call( + wrapped: Callable[..., Any], + instance: Any, + args: tuple[Any, ...], + kwargs: dict[str, Any], + ) -> Any: + messages = args[0] if args else kwargs.get("messages") + provider = str( + getattr(instance, "provider", None) + or getattr(instance, "model", "") + ) + request_model = str(getattr(instance, "model", "")) + + invocation: InferenceInvocation = handler.start_inference( + provider, request_model=request_model + ) + if capture_content and messages is not None: + invocation.input_messages = [ + InputMessage(role="user", parts=[Text(content=str(messages))]) + ] + + wait_event = threading.Event() + holder: list[Any] = [] + entry = (wait_event, holder) + with _pending_lock: + _pending_llm_calls.setdefault(id(instance), deque()).append(entry) + try: + result = wrapped(*args, **kwargs) + except Exception as exc: + with _pending_lock: + _discard_pending_entry(id(instance), entry) + invocation.fail(Error(type=type(exc), message=str(exc))) + raise + + # Success path only: emit() dispatches the completion event to a + # thread pool asynchronously, so it may not have arrived yet even + # though call() itself has already returned. Wait briefly, then + # stop listening for it either way. + wait_event.wait(timeout=_LLM_EVENT_WAIT_SECONDS) + with _pending_lock: + _discard_pending_entry(id(instance), entry) + completed_event = holder[0] if holder else None + _apply_llm_response(invocation, completed_event) + + if capture_content: + finish_reason = ( + invocation.finish_reasons[0] + if invocation.finish_reasons + else "stop" + ) + invocation.output_messages = [ + OutputMessage( + role="assistant", + parts=[Text(content=str(result))], + finish_reason=finish_reason, + ) + ] + + invocation.stop() + return result + + return _wrap_llm_call + + +def _discard_pending_entry( + instance_id: int, entry: tuple[threading.Event, list[Any]] +) -> None: + """Remove ``entry`` from its instance's waiter queue if it's still there + (already-serviced entries are gone) and drop the queue once it's empty, + so ``_pending_llm_calls`` never accumulates stale keys.""" + queue = _pending_llm_calls.get(instance_id) + if queue is None: + return + try: + queue.remove(entry) + except ValueError: + pass # already popped by on_llm_call_completed + if not queue: + del _pending_llm_calls[instance_id] + + +def _apply_llm_response(invocation: InferenceInvocation, event: Any) -> None: + if event is None: + return + usage: dict[str, Any] = getattr(event, "usage", None) or {} + input_tokens: int | None = usage.get("prompt_tokens") + output_tokens: int | None = usage.get("completion_tokens") + if input_tokens is not None: + invocation.input_tokens = input_tokens + if output_tokens is not None: + invocation.output_tokens = output_tokens + finish_reason = getattr(event, "finish_reason", None) + if finish_reason: + invocation.finish_reasons = [str(finish_reason)] + + +# --------------------------------------------------------------------------- +# tool.call -> ToolInvocation +# --------------------------------------------------------------------------- + + +def wrap_tool_run( + handler: TelemetryHandler, +) -> Callable[ + [Callable[..., Any], BaseTool, tuple[Any, ...], dict[str, Any]], Any +]: + """Build the wrapper for ``BaseTool.run``. + + ``run`` (not ``_run``) is the stable public entry point: every tool + subclass implements ``_run`` differently, but ``run`` is the single + common method every tool call passes through. + """ + capture_content = handler.should_capture_content() + + def _wrap_tool_run( + wrapped: Callable[..., Any], + instance: BaseTool, + args: tuple[Any, ...], + kwargs: dict[str, Any], + ) -> Any: + invocation = handler.start_tool( + getattr(instance, "name", "") or "", + arguments={"args": args, "kwargs": kwargs} + if capture_content + else None, + ) + try: + result = wrapped(*args, **kwargs) + except Exception as exc: + invocation.fail(Error(type=type(exc), message=str(exc))) + raise + + if capture_content: + invocation.tool_result = result + + invocation.stop() + return result + + return _wrap_tool_run diff --git a/instrumentation-genai/opentelemetry-instrumentation-crewai/src/opentelemetry/instrumentation/crewai/package.py b/instrumentation-genai/opentelemetry-instrumentation-crewai/src/opentelemetry/instrumentation/crewai/package.py new file mode 100644 index 0000000000..a3c5967f1f --- /dev/null +++ b/instrumentation-genai/opentelemetry-instrumentation-crewai/src/opentelemetry/instrumentation/crewai/package.py @@ -0,0 +1,8 @@ +# Copyright The OpenTelemetry Authors +# SPDX-License-Identifier: Apache-2.0 + +_instruments = ("crewai >= 1.15.5",) +# NOTE: only 1.15.5 has actually been tested against so far. The lower bound +# is set to that exact version deliberately, not a guess at an older +# compatible one -- widen it only after real testing against an older +# release, per tests/requirements.oldest.txt. diff --git a/instrumentation-genai/opentelemetry-instrumentation-crewai/src/opentelemetry/instrumentation/crewai/version.py b/instrumentation-genai/opentelemetry-instrumentation-crewai/src/opentelemetry/instrumentation/crewai/version.py new file mode 100644 index 0000000000..3716e78195 --- /dev/null +++ b/instrumentation-genai/opentelemetry-instrumentation-crewai/src/opentelemetry/instrumentation/crewai/version.py @@ -0,0 +1,4 @@ +# Copyright The OpenTelemetry Authors +# SPDX-License-Identifier: Apache-2.0 + +__version__ = "0.1.dev0" diff --git a/instrumentation-genai/opentelemetry-instrumentation-crewai/tests/__init__.py b/instrumentation-genai/opentelemetry-instrumentation-crewai/tests/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/instrumentation-genai/opentelemetry-instrumentation-crewai/tests/conftest.py b/instrumentation-genai/opentelemetry-instrumentation-crewai/tests/conftest.py new file mode 100644 index 0000000000..2f697a2925 --- /dev/null +++ b/instrumentation-genai/opentelemetry-instrumentation-crewai/tests/conftest.py @@ -0,0 +1,121 @@ +# Copyright The OpenTelemetry Authors +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +import os +from types import SimpleNamespace + +import pytest + +from opentelemetry.instrumentation._semconv import ( + OTEL_SEMCONV_STABILITY_OPT_IN, + _OpenTelemetrySemanticConventionStability, +) +from opentelemetry.instrumentation.crewai import _wrappers +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import SimpleSpanProcessor +from opentelemetry.sdk.trace.export.in_memory_span_exporter import ( + InMemorySpanExporter, +) +from opentelemetry.util.genai.environment_variables import ( + OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT, +) +from opentelemetry.util.genai.handler import TelemetryHandler + + +def _make_handler_and_exporter() -> tuple[ + TelemetryHandler, InMemorySpanExporter +]: + exp = InMemorySpanExporter() + provider = TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(exp)) + handler = TelemetryHandler(tracer_provider=provider) + return handler, exp + + +@pytest.fixture +def handler_and_exporter(): + """A TelemetryHandler and its backing InMemorySpanExporter, fresh per + test, with content capture off (the default).""" + handler, exp = _make_handler_and_exporter() + yield handler, exp + exp.clear() + + +@pytest.fixture +def content_capturing_handler_and_exporter(): + """Same as ``handler_and_exporter``, but with content capture on. + + Two things have to be true at once for message content (input/output + messages on inference/agent/workflow invocations) to actually land on a + span, both verified by reading opentelemetry-util-genai's source rather + than assumed: + + - ``OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT`` has to be a + real capturing-mode value (``span_only`` here), not just ``true`` -- + ``true`` only satisfies ``TelemetryHandler.should_capture_content()`` + on the legacy path, which gates *whether* our wrappers populate + ``arguments``/``tool_result``/message fields at all, but says nothing + about whether the util then puts that content on the span. + - ``OTEL_SEMCONV_STABILITY_OPT_IN=gen_ai_latest_experimental`` also has + to be set, because ``get_content_attributes()`` returns ``{}`` + unconditionally when not in experimental mode, regardless of the + capture-content setting. + + ``TelemetryHandler`` also reads the capture-content env var once, at + construction time (not per-call), so both env vars have to be set + before the handler is built. The stability opt-in itself is cached in + a module-level singleton the first time it's checked, so that singleton + has to be reset per test too, or a later test would see whatever mode + the first test that touched it happened to set. + """ + os.environ[OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT] = ( + "span_only" + ) + os.environ[OTEL_SEMCONV_STABILITY_OPT_IN] = "gen_ai_latest_experimental" + _OpenTelemetrySemanticConventionStability._initialized = False + _OpenTelemetrySemanticConventionStability._initialize() + + handler, exp = _make_handler_and_exporter() + yield handler, exp + exp.clear() + os.environ.pop(OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT, None) + os.environ.pop(OTEL_SEMCONV_STABILITY_OPT_IN, None) + _OpenTelemetrySemanticConventionStability._initialized = False + _OpenTelemetrySemanticConventionStability._initialize() + + +@pytest.fixture(autouse=True) +def _reset_instrumentation_state(): + """Module-level correlation state must not leak between tests.""" + yield + _wrappers._handoff_spans.clear() + _wrappers._pending_llm_calls.clear() + + +def fake_crew( + *, agents=None, tasks=None, process="sequential", crew_id="crew-1" +): + return SimpleNamespace( + id=crew_id, + agents=agents or [], + tasks=tasks or [], + process=SimpleNamespace(value=process), + ) + + +def fake_task(*, description="do the thing", context=None): + return SimpleNamespace(description=description, context=context) + + +def fake_agent(*, role="Researcher", crew=None): + return SimpleNamespace(role=role, crew=crew) + + +def fake_llm(*, model="gpt-4o-mini", provider="openai"): + return SimpleNamespace(model=model, provider=provider) + + +def fake_tool(*, name="word_count"): + return SimpleNamespace(name=name) diff --git a/instrumentation-genai/opentelemetry-instrumentation-crewai/tests/requirements.latest.txt b/instrumentation-genai/opentelemetry-instrumentation-crewai/tests/requirements.latest.txt new file mode 100644 index 0000000000..eef787af92 --- /dev/null +++ b/instrumentation-genai/opentelemetry-instrumentation-crewai/tests/requirements.latest.txt @@ -0,0 +1,36 @@ +# Copyright The OpenTelemetry Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# ******************************** +# WARNING: NOT HERMETIC !!!!!!!!!! +# ******************************** +# +# This "requirements.txt" is installed in conjunction with multiple other +# dependencies in the top-level "tox.ini" file, which provides +# opentelemetry-api, opentelemetry-sdk, and opentelemetry-semantic-conventions +# with a "dev" version based on the latest distribution. + +# This variant of the requirements aims to test the system using +# the newest supported version of external dependencies. +# +# NOTE: same caveat as requirements.oldest.txt -- 1.15.5 is the only +# version actually tested so far. + +crewai==1.15.5 +pytest==7.4.4 +wrapt==1.14.0 + +-e opentelemetry-instrumentation +-e util/opentelemetry-util-genai +-e instrumentation-genai/opentelemetry-instrumentation-crewai diff --git a/instrumentation-genai/opentelemetry-instrumentation-crewai/tests/requirements.oldest.txt b/instrumentation-genai/opentelemetry-instrumentation-crewai/tests/requirements.oldest.txt new file mode 100644 index 0000000000..316e7b6bf1 --- /dev/null +++ b/instrumentation-genai/opentelemetry-instrumentation-crewai/tests/requirements.oldest.txt @@ -0,0 +1,31 @@ +# Copyright The OpenTelemetry Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# This variant of the requirements aims to test the system using +# the oldest supported version of external dependencies. +# +# NOTE: 1.15.5 is pinned here as *both* the oldest and latest bound +# because it is the only version this instrumentation has actually been +# tested against so far (see package.py). Widen this only after testing +# against a real older release, not as a guess. + +crewai==1.15.5 +pytest==7.4.4 +wrapt==1.14.0 +opentelemetry-api==1.40 # when updating, also update in pyproject.toml +opentelemetry-sdk==1.40 # when updating, also update in pyproject.toml +opentelemetry-semantic-conventions==0.61b0 # when updating, also update in pyproject.toml +opentelemetry-util-genai==0.4b.0 # when updating, also update in pyproject.toml + +-e instrumentation-genai/opentelemetry-instrumentation-crewai diff --git a/instrumentation-genai/opentelemetry-instrumentation-crewai/tests/test_agent_span.py b/instrumentation-genai/opentelemetry-instrumentation-crewai/tests/test_agent_span.py new file mode 100644 index 0000000000..9695dbb602 --- /dev/null +++ b/instrumentation-genai/opentelemetry-instrumentation-crewai/tests/test_agent_span.py @@ -0,0 +1,54 @@ +# Copyright The OpenTelemetry Authors +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +import pytest + +from opentelemetry.instrumentation.crewai._wrappers import ( + wrap_agent_execute_task, +) +from opentelemetry.semconv.attributes import error_attributes +from opentelemetry.trace import StatusCode + +from .conftest import fake_agent, fake_task + + +def test_agent_span_nests_under_active_kickoff_span(handler_and_exporter): + handler, exporter = handler_and_exporter + wrapper = wrap_agent_execute_task(handler) + agent = fake_agent() + task = fake_task() + + kickoff_invocation = handler.start_workflow(name="crew.kickoff") + wrapper(lambda *a, **k: "agent output", agent, (task,), {}) + kickoff_invocation.stop() + + spans = exporter.get_finished_spans() + names = {s.name: s for s in spans} + assert set(names) == { + "invoke_workflow crew.kickoff", + "invoke_agent Researcher", + } + agent_span = names["invoke_agent Researcher"] + kickoff_span = names["invoke_workflow crew.kickoff"] + assert agent_span.parent.span_id == kickoff_span.context.span_id + assert agent_span.attributes["crewai.agent.role"] == "Researcher" + + +def test_agent_span_records_exception_and_reraises(handler_and_exporter): + handler, exporter = handler_and_exporter + wrapper = wrap_agent_execute_task(handler) + agent = fake_agent() + task = fake_task() + + def wrapped(*args, **kwargs): + raise RuntimeError("agent failed") + + with pytest.raises(RuntimeError, match="agent failed"): + wrapper(wrapped, agent, (task,), {}) + + (span,) = exporter.get_finished_spans() + assert span.name == "invoke_agent Researcher" + assert span.status.status_code == StatusCode.ERROR + assert span.attributes[error_attributes.ERROR_TYPE] == "RuntimeError" diff --git a/instrumentation-genai/opentelemetry-instrumentation-crewai/tests/test_handoff_links.py b/instrumentation-genai/opentelemetry-instrumentation-crewai/tests/test_handoff_links.py new file mode 100644 index 0000000000..332b379539 --- /dev/null +++ b/instrumentation-genai/opentelemetry-instrumentation-crewai/tests/test_handoff_links.py @@ -0,0 +1,127 @@ +# Copyright The OpenTelemetry Authors +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +from crewai.utilities.constants import NOT_SPECIFIED + +from opentelemetry.instrumentation.crewai import _wrappers +from opentelemetry.instrumentation.crewai._wrappers import ( + _reset_handoff_state, + wrap_agent_execute_task, + wrap_crew_kickoff, +) + +from .conftest import fake_agent, fake_crew, fake_task + + +def test_second_agent_links_to_first_when_context_explicit( + handler_and_exporter, +): + handler, exporter = handler_and_exporter + crew = fake_crew() + agent_a = fake_agent(role="Researcher", crew=crew) + agent_b = fake_agent(role="Writer", crew=crew) + task_a = fake_task(context=None) + task_b = fake_task(context=[task_a]) + + wrapper = wrap_agent_execute_task(handler) + _reset_handoff_state(crew) + + kickoff_invocation = handler.start_workflow(name="crew.kickoff") + wrapper(lambda *a, **k: "research output", agent_a, (task_a,), {}) + wrapper(lambda *a, **k: "final output", agent_b, (task_b,), {}) + kickoff_invocation.stop() + + spans = { + s.attributes.get("crewai.agent.role"): s + for s in exporter.get_finished_spans() + } + span_a, span_b = spans["Researcher"], spans["Writer"] + kickoff_span = spans[None] + + # It's a link, not a parent-child relationship: both are direct children + # of the crew.kickoff span, not of each other. + assert span_a.parent.span_id == kickoff_span.context.span_id + assert span_b.parent.span_id == kickoff_span.context.span_id + assert span_b.parent.span_id != span_a.context.span_id + + assert len(span_b.links) == 1 + assert span_b.links[0].context.span_id == span_a.context.span_id + assert span_a.links == () + + +def test_default_context_links_to_every_prior_task(handler_and_exporter): + """Task.context left unset (NOT_SPECIFIED) means CrewAI feeds every prior + task's output into this one -- the handoff links should match.""" + handler, exporter = handler_and_exporter + crew = fake_crew() + agent_a = fake_agent(role="A", crew=crew) + agent_b = fake_agent(role="B", crew=crew) + agent_c = fake_agent(role="C", crew=crew) + task_a = fake_task(context=None) + task_b = fake_task(context=None) + task_c = fake_task(context=NOT_SPECIFIED) + + wrapper = wrap_agent_execute_task(handler) + _reset_handoff_state(crew) + + kickoff_invocation = handler.start_workflow(name="crew.kickoff") + wrapper(lambda *a, **k: "a", agent_a, (task_a,), {}) + wrapper(lambda *a, **k: "b", agent_b, (task_b,), {}) + wrapper(lambda *a, **k: "c", agent_c, (task_c,), {}) + kickoff_invocation.stop() + + spans = { + s.attributes.get("crewai.agent.role"): s + for s in exporter.get_finished_spans() + } + linked_ids = {link.context.span_id for link in spans["C"].links} + assert linked_ids == { + spans["A"].context.span_id, + spans["B"].context.span_id, + } + + +def test_no_context_means_no_links(handler_and_exporter): + handler, exporter = handler_and_exporter + crew = fake_crew() + agent_a = fake_agent(role="A", crew=crew) + agent_b = fake_agent(role="B", crew=crew) + task_a = fake_task(context=None) + task_b = fake_task(context=None) # explicit "no dependency" + + wrapper = wrap_agent_execute_task(handler) + _reset_handoff_state(crew) + + kickoff_invocation = handler.start_workflow(name="crew.kickoff") + wrapper(lambda *a, **k: "a", agent_a, (task_a,), {}) + wrapper(lambda *a, **k: "b", agent_b, (task_b,), {}) + kickoff_invocation.stop() + + spans = { + s.attributes.get("crewai.agent.role"): s + for s in exporter.get_finished_spans() + } + assert spans["B"].links == () + + +def test_handoff_state_is_scoped_per_crew_run(handler_and_exporter): + """wrap_crew_kickoff must clear id(crew)'s entry out of the shared, + module-level handoff dict once the run ends, so state from one run can + never leak into a later, unrelated run against the same Crew object.""" + handler, _exporter = handler_and_exporter + crew = fake_crew() + kickoff_wrapper = wrap_crew_kickoff(handler) + agent_wrapper = wrap_agent_execute_task(handler) + agent = fake_agent(role="A", crew=crew) + task = fake_task(context=None) + + def run_with_one_agent_task(*args, **kwargs): + assert id(crew) in _wrappers._handoff_spans + agent_wrapper(lambda *a, **k: "out", agent, (task,), {}) + return "crew done" + + kickoff_wrapper(run_with_one_agent_task, crew, (), {}) + + assert id(crew) not in _wrappers._handoff_spans diff --git a/instrumentation-genai/opentelemetry-instrumentation-crewai/tests/test_kickoff_span.py b/instrumentation-genai/opentelemetry-instrumentation-crewai/tests/test_kickoff_span.py new file mode 100644 index 0000000000..7f76544758 --- /dev/null +++ b/instrumentation-genai/opentelemetry-instrumentation-crewai/tests/test_kickoff_span.py @@ -0,0 +1,56 @@ +# Copyright The OpenTelemetry Authors +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +import pytest + +from opentelemetry.instrumentation.crewai._wrappers import wrap_crew_kickoff +from opentelemetry.semconv.attributes import error_attributes +from opentelemetry.trace import StatusCode + +from .conftest import fake_agent, fake_crew, fake_task + + +def test_kickoff_produces_one_span_with_expected_attributes( + handler_and_exporter, +): + handler, exporter = handler_and_exporter + crew = fake_crew( + agents=[fake_agent(), fake_agent(role="Writer")], + tasks=[fake_task(), fake_task()], + process="sequential", + crew_id="crew-123", + ) + wrapper = wrap_crew_kickoff(handler) + + def wrapped(*args, **kwargs): + return "crew output" + + result = wrapper(wrapped, crew, (), {}) + + assert result == "crew output" + spans = exporter.get_finished_spans() + assert len(spans) == 1 + (span,) = spans + assert span.name == "invoke_workflow crew.kickoff" + assert span.attributes["crewai.crew.id"] == "crew-123" + assert span.attributes["crewai.crew.process"] == "sequential" + assert span.attributes["crewai.crew.agent_count"] == 2 + assert span.attributes["crewai.crew.task_count"] == 2 + + +def test_kickoff_records_exception_and_reraises(handler_and_exporter): + handler, exporter = handler_and_exporter + crew = fake_crew() + wrapper = wrap_crew_kickoff(handler) + + def wrapped(*args, **kwargs): + raise ValueError("boom") + + with pytest.raises(ValueError, match="boom"): + wrapper(wrapped, crew, (), {}) + + (span,) = exporter.get_finished_spans() + assert span.status.status_code == StatusCode.ERROR + assert span.attributes[error_attributes.ERROR_TYPE] == "ValueError" diff --git a/instrumentation-genai/opentelemetry-instrumentation-crewai/tests/test_llm_span.py b/instrumentation-genai/opentelemetry-instrumentation-crewai/tests/test_llm_span.py new file mode 100644 index 0000000000..6f093aaf85 --- /dev/null +++ b/instrumentation-genai/opentelemetry-instrumentation-crewai/tests/test_llm_span.py @@ -0,0 +1,207 @@ +# Copyright The OpenTelemetry Authors +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +import threading +import time +from types import SimpleNamespace + +import pytest + +from opentelemetry.instrumentation.crewai._wrappers import ( + _pending_llm_calls, + on_llm_call_completed, + wrap_llm_call, +) +from opentelemetry.semconv._incubating.attributes import gen_ai_attributes +from opentelemetry.semconv.attributes import error_attributes +from opentelemetry.trace import StatusCode + +from .conftest import fake_llm + + +def _fake_completed_event(**overrides): + defaults = dict( + usage={ + "prompt_tokens": 12, + "completion_tokens": 34, + "total_tokens": 46, + }, + finish_reason="stop", + ) + defaults.update(overrides) + return SimpleNamespace(**defaults) + + +def test_llm_span_has_genai_attributes(handler_and_exporter): + handler, exporter = handler_and_exporter + llm = fake_llm(model="gpt-4o-mini", provider="openai") + wrapper = wrap_llm_call(handler) + + def wrapped(*args, **kwargs): + # Simulates crewai_event_bus dispatching LLMCallCompletedEvent + # before call() returns, just as the real BaseLLM subclasses do. + on_llm_call_completed(llm, _fake_completed_event()) + return "the answer" + + result = wrapper(wrapped, llm, ("hello",), {}) + + assert result == "the answer" + (span,) = exporter.get_finished_spans() + assert span.name == "chat gpt-4o-mini" + assert ( + span.attributes[gen_ai_attributes.GEN_AI_REQUEST_MODEL] + == "gpt-4o-mini" + ) + assert span.attributes[gen_ai_attributes.GEN_AI_PROVIDER_NAME] == "openai" + assert span.attributes[gen_ai_attributes.GEN_AI_USAGE_INPUT_TOKENS] == 12 + assert span.attributes[gen_ai_attributes.GEN_AI_USAGE_OUTPUT_TOKENS] == 34 + assert span.attributes[ + gen_ai_attributes.GEN_AI_RESPONSE_FINISH_REASONS + ] == ("stop",) + + +def test_llm_span_omits_content_by_default(handler_and_exporter): + handler, exporter = handler_and_exporter + llm = fake_llm() + wrapper = wrap_llm_call(handler) + + def wrapped(*args, **kwargs): + on_llm_call_completed(llm, _fake_completed_event()) + return "secret answer" + + wrapper(wrapped, llm, ("secret prompt",), {}) + + (span,) = exporter.get_finished_spans() + assert gen_ai_attributes.GEN_AI_INPUT_MESSAGES not in span.attributes + assert gen_ai_attributes.GEN_AI_OUTPUT_MESSAGES not in span.attributes + + +def test_llm_span_captures_content_when_flag_enabled( + content_capturing_handler_and_exporter, +): + handler, exporter = content_capturing_handler_and_exporter + llm = fake_llm() + wrapper = wrap_llm_call(handler) + + def wrapped(*args, **kwargs): + on_llm_call_completed(llm, _fake_completed_event()) + return "the real answer" + + wrapper(wrapped, llm, ("the real prompt",), {}) + + (span,) = exporter.get_finished_spans() + assert ( + "the real prompt" + in span.attributes[gen_ai_attributes.GEN_AI_INPUT_MESSAGES] + ) + assert ( + "the real answer" + in span.attributes[gen_ai_attributes.GEN_AI_OUTPUT_MESSAGES] + ) + + +def test_llm_span_records_exception_and_reraises(handler_and_exporter): + handler, exporter = handler_and_exporter + llm = fake_llm() + wrapper = wrap_llm_call(handler) + + def wrapped(*args, **kwargs): + raise TimeoutError("provider timed out") + + with pytest.raises(TimeoutError, match="provider timed out"): + wrapper(wrapped, llm, ("hi",), {}) + + (span,) = exporter.get_finished_spans() + assert span.name == "chat gpt-4o-mini" + assert span.status.status_code == StatusCode.ERROR + assert span.attributes[error_attributes.ERROR_TYPE] == "TimeoutError" + + +def test_concurrent_calls_on_different_instances_do_not_cross_talk( + handler_and_exporter, +): + """The common concurrent case: two agents, each with their own LLM + instance, calling the model at the same time. id(instance) differs, so + each call must land its own usage data on its own span -- never the + other's.""" + handler, exporter = handler_and_exporter + llm_a = fake_llm(model="model-a") + llm_b = fake_llm(model="model-b") + wrapper = wrap_llm_call(handler) + barrier = threading.Barrier(2) + + def run(llm, tag, delay): + def wrapped(*args, **kwargs): + barrier.wait(timeout=5) # force real overlap between the two calls + time.sleep(delay) + on_llm_call_completed( + llm, + _fake_completed_event( + usage={"prompt_tokens": tag, "completion_tokens": tag} + ), + ) + return f"answer-{tag}" + + wrapper(wrapped, llm, (f"prompt-{tag}",), {}) + + thread_a = threading.Thread(target=run, args=(llm_a, 1, 0.05)) + thread_b = threading.Thread(target=run, args=(llm_b, 2, 0.0)) + thread_a.start() + thread_b.start() + thread_a.join(timeout=5) + thread_b.join(timeout=5) + + spans = { + s.attributes[gen_ai_attributes.GEN_AI_REQUEST_MODEL]: s + for s in exporter.get_finished_spans() + } + assert ( + spans["model-a"].attributes[ + gen_ai_attributes.GEN_AI_USAGE_INPUT_TOKENS + ] + == 1 + ) + assert ( + spans["model-b"].attributes[ + gen_ai_attributes.GEN_AI_USAGE_INPUT_TOKENS + ] + == 2 + ) + assert ( + not _pending_llm_calls + ) # no leaked queue entries after both complete + + +def test_back_to_back_calls_on_shared_instance_do_not_leak_state( + handler_and_exporter, +): + """A single LLM instance reused for two sequential (non-overlapping) + calls -- the realistic "shared LLM object" case when calls don't + actually overlap. The second call must not see the first call's event.""" + handler, exporter = handler_and_exporter + llm = fake_llm() + wrapper = wrap_llm_call(handler) + + def wrapped_one(*args, **kwargs): + on_llm_call_completed(llm, _fake_completed_event(finish_reason="stop")) + return "first" + + def wrapped_two(*args, **kwargs): + on_llm_call_completed( + llm, _fake_completed_event(finish_reason="length") + ) + return "second" + + wrapper(wrapped_one, llm, ("p1",), {}) + wrapper(wrapped_two, llm, ("p2",), {}) + + span_one, span_two = exporter.get_finished_spans() + assert span_one.attributes[ + gen_ai_attributes.GEN_AI_RESPONSE_FINISH_REASONS + ] == ("stop",) + assert span_two.attributes[ + gen_ai_attributes.GEN_AI_RESPONSE_FINISH_REASONS + ] == ("length",) + assert not _pending_llm_calls diff --git a/instrumentation-genai/opentelemetry-instrumentation-crewai/tests/test_tool_span.py b/instrumentation-genai/opentelemetry-instrumentation-crewai/tests/test_tool_span.py new file mode 100644 index 0000000000..9734ba4d86 --- /dev/null +++ b/instrumentation-genai/opentelemetry-instrumentation-crewai/tests/test_tool_span.py @@ -0,0 +1,67 @@ +# Copyright The OpenTelemetry Authors +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +import pytest + +from opentelemetry.instrumentation.crewai._wrappers import wrap_tool_run +from opentelemetry.semconv._incubating.attributes import gen_ai_attributes +from opentelemetry.semconv.attributes import error_attributes +from opentelemetry.trace import StatusCode + +from .conftest import fake_tool + + +def test_tool_span_has_tool_name(handler_and_exporter): + handler, exporter = handler_and_exporter + tool = fake_tool(name="word_count") + wrapper = wrap_tool_run(handler) + + result = wrapper( + lambda *a, **k: "42 words", tool, (), {"text": "hello world"} + ) + + assert result == "42 words" + (span,) = exporter.get_finished_spans() + assert span.name == "execute_tool word_count" + assert span.attributes[gen_ai_attributes.GEN_AI_TOOL_NAME] == "word_count" + assert gen_ai_attributes.GEN_AI_TOOL_CALL_ARGUMENTS not in span.attributes + assert gen_ai_attributes.GEN_AI_TOOL_CALL_RESULT not in span.attributes + + +def test_tool_span_captures_arguments_and_result_when_flag_enabled( + content_capturing_handler_and_exporter, +): + handler, exporter = content_capturing_handler_and_exporter + tool = fake_tool(name="word_count") + wrapper = wrap_tool_run(handler) + + wrapper(lambda *a, **k: "42 words", tool, (), {"text": "hello world"}) + + (span,) = exporter.get_finished_spans() + assert ( + "hello world" + in span.attributes[gen_ai_attributes.GEN_AI_TOOL_CALL_ARGUMENTS] + ) + assert ( + span.attributes[gen_ai_attributes.GEN_AI_TOOL_CALL_RESULT] + == '"42 words"' + ) + + +def test_tool_span_records_exception_and_reraises(handler_and_exporter): + handler, exporter = handler_and_exporter + tool = fake_tool() + wrapper = wrap_tool_run(handler) + + def wrapped(*args, **kwargs): + raise ValueError("bad tool input") + + with pytest.raises(ValueError, match="bad tool input"): + wrapper(wrapped, tool, (), {}) + + (span,) = exporter.get_finished_spans() + assert span.name == "execute_tool word_count" + assert span.status.status_code == StatusCode.ERROR + assert span.attributes[error_attributes.ERROR_TYPE] == "ValueError" diff --git a/pyproject.toml b/pyproject.toml index 3ecc82f262..c56c63d3b8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -207,6 +207,7 @@ include = [ "instrumentation/opentelemetry-instrumentation-threading", "instrumentation-genai/opentelemetry-instrumentation-anthropic", "instrumentation-genai/opentelemetry-instrumentation-claude-agent-sdk", + "instrumentation-genai/opentelemetry-instrumentation-crewai", "instrumentation-genai/opentelemetry-instrumentation-vertexai", "instrumentation-genai/opentelemetry-instrumentation-langchain", "instrumentation-genai/opentelemetry-instrumentation-weaviate", @@ -227,6 +228,8 @@ exclude = [ "instrumentation-genai/opentelemetry-instrumentation-anthropic/examples/**/*.py", "instrumentation-genai/opentelemetry-instrumentation-claude-agent-sdk/tests/**/*.py", "instrumentation-genai/opentelemetry-instrumentation-claude-agent-sdk/examples/**/*.py", + "instrumentation-genai/opentelemetry-instrumentation-crewai/tests/**/*.py", + "instrumentation-genai/opentelemetry-instrumentation-crewai/examples/**/*.py", "instrumentation-genai/opentelemetry-instrumentation-vertexai/tests/**/*.py", "instrumentation-genai/opentelemetry-instrumentation-vertexai/examples/**/*.py", "instrumentation-genai/opentelemetry-instrumentation-langchain/tests/**/*.py", diff --git a/tox.ini b/tox.ini index 85dd4e9b0c..52975f3f62 100644 --- a/tox.ini +++ b/tox.ini @@ -41,6 +41,12 @@ envlist = # pypy3-test-instrumentation-claude-agent-sdk-{oldest,latest} lint-instrumentation-claude-agent-sdk + ; instrumentation-crewai + py3{10,11,12,13}-test-instrumentation-crewai-{oldest,latest} + # Disabling pypy3: not yet tested against crewai's dependency stack. + # pypy3-test-instrumentation-crewai-{oldest,latest} + lint-instrumentation-crewai + ; opentelemetry-resource-detector-containerid py3{10,11,12,13,14}-test-resource-detector-containerid pypy3-test-resource-detector-containerid @@ -514,6 +520,11 @@ deps = claude-agent-sdk-latest: -r {toxinidir}/instrumentation-genai/opentelemetry-instrumentation-claude-agent-sdk/tests/requirements.latest.txt lint-instrumentation-claude-agent-sdk: -r {toxinidir}/instrumentation-genai/opentelemetry-instrumentation-claude-agent-sdk/tests/requirements.oldest.txt + crewai-oldest: -r {toxinidir}/instrumentation-genai/opentelemetry-instrumentation-crewai/tests/requirements.oldest.txt + crewai-latest: {[testenv]test_deps} + crewai-latest: -r {toxinidir}/instrumentation-genai/opentelemetry-instrumentation-crewai/tests/requirements.latest.txt + lint-instrumentation-crewai: -r {toxinidir}/instrumentation-genai/opentelemetry-instrumentation-crewai/tests/requirements.oldest.txt + asgi: {[testenv]test_deps} asgi: -r {toxinidir}/instrumentation/opentelemetry-instrumentation-asgi/test-requirements.txt @@ -940,6 +951,9 @@ commands = test-instrumentation-claude-agent-sdk: pytest {toxinidir}/instrumentation-genai/opentelemetry-instrumentation-claude-agent-sdk/tests --vcr-record=none {posargs} lint-instrumentation-claude-agent-sdk: sh -c "cd instrumentation-genai && pylint --rcfile ../.pylintrc opentelemetry-instrumentation-claude-agent-sdk" + test-instrumentation-crewai: pytest {toxinidir}/instrumentation-genai/opentelemetry-instrumentation-crewai/tests {posargs} + lint-instrumentation-crewai: sh -c "cd instrumentation-genai && pylint --rcfile ../.pylintrc opentelemetry-instrumentation-crewai" + test-instrumentation-sio-pika: pytest {toxinidir}/instrumentation/opentelemetry-instrumentation-pika/tests {posargs} lint-instrumentation-sio-pika: sh -c "cd instrumentation && pylint --rcfile ../.pylintrc opentelemetry-instrumentation-pika" @@ -1183,6 +1197,7 @@ deps = {toxinidir}/instrumentation-genai/opentelemetry-instrumentation-anthropic[instruments] {toxinidir}/instrumentation-genai/opentelemetry-instrumentation-langchain[instruments] {toxinidir}/instrumentation-genai/opentelemetry-instrumentation-claude-agent-sdk[instruments] + {toxinidir}/instrumentation-genai/opentelemetry-instrumentation-crewai[instruments] {toxinidir}/instrumentation/opentelemetry-instrumentation-aiokafka[instruments] {toxinidir}/instrumentation/opentelemetry-instrumentation-asyncclick[instruments] {toxinidir}/exporter/opentelemetry-exporter-credential-provider-gcp diff --git a/util/opentelemetry-util-genai/.changelog/4860.fixed b/util/opentelemetry-util-genai/.changelog/4860.fixed new file mode 100644 index 0000000000..f41d908a7c --- /dev/null +++ b/util/opentelemetry-util-genai/.changelog/4860.fixed @@ -0,0 +1 @@ +`opentelemetry-util-genai`: fix `ToolInvocation` not setting `gen_ai.tool.call.result` on the span, and JSON-encode `arguments`/`tool_result` instead of passing raw Python values diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_tool_invocation.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_tool_invocation.py index f9d802b008..f4a12e3951 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_tool_invocation.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_tool_invocation.py @@ -13,6 +13,7 @@ from opentelemetry.util.genai._invocation import Error, GenAIInvocation from opentelemetry.util.genai.completion_hook import CompletionHook from opentelemetry.util.genai.metrics import InvocationMetricsRecorder +from opentelemetry.util.genai.utils import gen_ai_json_dumps class ToolInvocation(GenAIInvocation): @@ -94,7 +95,18 @@ def _apply_finish(self, error: Error | None = None) -> None: (GenAI.GEN_AI_TOOL_CALL_ID, self.tool_call_id), (GenAI.GEN_AI_TOOL_TYPE, self.tool_type), (GenAI.GEN_AI_TOOL_DESCRIPTION, self.tool_description), - (GenAI.GEN_AI_TOOL_CALL_ARGUMENTS, self.arguments), + ( + GenAI.GEN_AI_TOOL_CALL_ARGUMENTS, + gen_ai_json_dumps(self.arguments) + if self.arguments is not None + else None, + ), + ( + GenAI.GEN_AI_TOOL_CALL_RESULT, + gen_ai_json_dumps(self.tool_result) + if self.tool_result is not None + else None, + ), ) attributes: dict[str, Any] = { GenAI.GEN_AI_OPERATION_NAME: self._operation_name, diff --git a/util/opentelemetry-util-genai/tests/test_toolcall.py b/util/opentelemetry-util-genai/tests/test_toolcall.py index 54648f061a..f530712551 100644 --- a/util/opentelemetry-util-genai/tests/test_toolcall.py +++ b/util/opentelemetry-util-genai/tests/test_toolcall.py @@ -131,6 +131,24 @@ def test_server_tool_call_in_message(): assert isinstance(msg.parts[1], ServerToolCallResponse) +def test_tool_result_is_set_on_span_at_finish(): + """gen_ai.tool.call.result must actually reach the span once the + invocation is stopped, not just be stored on the dataclass field.""" + span_exporter = InMemorySpanExporter() + provider = TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(span_exporter)) + handler = TelemetryHandler(tracer_provider=provider) + + invocation = handler.start_tool("get_weather") + invocation.tool_result = {"temperature": 70} + invocation.stop() + + (span,) = span_exporter.get_finished_spans() + assert ( + span.attributes[GenAI.GEN_AI_TOOL_CALL_RESULT] == '{"temperature":70}' + ) + + if __name__ == "__main__": pytest.main([__file__, "-v"])