### Install Monocle from Source Source: https://github.com/monocle2ai/monocle/blob/main/Monocle_User_Guide.md Build and install the Monocle library locally from source using pip. Also shows how to install optional test dependencies. ```bash pip install . ``` ```bash pip install -e ".[dev]" ``` -------------------------------- ### Install development dependencies Source: https://github.com/monocle2ai/monocle/blob/main/Monocle_committer_guide.md Install the package in editable mode with development dependencies. ```bash monocle% pip install -e '.[dev]' ``` -------------------------------- ### Build and Install Monocle Locally Source: https://github.com/monocle2ai/monocle/blob/main/apptrace/src/README.md Build and install Monocle from source. This includes installing development dependencies. ```bash pip install . pip install -e ".[dev]" python3 -m pip install pipenv pipenv install build ``` -------------------------------- ### Run Monocle MCP Server with StdIO Source: https://github.com/monocle2ai/monocle/blob/main/mcp/README.md Use this command in your bash terminal to start the Monocle MCP server. Ensure monocle-apptrace is installed. ```bash monocle_apptrace ``` -------------------------------- ### Activate virtual environment Source: https://github.com/monocle2ai/monocle/blob/main/Monocle_committer_guide.md Prepare the development environment by installing pipenv and activating the virtual environment. ```bash cd monocle monocle% python -m pip install pipenv monocle% pipenv --python 3.11.9 monocle% source $(pipenv --venv)/bin/activate ``` -------------------------------- ### Install Monocle from PyPI Source: https://github.com/monocle2ai/monocle/blob/main/apptrace/src/README.md Install Monocle using pip. Ensure pipenv is installed first. ```bash python3 -m pip install pipenv pip install monocle-apptrace ``` -------------------------------- ### Setup Monocle Telemetry for LlamaIndex Source: https://github.com/monocle2ai/monocle/blob/main/apptrace/tests/llama_starter.md Configure Monocle telemetry by importing `setup_monocle_telemetry`. The `workflow_name` is mandatory. This example shows how to instrument the `chat` method of `llama_index.llms.openai.base.OpenAI`. ```python from monocle_apptrace.instrumentation.common.instrumentor import setup_monocle_telemetry from monocle_apptrace.instrumentation.common.wrapper_method import WrapperMethod from monocle_apptrace.instrumentation.common.wrapper import task_wrapper from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter setup_monocle_telemetry( workflow_name="llama_index_1", wrapper_methods=[ WrapperMethod( package="llama_index.llms.openai.base", object_name="OpenAI", method="chat", span_name="llamaindex.openai", wrapper_method=task_wrapper), ] ) ``` -------------------------------- ### Install Monocle Apptrace Source: https://github.com/monocle2ai/monocle/blob/main/Monocle_User_Guide.md Install the Monocle library using pip. Use the 'azure' or 'aws' extras for cloud support. ```bash pip install monocle_apptrace ``` ```bash pip install monocle_apptrace[azure] ``` ```bash pip install monocle_apptrace[aws] ``` -------------------------------- ### Initialize Monocle Instrumentation Source: https://github.com/monocle2ai/monocle/blob/main/apptrace/README.md Call the setup function within your main application entry point, providing a unique workflow name. ```python setup_monocle_telemetry(workflow_name="your-app-name") ``` -------------------------------- ### Quick Start: Test Travel Agent Source: https://github.com/monocle2ai/monocle/blob/main/test_tools/README.md Example of how to set up and run test cases for a travel booking agent using MonocleValidator. It includes defining test cases with inputs, expected outputs, and tool invocation checks. ```python from monocle_test_tools import TestCase, MonocleValidator from adk_travel_agent import root_travel_agent # Test cases for testing travel booking agent agent_test_cases:list[TestCase] = [ { "test_input": ["Book a flight from San Francisco to Mumbai for 26th Nov 2025. Book a two queen room at Marriot Intercontinental at Juhu, Mumbai for 27th Nov 2025 for 4 nights."], "test_output": "A flight from San Francisco to Mumbai has been booked, along with a four night stay in a two queen room at the Marriot Intercontinental in Juhu, Mumbai, starting November 27th, 2025.", "comparer": "similarity", }, { "test_input": ["Book a flight from San Francisco to Mumbai for 26th Nov 2025. Book a two queen room at Marriot Intercontinental at Juhu, Mumbai for 27th Nov 2025 for 4 nights."], "test_spans": [ { "span_type": "agentic.tool.invocation", "entities": [ {"type": "tool", "name": "adk_book_hotel"}, {"type": "agent", "name": "adk_hotel_booking_agent"} ], } ] } ] # Run test cases using Monocle test framework @MonocleValidator().monocle_testcase(agent_test_cases) async def test_run_workflows(my_test_case: TestCase): await MonocleValidator().test_workflow_async(root_travel_agent, my_test_case) if __name__ == "__main__": pytest.main([__file__]) ``` -------------------------------- ### Import Monocle Telemetry Source: https://github.com/monocle2ai/monocle/blob/main/apptrace/README.md Import the setup function from the instrumentor module to begin configuration. ```python from monocle_apptrace.instrumentor import setup_monocle_telemetry ``` -------------------------------- ### Install Monocle package Source: https://github.com/monocle2ai/monocle/blob/main/Monocle_committer_guide.md Install the package using pipenv after configuring credentials. ```bash > python3 -m pip install pipenv > pipenv install monocle-apptrace ``` -------------------------------- ### Setup Monocle Telemetry Source: https://context7.com/monocle2ai/monocle/llms.txt Initializes Monocle tracing for your application. Use the basic setup for saving traces to the local .monocle folder, or configure with console output for debugging. Custom exporters can be specified by overriding the MONOCLE_EXPORTER environment variable. ```python from monocle_apptrace.instrumentor import setup_monocle_telemetry from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter # Basic setup - traces saved to local .monocle folder setup_monocle_telemetry(workflow_name="my_chatbot_app") # Setup with console output for debugging setup_monocle_telemetry( workflow_name="my_chatbot_app", span_processors=[BatchSpanProcessor(ConsoleSpanExporter())] ) # Setup with custom exporter list (overrides MONOCLE_EXPORTER env var) setup_monocle_telemetry( workflow_name="my_chatbot_app", monocle_exporters_list="file,console" ) ``` ```python # Complete example with LangChain from langchain.chains import LLMChain from langchain_openai import OpenAI from langchain.prompts import PromptTemplate setup_monocle_telemetry(workflow_name="simple_math_app") llm = OpenAI() prompt = PromptTemplate.from_template("1 + {number} = ") chain = LLMChain(llm=llm, prompt=prompt) result = chain.invoke({"number": 2}) # Traces are automatically captured and exported ``` -------------------------------- ### Install Monocle Package Source: https://github.com/monocle2ai/monocle/blob/main/apptrace/README.md Use pip to install the monocle_apptrace package into your environment. ```bash pip install monocle_apptrace ``` -------------------------------- ### Install Monocle Observability Source: https://github.com/monocle2ai/monocle/blob/main/apptrace/tests/langchain_starter.md Install the Monocle observability package using pip. This is the first step to integrating Monocle into your project. ```shell pip install monocle-observability ``` -------------------------------- ### Evaluate Agent Sessions with Combined Expected and not_expected Source: https://github.com/monocle2ai/monocle/blob/main/docs/monocle_evaluation_api.md This example demonstrates evaluating agent sessions using both `expected` and `not_expected` parameters. It checks for role adherence and task completion status. ```python await monocle_trace_asserter.run_agent_async(root_agent, "google_adk", "query") # Role adherence must be excellent or good, NOT poor or none monocle_trace_asserter.with_evaluation("okahu") .check_eval(fact_name="agent_sessions", eval_name="role_adherence", expected=["excellent_adherence", "good_adherence"], not_expected=["poor_adherence", "no_adherence"]) # Task must be completed or partially completed, NOT failed monocle_trace_asserter.check_eval(fact_name="agent_sessions", eval_name="mcp_task_completion", expected=["completed", "partially_completed"], not_expected="failed") ``` -------------------------------- ### Module-Scoped Pytest Fixture Example Source: https://github.com/monocle2ai/monocle/blob/main/apptrace/tests/MONOCLE_TESTING.md Illustrates a module-scoped pytest fixture. Use with caution, as it shares setup across all tests in a module. Cleanup is still required to avoid affecting other modules. ```python @pytest.fixture(scope="module") def setup(): # Shared setup across all tests in the module # Still need cleanup to avoid affecting other modules ``` -------------------------------- ### Function-Scoped Pytest Fixture Example Source: https://github.com/monocle2ai/monocle/blob/main/apptrace/tests/MONOCLE_TESTING.md Demonstrates a function-scoped pytest fixture for isolated test setup. This ensures each test runs with a fresh instrumentation state, preventing cross-test contamination. ```python @pytest.fixture(scope="function") def setup(): # Fresh setup for each test # Prevents cross-test contamination ``` -------------------------------- ### Install Monocle Test Tools Source: https://github.com/monocle2ai/monocle/blob/main/test_tools/README.md Install the monocle_test_tools package using pip. This command is used for setting up the testing framework. ```bash pip install monocle_test_tools ``` -------------------------------- ### Python Class with LLM Example Source: https://github.com/monocle2ai/monocle/blob/main/docs/design.md An example Python class demonstrating a typical structure for an LLM interaction, including a model name and a function to call the API. ```python class My_LLM(): model_name = "gpt-4" def call_llm(prompt_text): llm_response = make_api_call(prompt_text) return llm_response ``` -------------------------------- ### Setup Monocle Telemetry for Langchain Source: https://github.com/monocle2ai/monocle/blob/main/apptrace/tests/langchain_starter.md Import and setup Monocle telemetry in your application. The `workflow_name` parameter is mandatory and identifies your application. ```python from monocle_apptrace.instrumentation.common.instrumentor import setup_monocle_telemetry setup_monocle_telemetry( workflow_name="langchain_app_1") ``` -------------------------------- ### Validate Token Usage Source: https://github.com/monocle2ai/monocle/blob/main/docs/monocle_evaluation_api.md This example demonstrates how to validate token usage for AI agents, ensuring that the total token consumption does not exceed a specified limit. This is crucial for cost and performance monitoring. ```python await monocle_trace_asserter.run_agent_async(root_agent, "google_adk", "query") # Assert total token usage does not exceed 1500 tokens monocle_trace_asserter.under_token_limit(1500) ``` -------------------------------- ### Full LangGraph Example with Monocle Tracing Source: https://context7.com/monocle2ai/monocle/llms.txt Integrate Monocle for comprehensive tracing of LangGraph applications, automatically capturing workflow, agent, tool, and LLM inference spans. Initialize Monocle telemetry with `setup_monocle_telemetry` before defining your LangGraph components. ```python from monocle_apptrace.instrumentor import setup_monocle_telemetry from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, MessagesState, START, END from langgraph.prebuilt import ToolNode, tools_condition import os # Setup Monocle - captures all LangGraph operations automatically setup_monocle_telemetry(workflow_name="langgraph_agent") # Define tools def search_weather(location: str) -> str: """Search for weather information.""" return f"Weather in {location}: 72°F, sunny" def book_restaurant(restaurant: str, time: str) -> str: """Book a restaurant reservation.""" return f"Booked {restaurant} for {time}" tools = [search_weather, book_restaurant] # Create LLM with tools llm = ChatOpenAI(model="gpt-4", temperature=0) llm_with_tools = llm.bind_tools(tools) # Define agent node def agent(state: MessagesState): return {"messages": [llm_with_tools.invoke(state["messages"])]} # Build graph builder = StateGraph(MessagesState) builder.add_node("agent", agent) builder.add_node("tools", ToolNode(tools)) builder.add_edge(START, "agent") builder.add_conditional_edges("agent", tools_condition) builder.add_edge("tools", "agent") graph = builder.compile() # Run the agent - all traces captured automatically result = graph.invoke({ "messages": [("user", "What's the weather in Seattle and book a table at Canlis for 7pm")] }) print(result["messages"][-1].content) # Traces include: workflow span, agent spans, tool spans, LLM inference spans ``` -------------------------------- ### Enable Monocle Tracing in Python Application Source: https://github.com/monocle2ai/monocle/blob/main/Monocle_User_Guide.md Import Monocle and call `setup_monocle_telemetry` to enable tracing for a specified workflow. This example demonstrates integration with Langchain and OpenAI. ```python from monocle_apptrace.instrumentor import setup_monocle_telemetry from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter from langchain.chains import LLMChain from langchain_openai import OpenAI from langchain.prompts import PromptTemplate # Call the setup Monocle telemetry method setup_monocle_telemetry(workflow_name = "simple_math_app") llm = OpenAI() prompt = PromptTemplate.from_template("1 + {number} = ") chain = LLMChain(llm=llm, prompt=prompt) chain.invoke({"number":2}) # Request callbacks: Finally, let's use the request `callbacks` to achieve the same result chain = LLMChain(llm=llm, prompt=prompt) chain.invoke({"number":2}, {"callbacks":[handler]}) ``` -------------------------------- ### Instrument Langchain LLMChain with Monocle Source: https://github.com/monocle2ai/monocle/blob/main/apptrace/src/README.md Instrument a Langchain LLMChain application using Monocle's setup_monocle_telemetry. This example uses ConsoleSpanExporter to display traces. ```python from langchain.chains import LLMChain from langchain_openai import OpenAI from langchain.prompts import PromptTemplate # Import the monocle_apptrace instrumentation method from monocle_apptrace.instrumentor import setup_monocle_telemetry from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter # Call the setup Monocle telemetry method setup_monocle_telemetry(workflow_name = "simple_math_app", span_processors=[BatchSpanProcessor(ConsoleSpanExporter())]) llm = OpenAI() prompt = PromptTemplate.from_template("1 + {number} = ") chain = LLMChain(llm=llm, prompt=prompt) chain.invoke({"number":2}) # Trace is generated when invoke() method is called ``` -------------------------------- ### LlamaIndex Application Code with ChromaDB and OpenAI Source: https://github.com/monocle2ai/monocle/blob/main/apptrace/tests/llama_starter.md Example application code demonstrating the use of LlamaIndex with ChromaDB for vector storage and OpenAI for embeddings and language models. Ensure you have a 'data' directory with documents. ```python import chromadb from llama_index.vector_stores.chroma import ChromaVectorStore from llama_index.llms.openai import OpenAI from llama_index.core import StorageContext from llama_index.core import VectorStoreIndex, SimpleDirectoryReader from llama_index.vector_stores.chroma import ChromaVectorStore from llama_index.core import StorageContext from llama_index.embeddings.openai import OpenAIEmbedding import os # Creating a Chroma client # EphemeralClient operates purely in-memory, PersistentClient will also save to disk chroma_client = chromadb.EphemeralClient() chroma_collection = chroma_client.create_collection("quickstart") # construct vector store vector_store = ChromaVectorStore( chroma_collection=chroma_collection, ) dir_path = os.path.dirname(os.path.realpath(__file__)) documents = SimpleDirectoryReader(dir_path + "/data").load_data() embed_model = OpenAIEmbedding(model="text-embedding-3-large") storage_context = StorageContext.from_defaults(vector_store=vector_store) index = VectorStoreIndex.from_documents( documents, storage_context=storage_context, embed_model=embed_model ) llm = OpenAI(temperature=0.1, model="gpt-4") query_engine = index.as_query_engine(llm= llm, ) response = query_engine.query("What did the author do growing up?") ``` -------------------------------- ### Example JSON for Custom Attributes Source: https://github.com/monocle2ai/monocle/blob/main/docs/design.md A JSON configuration demonstrating how to define custom attributes for a 'retrieval' type span. It shows how to specify attribute names and use lambda functions to access data from the arguments dictionary. ```json { "type": "retrieval", "attributes": [ [ { "_comment": "vector store name and type", "attribute": "name", "accessor": "lambda arguments: type(arguments['instance'].vectorstore).__name__" }, { "attribute": "type", "accessor": "lambda arguments: 'vectorstore.'+type(arguments['instance'].vectorstore).__name__" } ] ] } ``` -------------------------------- ### check_eval with Both Expected and Not Expected Parameters Source: https://github.com/monocle2ai/monocle/blob/main/docs/monocle_evaluation_api.md Combine 'expected' and 'not_expected' parameters for comprehensive evaluation criteria. This example defaults to the 'traces' fact. ```python .check_eval("sentiment", expected=["positive", "neutral"], not_expected="negative") ``` -------------------------------- ### Instrumenting a RAG Pattern Source: https://github.com/monocle2ai/monocle/blob/main/apptrace/src/monocle_apptrace/instrumentation/common/tracing.md Example of how anchor spans capture the hierarchy of a RAG chain invocation. ```python response = rag_chain.invoke(prompt) --> cleaned_prompt = llm1.chat(prompt) --> context = vector_store.retrieve(cleaned_prompt) --> response = llm2.chat(system_prompt+context+cleaned_prompt) --> return response ``` -------------------------------- ### Setup Fixture with Environment Cleanup Source: https://github.com/monocle2ai/monocle/blob/main/apptrace/tests/MONOCLE_TESTING.md This fixture sets up Monocle telemetry for integration tests, captures spans using a custom exporter, and ensures proper cleanup of environment variables and instrumentation in a finally block. ```python import os import pytest from common.custom_exporter import CustomConsoleSpanExporter from monocle_apptrace import setup_monocle_telemetry from opentelemetry.sdk.trace.export import SimpleSpanProcessor @pytest.fixture(scope="function") def setup(): # Save original environment state original_env_var = os.environ.get("SOME_CONFIG_PATH") instrumentor = None try: # Setup custom exporter to capture spans custom_exporter = CustomConsoleSpanExporter() # Set test-specific environment if needed os.environ["SOME_CONFIG_PATH"] = "/path/to/test/config" # Setup Monocle telemetry instrumentor = setup_monocle_telemetry( workflow_name="integration_test", span_processors=[SimpleSpanProcessor(custom_exporter)], wrapper_methods=[] ) yield custom_exporter finally: # CRITICAL: Uninstrument to clean up global state if instrumentor and instrumentor.is_instrumented_by_opentelemetry: instrumentor.uninstrument() # Restore original environment state if original_env_var is not None: os.environ["SOME_CONFIG_PATH"] = original_env_var elif "SOME_CONFIG_PATH" in os.environ: del os.environ["SOME_CONFIG_PATH"] ``` -------------------------------- ### Test Trace Generation and Span Validation Source: https://github.com/monocle2ai/monocle/blob/main/apptrace/tests/MONOCLE_TESTING.md This test function utilizes the setup fixture to execute traced operations and then validates that spans were captured by the custom exporter. It also demonstrates resetting the exporter. ```python def test_trace_generation(setup): """Test that demonstrates span capture and validation""" custom_exporter = setup # Execute your traced operations # ... your application code here ... # Validate captured spans spans = custom_exporter.get_captured_spans() assert len(spans) > 0, "Expected spans to be captured" # Reset exporter for next assertion if needed custom_exporter.reset() ``` -------------------------------- ### Manage span lifecycle manually with start_trace and stop_trace Source: https://context7.com/monocle2ai/monocle/llms.txt Provides direct control over span start and end points, useful for complex logic across different code paths. ```python from monocle_apptrace.instrumentation.common.instrumentor import start_trace, stop_trace, setup_monocle_telemetry setup_monocle_telemetry(workflow_name="manual_span_app") def process_with_manual_tracing(data): # Start a trace and get token for later stopping token = start_trace( span_name="complex_processing", attributes={"input.size": len(data)} ) try: # Perform operations intermediate = preprocess(data) result = run_model(intermediate) # Stop trace with final attributes on success stop_trace(token, final_attributes={ "result.status": "success", "output.size": len(result) }) return result except Exception as e: # Stop trace with error attributes on failure stop_trace(token, final_attributes={ "result.status": "error", "error.message": str(e) }) raise ``` -------------------------------- ### Integrate Monocle with Test Fixtures Source: https://github.com/monocle2ai/monocle/blob/main/apptrace/tests/MONOCLE_TESTING.md Uses temporary_env_vars to isolate environment configuration during telemetry instrumentation setup. ```python from config.conftest import temporary_env_vars @pytest.fixture(scope="function") def setup_with_test_env(): instrumentor = None try: with temporary_env_vars( MONOCLE_DEBUG="true", SCOPE_CONFIG_PATH="/test/config" ): custom_exporter = CustomConsoleSpanExporter() instrumentor = setup_monocle_telemetry( workflow_name="isolated_test", span_processors=[SimpleSpanProcessor(custom_exporter)] ) yield custom_exporter finally: if instrumentor and instrumentor.is_instrumented_by_opentelemetry: instrumentor.uninstrument() # Environment variables automatically restored by context managers ``` -------------------------------- ### Setup with Multiple Span Processors Source: https://github.com/monocle2ai/monocle/blob/main/apptrace/tests/MONOCLE_TESTING.md Configures Monocle telemetry to use multiple span processors, such as a custom console exporter and a file exporter. This allows for simultaneous trace capture and storage. ```python def setup_with_multiple_processors(): custom_exporter = CustomConsoleSpanExporter() file_exporter = FileSpanExporter("test_traces.json") instrumentor = None try: instrumentor = setup_monocle_telemetry( workflow_name="multi_processor_test", span_processors=[ SimpleSpanProcessor(custom_exporter), BatchSpanProcessor(file_exporter) ] ) yield custom_exporter finally: if instrumentor and instrumentor.is_instrumented_by_opentelemetry: instrumentor.uninstrument() ``` -------------------------------- ### Use amonocle_trace for Asynchronous Tracing Source: https://github.com/monocle2ai/monocle/blob/main/docs/monocle_trace_api.md This async context manager starts a new trace (span) for asynchronous code blocks. Ensure you are in an async function to use this. ```python from monocle_apptrace.instrumentation.common.instrumentor import amonocle_trace async def my_async_func(): async with amonocle_trace(span_name="async_op"): await do_async_work() ``` -------------------------------- ### Monitor Custom Methods with Monocle in Python Source: https://github.com/monocle2ai/monocle/blob/main/Monocle_User_Guide.md This Python example demonstrates how to extend Monocle's default wrapped methods to monitor custom components. It shows how to set up Monocle telemetry with a custom application name and includes WrapperMethod configurations for 'RunnableParallel' in Langchain. ```python from monocle_apptrace.wrapper import WrapperMethod,task_wrapper,atask_wrapper from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter # extend the default wrapped methods list as follows app_name = "simple_math_app" setup_monocle_telemetry( workflow_name=app_name, span_processors=[BatchSpanProcessor(ConsoleSpanExporter())], wrapper_methods=[ WrapperMethod( package="langchain.schema.runnable", object_name="RunnableParallel", method="invoke", span_name="langchain.workflow", wrapper=task_wrapper), WrapperMethod( package="langchain.schema.runnable", object_name="RunnableParallel", method="ainvoke", span_name="langchain.workflow", wrapper=atask_wrapper) ]) ``` -------------------------------- ### Monocle Manual Trace Start/Stop API Source: https://github.com/monocle2ai/monocle/blob/main/docs/monocle_trace_api.md Provides low-level control for manually starting and stopping trace spans. This is useful for advanced scenarios where direct control over the span lifecycle is required. ```APIDOC ## `start_trace` and `stop_trace` API ### Description Low-level API to manually start and stop a trace span. Useful for advanced scenarios or when you need to control the span lifecycle directly. ### Parameters #### `start_trace` Arguments - **span_name** (string) - Optional - Name for the span. Defaults to "custom_span". - **attributes** (dict) - Optional - Dict of custom attributes to set on the span initially. - **events** (list) - Optional - List of events to add to the span initially. #### `stop_trace` Arguments - **token** (object) - Required - The token returned by `start_trace`. - **final_attributes** (dict) - Optional - Dict of custom attributes to set on the span before it ends. - **error** (bool) - Optional - Indicates if an error occurred. ### Request Example ```python from monocle_apptrace.instrumentation.common.instrumentor import start_trace, stop_trace token = start_trace(span_name="manual_span", attributes={"foo": "bar"}) try: result = do_work() stop_trace(token, final_attributes={"result": result}) except Exception: stop_trace(token, final_attributes={"error": True}) raise ``` ``` -------------------------------- ### Langchain RAG Application with Monocle Tracing Source: https://github.com/monocle2ai/monocle/blob/main/apptrace/tests/langchain_starter.md A complete example of a Langchain RAG application instrumented with Monocle. This code loads web content, splits it, creates a vector store, and sets up a retrieval chain. Ensure you have the necessary OpenAI and ChromaDB configurations. ```python import bs4 from langchain import hub from langchain_community.document_loaders import WebBaseLoader from langchain_chroma import Chroma from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnablePassthrough from langchain_openai import OpenAIEmbeddings from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_openai import ChatOpenAI llm = ChatOpenAI(model="gpt-3.5-turbo-0125") # Load, chunk and index the contents of the blog. loader = WebBaseLoader( web_paths=("https://lilianweng.github.io/posts/2023-06-23-agent/",), bs_kwargs=dict( parse_only=bs4.SoupStrainer( class_=("post-content", "post-title", "post-header") ) ), ) docs = loader.load() text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200) splits = text_splitter.split_documents(docs) vectorstore = Chroma.from_documents(documents=splits, embedding=OpenAIEmbeddings()) # Retrieve and generate using the relevant snippets of the blog. retriever = vectorstore.as_retriever() prompt = hub.pull("rlm/rag-prompt") def format_docs(docs): return "\n\n".join(doc.page_content for doc in docs) rag_chain = ( {"context": retriever | format_docs, "question": RunnablePassthrough()} | prompt | llm | StrOutputParser() ) result = rag_chain.invoke("What is Task Decomposition?") print(result) ``` -------------------------------- ### Test Flight Booking Agent with Monocle Trace Asserter Source: https://context7.com/monocle2ai/monocle/llms.txt Use the `monocle_trace_asserter` fixture to run an agent asynchronously and assert that specific agents and tools were called or not called. This example demonstrates basic agent and tool invocation assertions. ```python import pytest from monocle_test_tools.pytest_plugin import monocle_trace_asserter @pytest.mark.asyncio async def test_flight_booking_agent(monocle_trace_asserter): # Run the agent and capture traces await monocle_trace_asserter.run_agent_async( root_agent, "google_adk", "Book a flight from San Jose to Seattle for November 27th, 2025" ) # Assert specific agent was called monocle_trace_asserter.called_agent("flight_booking_agent") # Assert specific tool was called monocle_trace_asserter.called_tool("search_flights") # Assert tool was NOT called monocle_trace_asserter.does_not_call_tool("cancel_booking") # Chain multiple assertions monocle_trace_asserter\ .called_agent("flight_booking_agent")\ .called_tool("search_flights")\ .called_tool("book_flight") ``` -------------------------------- ### Workflow Span Initialization Source: https://github.com/monocle2ai/monocle/blob/main/apptrace/src/monocle_apptrace/instrumentation/common/tracing.md Demonstrates setting up workflow telemetry and the resulting span hierarchy. ```python setup_monocle_telemetry(workflow='bot') rag_chain.invoke() --> context = retrieval() --> new_prompt = REST --> azure.func.chat(prompt) --> setup_monocle_telemetry(workflow='moderator') return llm(moderator_system_prompt+prompt) --> response = llm(new_prompt) ``` ```text Span{name='workflow.bot', type= workflow, traceID = xx1, spanID = yy0, parentID=None} ==> Workflow for new trace start Span{name='chain.invoke', type=anchor, traceID = xx1, spanID = yy1, parentID=yy0} ==> anchor span for chain invoke Span{name='chain.retrieval', type=retrieval, traceID = xx1, spanID = yy2, parentID = yy1} ==> Retrieval API span Span{name='workflow.moderator', type=workflow, traceID = xx1, spanID = zz1, parentID=yy1} ==> Workflow for propogated trace fragement Span{name='az.func.chat', type=anchor, traceID = xx1, spanID = zz2, parentID=zz1} ==> anchor span for az function invoke Span{name='chain.infer', type=inference, traceID = xx1, spanID = zz2, parentID=zz2} ==> inference Span{name='chain.infer',type=inference, traceID = xx1, spanID = yy3, parentID=yy1} ==> inference ``` -------------------------------- ### Manage Configuration Paths with Environment Utilities Source: https://github.com/monocle2ai/monocle/blob/main/apptrace/tests/MONOCLE_TESTING.md Compares manual environment variable management in a fixture against the cleaner temporary_env_var utility approach. ```python # From test_scopes.py - Manual approach (now you can use conftest utilities) @pytest.fixture(scope="function") def setup(): # Save original environment variable value original_scope_config = os.environ.get(SCOPE_CONFIG_PATH) try: os.environ[SCOPE_CONFIG_PATH] = "/path/to/test/config" # Setup code... yield custom_exporter finally: # Restore original environment variable value if original_scope_config is not None: os.environ[SCOPE_CONFIG_PATH] = original_scope_config elif SCOPE_CONFIG_PATH in os.environ: del os.environ[SCOPE_CONFIG_PATH] # Better approach using conftest utilities: def test_with_scope_config(setup): with temporary_env_var(SCOPE_CONFIG_PATH, "/path/to/test/config"): # Test code here pass ``` -------------------------------- ### Configure OTLP Exporter Programmatically Source: https://github.com/monocle2ai/monocle/blob/main/Monocle_User_Guide.md Initialize the OTLP exporter with custom settings and register it via setup_monocle_telemetry. ```python from monocle_apptrace.instrumentor import setup_monocle_telemetry from opentelemetry.sdk.trace.export import BatchSpanProcessor from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter # Configure OTLP exporter with custom settings otlp_exporter = OTLPSpanExporter( endpoint="http://localhost:4318/v1/traces", headers={"api-key": "your-api-key"}, timeout=15 # timeout in seconds ) setup_monocle_telemetry( workflow_name="my_genai_app", span_processors=[BatchSpanProcessor(otlp_exporter)] ) ``` -------------------------------- ### Run integration tests Source: https://github.com/monocle2ai/monocle/blob/main/Monocle_committer_guide.md Execute the full suite of integration tests. ```bash monocle% pytest -m integration ``` -------------------------------- ### Integrate with Grafana Tempo Source: https://github.com/monocle2ai/monocle/blob/main/Monocle_User_Guide.md Configure the OTLP exporter to send traces to a Tempo instance with authentication headers. ```bash # Configure Monocle to send traces to Grafana Tempo export MONOCLE_EXPORTER=otlp export OTEL_EXPORTER_OTLP_ENDPOINT=http://tempo:4318 export OTEL_EXPORTER_OTLP_HEADERS="Authorization=Basic your-base64-credentials" ``` -------------------------------- ### Integrate with OpenTelemetry Collector Source: https://github.com/monocle2ai/monocle/blob/main/Monocle_User_Guide.md Run a local OTLP collector and configure the application to export traces to it. ```bash # Run OpenTelemetry Collector locally docker run -p 4318:4318 \ -v $(pwd)/otel-collector-config.yaml:/etc/otel-collector-config.yaml \ otel/opentelemetry-collector:latest \ --config=/etc/otel-collector-config.yaml # Configure Monocle to use OTLP exporter export MONOCLE_EXPORTER=otlp export OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4318 # Run your application python your_app.py ``` -------------------------------- ### Run individual integration test Source: https://github.com/monocle2ai/monocle/blob/main/Monocle_committer_guide.md Execute a specific integration test file. ```bash monocle% pytest tests/integration/test_langchain_rag_l_to_m.py ``` -------------------------------- ### Invalid check_eval Example Source: https://github.com/monocle2ai/monocle/blob/main/docs/monocle_evaluation_api.md This example demonstrates an invalid usage of check_eval where a value is present in both 'expected' and 'not_expected' parameters, which will raise a ValueError. ```python # ERROR: "positive" appears in both expected and not_expected .check_eval("sentiment", expected=["positive", "neutral"], not_expected="positive") # ValueError! ``` -------------------------------- ### Run Pytest Tests via Command Line Source: https://github.com/monocle2ai/monocle/blob/main/docs/monocle_evaluation_api.md These commands demonstrate how to run pytest tests from the command line, including running all tests, verbose output, specific tests, and tests matching a keyword. ```bash pytest test_evals.py # Run all tests pytest test_evals.py -v # Verbose output pytest test_evals.py::test_name # Run specific test pytest -k "sentiment" # Run matching keyword ``` -------------------------------- ### Performance Failure Error Messages Source: https://github.com/monocle2ai/monocle/blob/main/docs/monocle_evaluation_api.md Examples of error messages generated when performance assertions fail. ```text AssertionError: Token limit exceeded: 1623 > 1500 ``` ```text AssertionError: Duration limit exceeded: workflow took 13.45 seconds (limit: 12.5 seconds) ``` ```text AssertionError: Duration limit exceeded for agent 'flight_booking_agent': agent_invocation took 0.23 minutes (limit: 0.2 minutes) ``` ```text AssertionError: Duration limit exceeded for tool 'book_flight': tool_invocation took 5.5 ms (limit: 5 ms) ``` -------------------------------- ### Configuration Methods Source: https://github.com/monocle2ai/monocle/blob/main/docs/monocle_test_assertions.md Methods to configure evaluation and comparison strategies. ```APIDOC ## Configuration Methods ### Description Methods to configure evaluation and comparison strategies. ### Method `with_evaluation(eval, eval_options=None)` ### Description Configures the evaluation method for input/output comparisons (e.g., LLM-based). ### Method `with_comparer(comparer)` ### Description Sets the comparison strategy for input/output matching (e.g., exact, token-based). ``` -------------------------------- ### Build Python package Source: https://github.com/monocle2ai/monocle/blob/main/Monocle_committer_guide.md Use the build module to create distribution packages. ```bash > python3 -m build ``` -------------------------------- ### Simple Syntax Check Eval Source: https://github.com/monocle2ai/monocle/blob/main/docs/monocle_evaluation_api.md Demonstrates the simple syntax for `check_eval` where the metric name and expected value are provided. ```python .check_eval("metric_name", "expected_value") .check_eval("metric_name", expected="value") .check_eval("metric_name", not_expected="value") ``` -------------------------------- ### Evaluate Inference Spans with not_expected Source: https://github.com/monocle2ai/monocle/blob/main/docs/monocle_evaluation_api.md This example evaluates the sentiment of each inference span, ensuring it is not negative. It uses the `not_expected` parameter for assertion. ```python await monocle_trace_asserter.run_agent_async(root_agent, "google_adk", "query") # Evaluate sentiment on each inference span - must NOT be negative monocle_trace_asserter.with_evaluation("okahu") .check_eval(fact_name="inferences", eval_name="sentiment", not_expected="negative") ``` -------------------------------- ### Run unit tests Source: https://github.com/monocle2ai/monocle/blob/main/Monocle_committer_guide.md Execute all unit tests located in the tests/unit directory. ```bash monocle% pytest tests/unit/*_test.py ``` -------------------------------- ### Test with Multiple Temporary Environment Variables Source: https://github.com/monocle2ai/monocle/blob/main/apptrace/tests/MONOCLE_TESTING.md Illustrates using the `temporary_env_vars` context manager to set multiple environment variables for a test. This is useful for scenarios requiring specific configurations for API keys, models, and debug flags. ```python from config.conftest import temporary_env_vars def test_with_multiple_env_vars(setup): """Test with multiple temporary environment variables""" with temporary_env_vars( OPENAI_API_KEY="test_key", OPENAI_MODEL="gpt-3.5-turbo", MONOCLE_DEBUG="true" ): # Test code with temporary environment pass # All original values restored automatically ``` -------------------------------- ### Decorate Methods with monocle_trace_method Source: https://github.com/monocle2ai/monocle/blob/main/docs/monocle_trace_api.md Apply this decorator to synchronous or asynchronous functions to automatically start and stop a trace for the function's execution. Spans within the function share the same trace ID. ```python from monocle_apptrace.instrumentation.common.instrumentor import monocle_trace_method @monocle_trace_method(span_name="process_request") def process_request(data): ... ``` ```python @monocle_trace_method(span_name="process_job") async def process_job(data): ... ``` -------------------------------- ### Preserve Environment State with preserve_env Source: https://github.com/monocle2ai/monocle/blob/main/apptrace/tests/MONOCLE_TESTING.md Use the preserve_env fixture to automatically back up and restore the entire os.environ state during a test. ```python def test_with_env_preservation(preserve_env, setup): """Test that modifies environment but ensures cleanup""" # This test can modify os.environ directly without worrying about cleanup os.environ["SOME_CONFIG"] = "test_value" os.environ["ANOTHER_CONFIG"] = "another_value" os.environ["THIRD_CONFIG"] = "third_value" # Test code that relies on these environment variables # No manual cleanup needed - preserve_env fixture handles complete restoration ``` ```python # Example: Testing configuration loading with multiple environment variables def test_config_loading_with_multiple_vars(preserve_env, setup): """Test configuration loading with various environment combinations""" # Set up a complex environment scenario os.environ["MONOCLE_DEBUG"] = "true" os.environ["MONOCLE_EXPORTERS"] = "console,file" os.environ["OPENAI_API_KEY"] = "test-key-123" os.environ["AZURE_OPENAI_ENDPOINT"] = "https://test.openai.azure.com" os.environ["SCOPE_CONFIG_PATH"] = "/tmp/test-scope-config.json" # Test code that uses these environment variables config = load_monocle_configuration() assert config.debug_enabled is True assert "console" in config.exporters # Environment automatically restored to original state after test ``` -------------------------------- ### Use monocle_trace for Synchronous Tracing Source: https://github.com/monocle2ai/monocle/blob/main/docs/monocle_trace_api.md Use this context manager in synchronous code to start a new trace (span) for all code within its block. It accepts optional span name and attributes. ```python from monocle_apptrace.instrumentation.common.instrumentor import monocle_trace with monocle_trace(span_name="my_operation", attributes={"user.id": "user-123"}): result = my_function() ``` -------------------------------- ### Example Monocle Trace Output Source: https://github.com/monocle2ai/monocle/blob/main/apptrace/tests/llama_starter.md This JSON represents the trace data captured by Monocle for a LlamaIndex workflow. It includes span names, timestamps, durations, and attributes like workflow input and output. ```json [ { "span_name": "llamaindex.retrieve", "start_time": "2024-04-15T23:27:54.806773Z", "end_time": "2024-04-15T23:27:55.732604Z", "duration_ms": "926", "span_id": "0x030cf03872d4a092", "trace_id": "0xbd54e5d0edcd96634fa8a02c25c27519", "parent_id": "0xb4b14a8f14e7e770", "attributes": {}, "events": [] }, { "span_name": "llamaindex.openai", "start_time": "2024-04-15T23:27:55.740299Z", "end_time": "2024-04-15T23:27:57.181992Z", "duration_ms": "1442", "span_id": "0x225fbfb58481e58c", "trace_id": "0xbd54e5d0edcd96634fa8a02c25c27519", "parent_id": "0xb4b14a8f14e7e770", "attributes": {}, "events": [] }, { "span_name": "llamaindex.query", "start_time": "2024-04-15T23:27:54.806477Z", "end_time": "2024-04-15T23:27:57.182261Z", "duration_ms": "2376", "span_id": "0xb4b14a8f14e7e770", "trace_id": "0xbd54e5d0edcd96634fa8a02c25c27519", "parent_id": "None", "attributes": { "workflow_input": "What did the author do growing up?", "workflow_name": "llama_index_1", "workflow_output": "The context does not provide information about what the author did while growing up.", "workflow_type": "workflow.llamaindex" }, "events": [] } ] ``` -------------------------------- ### Configure OTLP Exporter Environment Variables Source: https://github.com/monocle2ai/monocle/blob/main/Monocle_User_Guide.md Define OTLP-specific settings such as endpoints, headers, and timeouts using standard OpenTelemetry environment variables. ```bash # Set the exporter to OTLP export MONOCLE_EXPORTER=otlp # Configure the OTLP endpoint (default: http://localhost:4318) export OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4318 # Optional: Set specific traces endpoint export OTEL_EXPORTER_OTLP_TRACES_ENDPOINT=http://localhost:4318/v1/traces # Optional: Configure authentication headers export OTEL_EXPORTER_OTLP_HEADERS="api-key=your-api-key" # Optional: Configure timeout (in milliseconds, default: 10000) export OTEL_EXPORTER_OTLP_TIMEOUT=15000 ``` -------------------------------- ### Monocle Trace Context Manager (Synchronous) Source: https://github.com/monocle2ai/monocle/blob/main/docs/monocle_trace_api.md Use `monocle_trace` as a context manager to automatically start and stop a trace span for synchronous code blocks. It allows for custom span names, attributes, and events. ```APIDOC ## `monocle_trace` Context Manager ### Description A context manager for starting a new trace (span) for all code executed within its block. Use in synchronous code. ### Parameters #### Context Manager Arguments - **span_name** (string) - Optional - Name for the span. Defaults to "custom_span". - **attributes** (dict) - Optional - Dict of custom attributes to set on the span. - **events** (list) - Optional - List of events to add to the span. ### Request Example ```python from monocle_apptrace.instrumentation.common.instrumentor import monocle_trace with monocle_trace(span_name="my_operation", attributes={"user.id": "user-123"}): result = my_function() ``` ``` -------------------------------- ### Configure OTLP Exporter Source: https://context7.com/monocle2ai/monocle/llms.txt Sets up an OTLP span exporter with custom endpoint, headers, and timeout settings. ```python otlp_exporter = OTLPSpanExporter( endpoint="http://localhost:4318/v1/traces", headers={"api-key": "your-api-key"}, timeout=15 # timeout in seconds ) setup_monocle_telemetry( workflow_name="my_genai_app", span_processors=[BatchSpanProcessor(otlp_exporter)] ) ``` -------------------------------- ### Setup Monocle Telemetry with Custom Exporter Fixture Source: https://github.com/monocle2ai/monocle/blob/main/apptrace/tests/MONOCLE_TESTING.md Use this pytest fixture to set up Monocle telemetry with a custom exporter for capturing spans. Ensure `instrumentor.uninstrument()` is called in the `finally` block to clean up global state. ```python import pytest from common.custom_exporter import CustomConsoleSpanExporter from monocle_apptrace import setup_monocle_telemetry from opentelemetry.sdk.trace.export import SimpleSpanProcessor @pytest.fixture(scope="function") def setup(): custom_exporter = CustomConsoleSpanExporter() instrumentor = None try: # Setup Monocle telemetry with custom exporter for span capture instrumentor = setup_monocle_telemetry( workflow_name="test_app", span_processors=[SimpleSpanProcessor(custom_exporter)], wrapper_methods=[] ) # Yield the exporter so tests can access captured spans yield custom_exporter finally: # CRITICAL: Clean up instrumentor to avoid global state leakage if instrumentor and instrumentor.is_instrumented_by_opentelemetry: instrumentor.uninstrument() def test_my_functionality(setup): custom_exporter = setup # Your test code here # ... execute traced operations ... # Access captured spans for validation captured_spans = custom_exporter.get_captured_spans() assert len(captured_spans) > 0 ``` -------------------------------- ### Test with Temporary Environment Variable Source: https://github.com/monocle2ai/monocle/blob/main/apptrace/tests/MONOCLE_TESTING.md Demonstrates how to use the `temporary_env_var` context manager within a test function to isolate the test's environment. The `OPENAI_API_KEY` is temporarily set and automatically restored. ```python from config.conftest import temporary_env_var def test_with_invalid_api_key(setup): """Test error handling with invalid API key""" with temporary_env_var("OPENAI_API_KEY", "INVALID_KEY"): # Test code that should handle invalid API key # Environment variable is automatically restored after the block pass # Original OPENAI_API_KEY value is restored here ``` -------------------------------- ### Manually Control Traces with start_trace and stop_trace Source: https://github.com/monocle2ai/monocle/blob/main/docs/monocle_trace_api.md Use these low-level functions for manual trace span control. `start_trace` returns a token to be used with `stop_trace`. Useful for advanced scenarios or custom span lifecycle management. ```python from monocle_apptrace.instrumentation.common.instrumentor import start_trace, stop_trace token = start_trace(span_name="manual_span", attributes={"foo": "bar"}) try: result = do_work() stop_trace(token, final_attributes={"result": result}) except Exception: stop_trace(token, final_attributes={"error": True}) raise ```