validate_environment() — langchain Function Reference

Architecture documentation for the validate_environment() function in base.py from the langchain codebase.

Function python LangChainCore LanguageModelBase

Entity Profile

LangChainCore→ LanguageModelBase→ validate_environment() — langchain Function Reference

Dependency Diagram

graph TD
  b7a95e1a_8067_79e6_ae42_61d2648ee8ad["validate_environment()"]
  2a683305_667b_3567_cab9_9f77e29d4afa["BaseChatOpenAI"]
  b7a95e1a_8067_79e6_ae42_61d2648ee8ad -->|defined in| 2a683305_667b_3567_cab9_9f77e29d4afa
  style b7a95e1a_8067_79e6_ae42_61d2648ee8ad fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/partners/openai/langchain_openai/chat_models/base.py lines 904–1026

    def validate_environment(self) -> Self:
        """Validate that api key and python package exists in environment."""
        if self.n is not None and self.n < 1:
            msg = "n must be at least 1."
            raise ValueError(msg)
        if self.n is not None and self.n > 1 and self.streaming:
            msg = "n must be 1 when streaming."
            raise ValueError(msg)

        # Check OPENAI_ORGANIZATION for backwards compatibility.
        self.openai_organization = (
            self.openai_organization
            or os.getenv("OPENAI_ORG_ID")
            or os.getenv("OPENAI_ORGANIZATION")
        )
        self.openai_api_base = self.openai_api_base or os.getenv("OPENAI_API_BASE")

        # Enable stream_usage by default if using default base URL and client
        if (
            all(
                getattr(self, key, None) is None
                for key in (
                    "stream_usage",
                    "openai_proxy",
                    "openai_api_base",
                    "base_url",
                    "client",
                    "root_client",
                    "async_client",
                    "root_async_client",
                    "http_client",
                    "http_async_client",
                )
            )
            and "OPENAI_BASE_URL" not in os.environ
        ):
            self.stream_usage = True

        # Resolve API key from SecretStr or Callable
        sync_api_key_value: str | Callable[[], str] | None = None
        async_api_key_value: str | Callable[[], Awaitable[str]] | None = None

        if self.openai_api_key is not None:
            # Because OpenAI and AsyncOpenAI clients support either sync or async
            # callables for the API key, we need to resolve separate values here.
            sync_api_key_value, async_api_key_value = _resolve_sync_and_async_api_keys(
                self.openai_api_key
            )

        client_params: dict = {
            "organization": self.openai_organization,
            "base_url": self.openai_api_base,
            "timeout": self.request_timeout,
            "default_headers": self.default_headers,
            "default_query": self.default_query,
        }
        if self.max_retries is not None:
            client_params["max_retries"] = self.max_retries

        if self.openai_proxy and (self.http_client or self.http_async_client):
            openai_proxy = self.openai_proxy
            http_client = self.http_client
            http_async_client = self.http_async_client
            msg = (
                "Cannot specify 'openai_proxy' if one of "
                "'http_client'/'http_async_client' is already specified. Received:\n"
                f"{openai_proxy=}\n{http_client=}\n{http_async_client=}"
            )
            raise ValueError(msg)
        if not self.client:
            if sync_api_key_value is None:
                # No valid sync API key, leave client as None and raise informative
                # error on invocation.
                self.client = None
                self.root_client = None
            else:
                if self.openai_proxy and not self.http_client:
                    try:
                        import httpx
                    except ImportError as e:
                        msg = (