Source code for langchain.callbacks.tracers.langchain_v1
from __future__ import annotations
import logging
import os
from typing import Any, Dict, Optional, Union
import requests
from langchain.callbacks.tracers.base import BaseTracer
from langchain.callbacks.tracers.schemas import (
    ChainRun,
    LLMRun,
    Run,
    ToolRun,
    TracerSession,
    TracerSessionV1,
    TracerSessionV1Base,
)
from langchain.schema.messages import get_buffer_string
from langchain.utils import raise_for_status_with_text
logger = logging.getLogger(__name__)
[docs]def get_headers() -> Dict[str, Any]:
    """Get the headers for the LangChain API."""
    headers: Dict[str, Any] = {"Content-Type": "application/json"}
    if os.getenv("LANGCHAIN_API_KEY"):
        headers["x-api-key"] = os.getenv("LANGCHAIN_API_KEY")
    return headers
def _get_endpoint() -> str:
    return os.getenv("LANGCHAIN_ENDPOINT", "http://localhost:8000")
[docs]class LangChainTracerV1(BaseTracer):
    """An implementation of the SharedTracer that POSTS to the langchain endpoint."""
[docs]    def __init__(self, **kwargs: Any) -> None:
        """Initialize the LangChain tracer."""
        super().__init__(**kwargs)
        self.session: Optional[TracerSessionV1] = None
        self._endpoint = _get_endpoint()
        self._headers = get_headers()
    def _convert_to_v1_run(self, run: Run) -> Union[LLMRun, ChainRun, ToolRun]:
        session = self.session or self.load_default_session()
        if not isinstance(session, TracerSessionV1):