To log traces, you must instrument your application either manually or automatically. To log to a remote instance of Phoenix, you must also configure the host and port where your traces will be sent.
When running running Phoenix locally on the default port of 6006, no additional configuration is necessary.
import phoenix as pxfrom phoenix.trace import LangChainInstrumentorpx.launch_app()LangChainInstrumentor().instrument()# run your LangChain application
If you are running a remote instance of Phoenix, you can configure your instrumentation to log to that instance using the PHOENIX_HOST and PHOENIX_PORT environment variables.
import osfrom phoenix.trace import LangChainInstrumentor# assume phoenix is running at 162.159.135.42:6007os.environ["PHOENIX_HOST"]="162.159.135.42"os.environ["PHOENIX_PORT"]="6007"LangChainInstrumentor().instrument()# logs to http://162.159.135.42:6007# run your LangChain application
Alternatively, you can use the PHOENIX_COLLECTOR_ENDPOINT environment variable.
import osfrom phoenix.trace import LangChainInstrumentor# assume phoenix is running at 162.159.135.42:6007os.environ["PHOENIX_COLLECTOR_ENDPOINT"]="162.159.135.42:6007"LangChainInstrumentor().instrument()# logs to http://162.159.135.42:6007# run your LangChain application
How to turn off tracing
Tracing can be paused temporarily or disabled permanently.
Pause tracing using context manager
If there is a section of your code for which tracing is not desired, e.g. the document chunking process, it can be put inside the suppress_tracing context manager as shown below.
from phoenix.trace import suppress_tracingwithsuppress_tracing():# Code running inside this block doesn't generate traces.# For example, running LLM evals here won't generate additional traces. ...# Tracing will resume outside the block....
Uninstrument the auto-instrumentors permanently
Calling .uninstrument() on the auto-instrumentors will remove tracing permanently. Below is the examples for LangChain, LlamaIndex and OpenAI, respectively.
LangChainInstrumentor().uninstrument()LlamaIndexInstrumentor().uninstrument()OpenAIInstrumentor().uninstrument()# etc.
For OpenAI, how do I get token counts when streaming?
To get token counts when streaming, install openai>=1.26 and set stream_options={"include_usage": True} when calling create. Below is an example Python code snippet. For more info, see here.
If you have customized a LangChain component (say a retriever), you might not get tracing for that component without some additional steps. Internally, instrumentation relies on components to inherit from LangChain base classes for the traces to show up. Below is an example of how to inherit from LanChain base classes to make a custom retriever and to make traces show up.
from typing import Listfrom langchain_core.callbacks import CallbackManagerForRetrieverRunfrom langchain_core.retrievers import BaseRetriever, Documentfrom openinference.instrumentation.langchain import LangChainInstrumentorfrom opentelemetry import trace as trace_apifrom opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporterfrom opentelemetry.sdk import trace as trace_sdkfrom opentelemetry.sdk.trace.export import SimpleSpanProcessorPHOENIX_COLLECTOR_ENDPOINT ="http://127.0.0.1:6006/v1/traces"tracer_provider = trace_sdk.TracerProvider()trace_api.set_tracer_provider(tracer_provider)tracer_provider.add_span_processor(SimpleSpanProcessor(OTLPSpanExporter(endpoint)))LangChainInstrumentor().instrument()classCustomRetriever(BaseRetriever):""" This example is taken from langchain docs. https://python.langchain.com/v0.1/docs/modules/data_connection/retrievers/custom_retriever/ A custom retriever that contains the top k documents that contain the user query. This retriever only implements the sync method _get_relevant_documents. If the retriever were to involve file access or network access, it could benefit from a native async implementation of `_aget_relevant_documents`. As usual, with Runnables, there's a default async implementation that's provided that delegates to the sync implementation running on another thread. """ k:int"""Number of top results to return"""def_get_relevant_documents(self,query:str,*, run_manager: CallbackManagerForRetrieverRun ) -> List[Document]:"""Sync implementations for retriever.""" matching_documents: List[Document]= []# Custom logic to find the top k documents that contain the queryfor index inrange(self.k): matching_documents.append(Document(page_content=f"dummy content at {index}", score=1.0))return matching_documentsretriever =CustomRetriever(k=3)if__name__=="__main__": documents = retriever.invoke("what is the meaning of life?")