LangServe
#!/usr/bin/env python
"""Example LangChain server exposes a conversational retrieval chain."""
from fastapi import FastAPI
from langchain.agents import AgentExecutor, tool
from langchain.agents.format_scratchpad import format_to_openai_functions
from langchain.agents.output_parsers import OpenAIFunctionsAgentOutputParser
from langchain.chat_models import ChatOpenAI
from langchain.embeddings import OpenAIEmbeddings
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.tools.render import format_tool_to_openai_function
from langchain.vectorstores import FAISS
from pydantic import BaseModel

from langserve import add_routes

vectorstore = FAISS.from_texts(
    ["cats like fish", "dogs like sticks"], embedding=OpenAIEmbeddings()
)
retriever = vectorstore.as_retriever()


@tool
def get_eugene_thoughts(query: str) -> list:
    """Returns Eugene's thoughts on a topic."""
    return retriever.get_relevant_documents(query)


tools = [get_eugene_thoughts]

prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a helpful assistant."),
        ("user", "{input}"),
        MessagesPlaceholder(variable_name="agent_scratchpad"),
    ]
)

llm = ChatOpenAI()

llm_with_tools = llm.bind(functions=[format_tool_to_openai_function(t) for t in tools])

agent = (
    {
        "input": lambda x: x["input"],
        "agent_scratchpad": lambda x: format_to_openai_functions(
            x["intermediate_steps"]
        ),
    }
    | prompt
    | llm_with_tools
    | OpenAIFunctionsAgentOutputParser()
)

agent_executor = AgentExecutor(agent=agent, tools=tools)

app = FastAPI(
    title="LangChain Server",
    version="1.0",
    description="Spin up a simple api server using Langchain's Runnable interfaces",
)


# We need to add these input/output schemas because the current AgentExecutor
# is lacking in schemas.
class Input(BaseModel):
    input: str


class Output(BaseModel):
    output: str


# Adds routes to the app for using the chain under:
# /invoke
# /batch
# /stream
add_routes(app, agent_executor, input_type=Input, output_type=Output)

if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="localhost", port=8000)

examples/agent/server.py
langchain-ai/langserve
#!/usr/bin/env python
"""Example LangChain server exposes a chain composed of a prompt and an LLM."""
from fastapi import FastAPI
from langchain.chat_models import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from typing_extensions import TypedDict

from langserve import add_routes

model = ChatOpenAI()
prompt = ChatPromptTemplate.from_template("tell me a joke about {topic}")
chain = prompt | model

app = FastAPI(
    title="LangChain Server",
    version="1.0",
    description="Spin up a simple api server using Langchain's Runnable interfaces",
)


# The input type is automatically inferred from the runnable
# interface; however, if you want to override it, you can do so
# by passing in the input_type argument to add_routes.
class ChainInput(TypedDict):
    """The input to the chain."""

    topic: str
    """The topic of the joke."""


add_routes(app, chain, input_type=ChainInput)

# Alternatively, you can rely on langchain's type inference
# to infer the input type from the runnable interface.
# add_routes(app, chain)

if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="localhost", port=8000)

examples/chain/server.py
langchain-ai/langserve
#!/usr/bin/env python
"""Example LangChain server exposes a conversational retrieval chain."""
from fastapi import FastAPI
from langchain.chains import ConversationalRetrievalChain
from langchain.chat_models import ChatOpenAI
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import FAISS

from langserve import add_routes

vectorstore = FAISS.from_texts(
    ["cats like fish", "dogs like sticks"], embedding=OpenAIEmbeddings()
)
retriever = vectorstore.as_retriever()

model = ChatOpenAI()

chain = ConversationalRetrievalChain.from_llm(model, retriever)

app = FastAPI(
    title="LangChain Server",
    version="1.0",
    description="Spin up a simple api server using Langchain's Runnable interfaces",
)
# Adds routes to the app for using the chain under:
# /invoke
# /batch
# /stream
add_routes(app, chain)

if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="localhost", port=8000)

examples/conversational_retrieval_chain/server.py
langchain-ai/langserve
#!/usr/bin/env python
"""Example LangChain server exposes multiple runnables (LLMs in this case)."""

from fastapi import FastAPI
from langchain.chat_models import ChatAnthropic, ChatOpenAI

from langserve import add_routes

app = FastAPI(
    title="LangChain Server",
    version="1.0",
    description="Spin up a simple api server using Langchain's Runnable interfaces",
)

add_routes(
    app,
    ChatOpenAI(),
    path="/openai",
)
add_routes(
    app,
    ChatAnthropic(),
    path="/anthropic",
)

if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="localhost", port=8000)

examples/llm/server.py
langchain-ai/langserve
#!/usr/bin/env python
"""Example LangChain server exposes a retriever."""
from fastapi import FastAPI
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import FAISS

from langserve import add_routes

vectorstore = FAISS.from_texts(
    ["cats like fish", "dogs like sticks"], embedding=OpenAIEmbeddings()
)
retriever = vectorstore.as_retriever()

app = FastAPI(
    title="LangChain Server",
    version="1.0",
    description="Spin up a simple api server using Langchain's Runnable interfaces",
)
# Adds routes to the app for using the retriever under:
# /invoke
# /batch
# /stream
add_routes(app, retriever)

if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="localhost", port=8000)

examples/retrieval/server.py
langchain-ai/langserve
"""Main entrypoint into package."""

from langserve.client import RemoteRunnable
from langserve.server import add_routes
from langserve.version import __version__

__all__ = ["RemoteRunnable", "add_routes", "__version__"]

langserve/__init__.py
langchain-ai/langserve
from __future__ import annotations

import asyncio
import weakref
from concurrent.futures import ThreadPoolExecutor
from typing import (
    Any,
    AsyncIterator,
    Iterator,
    List,
    Optional,
    Sequence,
    Union,
)
from urllib.parse import urljoin

import httpx
from langchain.callbacks.tracers.log_stream import RunLog, RunLogPatch
from langchain.load.dump import dumpd
from langchain.schema.runnable import Runnable
from langchain.schema.runnable.config import (
    RunnableConfig,
    ensure_config,
    get_async_callback_manager_for_config,
    get_callback_manager_for_config,
)
from langchain.schema.runnable.utils import Input, Output

from langserve.serialization import simple_dumpd, simple_loads


def _without_callbacks(config: Optional[RunnableConfig]) -> RunnableConfig:
    """Evict callbacks from the config since those are definitely not supported."""
    _config = config or {}
    return {k: v for k, v in _config.items() if k != "callbacks"}


def _raise_for_status(response: httpx.Response) -> None:
    """Re-raise with a more informative message.

    Args:
        response: The response to check

    Raises:
        httpx.HTTPStatusError: If the response is not 2xx, appending the response
                               text to the message
    """
    try:
        response.raise_for_status()
    except httpx.HTTPStatusError as e:
        message = str(e)
        # Append the response text if it exists, as it may contain more information
        # Especially useful when the user's request is malformed
        if e.response.text:
            message += f" for {e.response.text}"

        raise httpx.HTTPStatusError(
            message=message,
            request=e.request,
            response=e.response,
        )


def _is_async() -> bool:
    """Return True if we are in an async context."""
    try:
        asyncio.get_running_loop()
    except RuntimeError:
        return False
    else:
        return True


def _close_clients(sync_client: httpx.Client, async_client: httpx.AsyncClient) -> None:
    """Close the async and sync clients.

    _close_clients should not be a bound method since it is called by a weakref
    finalizer.

    Args:
        sync_client: The sync client to close
        async_client: The async client to close
    """
    sync_client.close()
    if _is_async():
        # Use a ThreadPoolExecutor to run async_client_close in a separate thread
        with ThreadPoolExecutor(max_workers=1) as executor:
            # Submit the async_client_close coroutine to the thread pool
            future = executor.submit(asyncio.run, async_client.aclose())
            future.result()
    else:
        asyncio.run(async_client.aclose())


class RemoteRunnable(Runnable[Input, Output]):
    """A RemoteRunnable is a runnable that is executed on a remote server.

    This client implements the majority of the runnable interface.

    The following features are not supported:

    - `batch` with `return_exceptions=True` since we do not support exception
      translation from the server.
    - Callbacks via the `config` argument as serialization of callbacks is not
      supported.
    """

    def __init__(
        self,
        url: str,
        *,
        timeout: Optional[float] = None,
    ) -> None:
        """Initialize the client.

        Args:
            url: The url of the server
            timeout: The timeout for requests
        """
        self.url = url
        self.sync_client = httpx.Client(base_url=url, timeout=timeout)
        self.async_client = httpx.AsyncClient(base_url=url, timeout=timeout)

        # Register cleanup handler once RemoteRunnable is garbage collected
        weakref.finalize(self, _close_clients, self.sync_client, self.async_client)

    def _invoke(
        self, input: Input, config: Optional[RunnableConfig] = None, **kwargs: Any
    ) -> Output:
        """Invoke the runnable with the given input and config."""
        response = self.sync_client.post(
            "/invoke",
            json={
                "input": simple_dumpd(input),
                "config": _without_callbacks(config),
                "kwargs": kwargs,
            },
        )
        _raise_for_status(response)
        return simple_loads(response.text)["output"]

    def invoke(
        self, input: Input, config: Optional[RunnableConfig] = None, **kwargs: Any
    ) -> Output:
        if kwargs:
            raise NotImplementedError("kwargs not implemented yet.")
        return self._call_with_config(self._invoke, input, config=config)

    async def _ainvoke(
        self, input: Input, config: Optional[RunnableConfig] = None, **kwargs: Any
    ) -> Output:
        response = await self.async_client.post(
            "/invoke",
            json={
                "input": simple_dumpd(input),
                "config": _without_callbacks(config),
                "kwargs": kwargs,
            },
        )
        _raise_for_status(response)
        return simple_loads(response.text)["output"]

    async def ainvoke(
        self, input: Input, config: Optional[RunnableConfig] = None, **kwargs: Any
    ) -> Output:
        if kwargs:
            raise NotImplementedError("kwargs not implemented yet.")
        return await self._acall_with_config(self._ainvoke, input, config)

    def _batch(
        self,
        inputs: List[Input],
        config: Optional[Union[RunnableConfig, List[RunnableConfig]]] = None,
        *,
        return_exceptions: bool = False,
        **kwargs: Optional[Any],
    ) -> List[Output]:
        if not inputs:
            return []
        if return_exceptions:
            raise NotImplementedError(
                "return_exceptions is not supported for remote clients"
            )

        if isinstance(config, list):
            _config = [_without_callbacks(c) for c in config]
        else:
            _config = _without_callbacks(config)

        response = self.sync_client.post(
            "/batch",
            json={
                "inputs": simple_dumpd(inputs),
                "config": _config,
                "kwargs": kwargs,
            },
        )
        _raise_for_status(response)
        return simple_loads(response.text)["output"]

    def batch(
        self,
        inputs: List[Input],
        config: Optional[RunnableConfig] = None,
        **kwargs: Any,
    ) -> List[Output]:
        if kwargs:
            raise NotImplementedError("kwargs not implemented yet.")
        return self._batch_with_config(self._batch, inputs, config)

    async def _abatch(
        self,
        inputs: List[Input],
        config: Optional[Union[RunnableConfig, List[RunnableConfig]]] = None,
        *,
        return_exceptions: bool = False,
        **kwargs: Optional[Any],
    ) -> List[Output]:
        """Batch invoke the runnable."""
        if not inputs:
            return []
        if return_exceptions:
            raise NotImplementedError(
                "return_exceptions is not supported for remote clients"
            )

        if isinstance(config, list):
            _config = [_without_callbacks(c) for c in config]
        else:
            _config = _without_callbacks(config)

        response = await self.async_client.post(
            "/batch",
            json={
                "inputs": simple_dumpd(inputs),
                "config": _config,
                "kwargs": kwargs,
            },
        )
        _raise_for_status(response)
        return simple_loads(response.text)["output"]

    async def abatch(
        self,
        inputs: List[Input],
        config: Optional[RunnableConfig] = None,
        *,
        return_exceptions: bool = False,
        **kwargs: Any,
    ) -> List[Output]:
        """Batch invoke the runnable."""
        if kwargs:
            raise NotImplementedError("kwargs not implemented yet.")
        if not inputs:
            return []
        return await self._abatch_with_config(self._abatch, inputs, config)

    def stream(
        self,
        input: Input,
        config: Optional[RunnableConfig] = None,
        **kwargs: Optional[Any],
    ) -> Iterator[Output]:
        """Stream invoke the runnable."""
        config = ensure_config(config)
        callback_manager = get_callback_manager_for_config(config)

        final_output: Optional[Output] = None

        run_manager = callback_manager.on_chain_start(
            dumpd(self),
            simple_dumpd(input),
            name=config.get("run_name"),
        )
        data = {
            "input": simple_dumpd(input),
            "config": _without_callbacks(config),
            "kwargs": kwargs,
        }
        endpoint = urljoin(self.url, "stream")

        try:
            from httpx_sse import connect_sse
        except ImportError:
            raise ImportError(
                "Missing `httpx_sse` dependency to use the stream method. "
                "Install via `pip install httpx_sse`'"
            )

        try:
            with connect_sse(
                self.sync_client, "POST", endpoint, json=data
            ) as event_source:
                for sse in event_source.iter_sse():
                    if sse.event == "data":
                        chunk = simple_loads(sse.data)
                        yield chunk

                        if final_output:
                            final_output += chunk
                        else:
                            final_output = chunk
                    elif sse.event == "end":
                        break
                    else:
                        raise NotImplementedError(f"Unknown event {sse.event}")
        except BaseException as e:
            run_manager.on_chain_error(e)
            raise
        else:
            run_manager.on_chain_end(final_output)

    async def astream(
        self,
        input: Input,
        config: Optional[RunnableConfig] = None,
        **kwargs: Optional[Any],
    ) -> AsyncIterator[Output]:
        config = ensure_config(config)
        callback_manager = get_async_callback_manager_for_config(config)

        final_output: Optional[Output] = None

        run_manager = await callback_manager.on_chain_start(
            dumpd(self),
            simple_dumpd(input),
            name=config.get("run_name"),
        )
        data = {
            "input": simple_dumpd(input),
            "config": _without_callbacks(config),
            "kwargs": kwargs,
        }
        endpoint = urljoin(self.url, "stream")

        try:
            from httpx_sse import aconnect_sse
        except ImportError:
            raise ImportError("You must install `httpx_sse` to use the stream method.")

        try:
            async with aconnect_sse(
                self.async_client, "POST", endpoint, json=data
            ) as event_source:
                async for sse in event_source.aiter_sse():
                    if sse.event == "data":
                        chunk = simple_loads(sse.data)
                        yield chunk

                        if final_output:
                            final_output += chunk
                        else:
                            final_output = chunk
                    elif sse.event == "end":
                        break

                    else:
                        raise NotImplementedError(f"Unknown event {sse.event}")
        except BaseException as e:
            await run_manager.on_chain_error(e)
            raise
        else:
            await run_manager.on_chain_end(final_output)

    async def astream_log(
        self,
        input: Input,
        config: Optional[RunnableConfig] = None,
        *,
        diff: bool = False,
        include_names: Optional[Sequence[str]] = None,
        include_types: Optional[Sequence[str]] = None,
        include_tags: Optional[Sequence[str]] = None,
        exclude_names: Optional[Sequence[str]] = None,
        exclude_types: Optional[Sequence[str]] = None,
        exclude_tags: Optional[Sequence[str]] = None,
        **kwargs: Optional[Any],
    ) -> Union[AsyncIterator[RunLogPatch], AsyncIterator[RunLog]]:
        """Stream all output from a runnable, as reported to the callback system.

        This includes all inner runs of LLMs, Retrievers, Tools, etc.

        Output is streamed as Log objects, which include a list of
        jsonpatch ops that describe how the state of the run has changed in each
        step, and the final state of the run.

        The jsonpatch ops can be applied in order to construct state.
        """

        # Create a stream handler that will emit Log objects
        config = ensure_config(config)
        callback_manager = get_async_callback_manager_for_config(config)

        final_output: Optional[Output] = None

        run_manager = await callback_manager.on_chain_start(
            dumpd(self),
            simple_dumpd(input),
            name=config.get("run_name"),
        )
        data = {
            "input": simple_dumpd(input),
            "config": _without_callbacks(config),
            "kwargs": kwargs,
            "diff": diff,
            "include_names": include_names,
            "include_types": include_types,
            "include_tags": include_tags,
            "exclude_names": exclude_names,
            "exclude_types": exclude_types,
            "exclude_tags": exclude_tags,
        }
        endpoint = urljoin(self.url, "stream_log")

        try:
            from httpx_sse import aconnect_sse
        except ImportError:
            raise ImportError("You must install `httpx_sse` to use the stream method.")

        try:
            async with aconnect_sse(
                self.async_client, "POST", endpoint, json=data
            ) as event_source:
                async for sse in event_source.aiter_sse():
                    if sse.event == "data":
                        data = simple_loads(sse.data)
                        if diff:
                            chunk = RunLogPatch(*data["ops"])
                        else:
                            chunk = RunLog(*data["ops"], state=data["state"])

                        yield chunk

                        if diff:
                            if final_output:
                                final_output += chunk
                            else:
                                final_output = chunk
                        else:
                            final_output = chunk
                    elif sse.event == "end":
                        break
                    else:
                        raise NotImplementedError(f"Unknown event {sse.event}")
        except BaseException as e:
            await run_manager.on_chain_error(e)
            raise
        else:
            await run_manager.on_chain_end(final_output)

langserve/client.py
langchain-ai/langserve
"""Serialization module for Well Known LangChain objects.

Specialized JSON serialization for well known LangChain objects that
can be expected to be frequently transmitted between chains.
"""
import json
from typing import Any, Union

from langchain.prompts.base import StringPromptValue
from langchain.prompts.chat import ChatPromptValueConcrete
from langchain.schema.agent import AgentAction, AgentActionMessageLog, AgentFinish
from langchain.schema.document import Document
from langchain.schema.messages import (
    AIMessage,
    AIMessageChunk,
    ChatMessage,
    ChatMessageChunk,
    FunctionMessage,
    FunctionMessageChunk,
    HumanMessage,
    HumanMessageChunk,
    SystemMessage,
    SystemMessageChunk,
)

try:
    from pydantic.v1 import BaseModel, ValidationError
except ImportError:
    from pydantic import BaseModel, ValidationError


class WellKnownLCObject(BaseModel):
    """A well known LangChain object.

    A pydantic model that defines what constitutes a well known LangChain object.

    All well-known objects are allowed to be serialized and de-serialized.
    """

    __root__: Union[
        Document,
        HumanMessage,
        SystemMessage,
        ChatMessage,
        FunctionMessage,
        AIMessage,
        HumanMessageChunk,
        SystemMessageChunk,
        ChatMessageChunk,
        FunctionMessageChunk,
        AIMessageChunk,
        StringPromptValue,
        ChatPromptValueConcrete,
        AgentAction,
        AgentFinish,
        AgentActionMessageLog,
    ]


# Custom JSON Encoder
class _LangChainEncoder(json.JSONEncoder):
    """Custom JSON Encoder that can encode pydantic objects as well."""

    def default(self, obj) -> Any:
        if isinstance(obj, BaseModel):
            return obj.dict()
        return super().default(obj)


# Custom JSON Decoder
class _LangChainDecoder(json.JSONDecoder):
    """Custom JSON Decoder that handles well known LangChain objects."""

    def __init__(self, *args: Any, **kwargs: Any) -> None:
        """Initialize the LangChainDecoder."""
        super().__init__(object_hook=self.decoder, *args, **kwargs)

    def decoder(self, value) -> Any:
        """Decode the value."""
        if isinstance(value, dict):
            try:
                obj = WellKnownLCObject.parse_obj(value)
                return obj.__root__
            except ValidationError:
                return {key: self.decoder(v) for key, v in value.items()}
        elif isinstance(value, list):
            return [self.decoder(item) for item in value]
        else:
            return value


# PUBLIC API


def simple_dumpd(obj: Any) -> Any:
    """Convert the given object to a JSON serializable object."""
    return json.loads(json.dumps(obj, cls=_LangChainEncoder))


def simple_dumps(obj: Any) -> str:
    """Dump the given object as a JSON string."""
    return json.dumps(obj, cls=_LangChainEncoder)


def simple_loads(s: str) -> Any:
    """Load the given JSON string."""
    return json.loads(s, cls=_LangChainDecoder)

langserve/serialization.py
langchain-ai/langserve
"""FastAPI integration for langchain runnables.

This code contains integration for langchain runnables with FastAPI.

The main entry point is the `add_routes` function which adds the routes to an existing
FastAPI app or APIRouter.
"""
from inspect import isclass
from typing import (
    Any,
    AsyncIterator,
    Dict,
    Literal,
    Mapping,
    Sequence,
    Type,
    Union,
)

from fastapi import Request
from langchain.callbacks.tracers.log_stream import RunLog, RunLogPatch
from langchain.load.serializable import Serializable
from langchain.schema.runnable import Runnable
from typing_extensions import Annotated

from langserve.version import __version__

try:
    from pydantic.v1 import BaseModel, create_model
except ImportError:
    from pydantic import BaseModel, Field, create_model

from langserve.serialization import simple_dumpd, simple_dumps
from langserve.validation import (
    create_batch_request_model,
    create_batch_response_model,
    create_invoke_request_model,
    create_invoke_response_model,
    create_stream_log_request_model,
    create_stream_request_model,
)

try:
    from fastapi import APIRouter, FastAPI
except ImportError:
    # [server] extra not installed
    APIRouter = FastAPI = Any


def _unpack_config(d: Union[BaseModel, Mapping], keys: Sequence[str]) -> Dict[str, Any]:
    """Project the given keys from the given dict."""
    _d = d.dict() if isinstance(d, BaseModel) else d
    return {k: _d[k] for k in keys if k in _d}


def _unpack_input(validated_model: BaseModel) -> Any:
    """Unpack the decoded input from the validated model."""
    if hasattr(validated_model, "__root__"):
        model = validated_model.__root__
    else:
        model = validated_model

    if isinstance(model, BaseModel) and not isinstance(model, Serializable):
        # If the model is a pydantic model, but not a Serializable, then
        # it was created by the server as part of validation and isn't expected
        # to be accepted by the runnables as input as a pydantic model,
        # instead we need to convert it into a corresponding python dict.
        return model.dict()

    return model


# This is a global registry of models to avoid creating the same model
# multiple times.
# Duplicated model names break fastapi's openapi generation.
_MODEL_REGISTRY = {}


def _resolve_model(type_: Union[Type, BaseModel], default_name: str) -> Type[BaseModel]:
    """Resolve the input type to a BaseModel."""
    if isclass(type_) and issubclass(type_, BaseModel):
        model = type_
    else:
        model = create_model(default_name, __root__=(type_, ...))

    hash_ = model.schema_json()

    if hash_ not in _MODEL_REGISTRY:
        _MODEL_REGISTRY[hash_] = model

    return _MODEL_REGISTRY[hash_]


def _add_namespace_to_model(namespace: str, model: Type[BaseModel]) -> Type[BaseModel]:
    """Prefix the name of the given model with the given namespace.

    Code is used to help avoid name collisions when hosting multiple runnables
    that may use the same underlying models.

    Args:
        namespace: The namespace to use for the model.
        model: The model to create a unique name for.

    Returns:
        A new model with name prepended with the given namespace.
    """

    class Config:
        arbitrary_types_allowed = True

    model_with_unique_name = create_model(
        f"{namespace}{model.__name__}",
        config=Config,
        **{
            name: (
                field.annotation,
                Field(
                    field.default,
                    title=name,
                    description=field.field_info.description,
                ),
            )
            for name, field in model.__fields__.items()
        },
    )
    model_with_unique_name.update_forward_refs()
    return model_with_unique_name


def _add_tracing_info_to_metadata(config: Dict[str, Any], request: Request) -> None:
    """Add information useful for tracing and debugging purposes.

    Args:
        config: The config to expand with tracing information.
        request: The request to use for expanding the metadata.
    """

    metadata = config["metadata"] if "metadata" in config else {}

    info = {
        "__useragent": request.headers.get("user-agent"),
        "__langserve_version": __version__,
    }
    metadata.update(info)
    config["metadata"] = metadata


# PUBLIC API


def add_routes(
    app: Union[FastAPI, APIRouter],
    runnable: Runnable,
    *,
    path: str = "",
    input_type: Union[Type, Literal["auto"], BaseModel] = "auto",
    output_type: Union[Type, Literal["auto"], BaseModel] = "auto",
    config_keys: Sequence[str] = (),
) -> None:
    """Register the routes on the given FastAPI app or APIRouter.


    The following routes are added per runnable under the specified `path`:

    * /invoke - for invoking a runnable with a single input
    * /batch - for invoking a runnable with multiple inputs
    * /stream - for streaming the output of a runnable
    * /stream_log - for streaming intermediate outputs for a runnable
    * /input_schema - for returning the input schema of the runnable
    * /output_schema - for returning the output schema of the runnable
    * /config_schema - for returning the config schema of the runnable

    Args:
        app: The FastAPI app or APIRouter to which routes should be added.
        runnable: The runnable to wrap, must not be stateful.
        path: A path to prepend to all routes.
        input_type: type to use for input validation.
            Default is "auto" which will use the InputType of the runnable.
            User is free to provide a custom type annotation.
        output_type: type to use for output validation.
            Default is "auto" which will use the OutputType of the runnable.
            User is free to provide a custom type annotation.
        config_keys: list of config keys that will be accepted, by default
                     no config keys are accepted.
    """
    try:
        from sse_starlette import EventSourceResponse
    except ImportError:
        raise ImportError(
            "sse_starlette must be installed to implement the stream and "
            "stream_log endpoints. "
            "Use `pip install sse_starlette` to install."
        )

    input_type_ = _resolve_model(
        runnable.input_schema if input_type == "auto" else input_type, "Input"
    )

    output_type_ = _resolve_model(
        runnable.output_schema if output_type == "auto" else output_type, "Output"
    )

    namespace = path or ""

    model_namespace = path.strip("/").replace("/", "_")

    config = _add_namespace_to_model(
        model_namespace, runnable.config_schema(include=config_keys)
    )
    InvokeRequest = create_invoke_request_model(model_namespace, input_type_, config)
    BatchRequest = create_batch_request_model(model_namespace, input_type_, config)
    StreamRequest = create_stream_request_model(model_namespace, input_type_, config)
    StreamLogRequest = create_stream_log_request_model(
        model_namespace, input_type_, config
    )
    # Generate the response models
    InvokeResponse = create_invoke_response_model(model_namespace, output_type_)
    BatchResponse = create_batch_response_model(model_namespace, output_type_)

    @app.post(
        f"{namespace}/invoke",
        response_model=InvokeResponse,
    )
    async def invoke(
        invoke_request: Annotated[InvokeRequest, InvokeRequest],
        request: Request,
    ) -> InvokeResponse:
        """Invoke the runnable with the given input and config."""
        # Request is first validated using InvokeRequest which takes into account
        # config_keys as well as input_type.
        config = _unpack_config(invoke_request.config, config_keys)
        _add_tracing_info_to_metadata(config, request)
        output = await runnable.ainvoke(
            _unpack_input(invoke_request.input), config=config
        )

        return InvokeResponse(output=simple_dumpd(output))

    #
    @app.post(f"{namespace}/batch", response_model=BatchResponse)
    async def batch(
        batch_request: Annotated[BatchRequest, BatchRequest],
        request: Request,
    ) -> BatchResponse:
        """Invoke the runnable with the given inputs and config."""
        if isinstance(batch_request.config, list):
            config = [
                _unpack_config(config, config_keys) for config in batch_request.config
            ]

            for c in config:
                _add_tracing_info_to_metadata(c, request)
        else:
            config = _unpack_config(batch_request.config, config_keys)
            _add_tracing_info_to_metadata(config, request)
        inputs = [_unpack_input(input_) for input_ in batch_request.inputs]
        output = await runnable.abatch(inputs, config=config)

        return BatchResponse(output=simple_dumpd(output))

    @app.post(f"{namespace}/stream")
    async def stream(
        stream_request: Annotated[StreamRequest, StreamRequest],
        request: Request,
    ) -> EventSourceResponse:
        """Invoke the runnable stream the output.

        This endpoint allows to stream the output of the runnable.

        The endpoint uses a server sent event stream to stream the output.

        https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events

        Important: Set the "text/event-stream" media type for request headers if
                   not using an existing SDK.

        This endpoint uses two different types of events:

        * data - for streaming the output of the runnable

            {
                "event": "data",
                "data": {
                ...
                }
            }

        * end - for signaling the end of the stream.

            This helps the client to know when to stop listening for events and
            know that the streaming has ended successfully.

            {
                "event": "end",
            }
        """
        # Request is first validated using InvokeRequest which takes into account
        # config_keys as well as input_type.
        # After validation, the input is loaded using LangChain's load function.
        input_ = _unpack_input(stream_request.input)
        config = _unpack_config(stream_request.config, config_keys)
        _add_tracing_info_to_metadata(config, request)

        async def _stream() -> AsyncIterator[dict]:
            """Stream the output of the runnable."""
            async for chunk in runnable.astream(
                input_,
                config=config,
            ):
                yield {"data": simple_dumps(chunk), "event": "data"}
            yield {"event": "end"}

        return EventSourceResponse(_stream())

    @app.post(f"{namespace}/stream_log")
    async def stream_log(
        stream_log_request: Annotated[StreamLogRequest, StreamLogRequest],
        request: Request,
    ) -> EventSourceResponse:
        """Invoke the runnable stream_log the output.

        This endpoint allows to stream the output of the runnable, including
        the output of all intermediate steps.

        The endpoint uses a server sent event stream to stream the output.

        https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events

        Important: Set the "text/event-stream" media type for request headers if
                   not using an existing SDK.

        This endpoint uses two different types of events:

        * data - for streaming the output of the runnable

            {
                "event": "data",
                "data": {
                ...
                }
            }

        * end - for signaling the end of the stream.

            This helps the client to know when to stop listening for events and
            know that the streaming has ended successfully.

            {
                "event": "end",
            }
        """
        # Request is first validated using InvokeRequest which takes into account
        # config_keys as well as input_type.
        # After validation, the input is loaded using LangChain's load function.
        input_ = _unpack_input(stream_log_request.input)
        config = _unpack_config(stream_log_request.config, config_keys)
        _add_tracing_info_to_metadata(config, request)

        async def _stream_log() -> AsyncIterator[dict]:
            """Stream the output of the runnable."""
            async for chunk in runnable.astream_log(
                input_,
                config=config,
                diff=stream_log_request.diff,
                include_names=stream_log_request.include_names,
                include_types=stream_log_request.include_types,
                include_tags=stream_log_request.include_tags,
                exclude_names=stream_log_request.exclude_names,
                exclude_types=stream_log_request.exclude_types,
                exclude_tags=stream_log_request.exclude_tags,
            ):
                if stream_log_request.diff:  # Run log patch
                    if not isinstance(chunk, RunLogPatch):
                        raise AssertionError(
                            f"Expected a RunLog instance got {type(chunk)}"
                        )
                    data = {
                        "ops": chunk.ops,
                    }
                else:
                    # Then it's a run log
                    if not isinstance(chunk, RunLog):
                        raise AssertionError(
                            f"Expected a RunLog instance got {type(chunk)}"
                        )
                    data = {
                        "state": chunk.state,
                        "ops": chunk.ops,
                    }

                # Temporary adapter
                yield {
                    "data": simple_dumps(data),
                    "event": "data",
                }
            yield {"event": "end"}

        return EventSourceResponse(_stream_log())

    @app.get(f"{namespace}/input_schema")
    async def input_schema() -> Any:
        """Return the input schema of the runnable."""
        return runnable.input_schema.schema()

    @app.get(f"{namespace}/output_schema")
    async def output_schema() -> Any:
        """Return the output schema of the runnable."""
        return runnable.output_schema.schema()

    @app.get(f"{namespace}/config_schema")
    async def config_schema() -> Any:
        """Return the config schema of the runnable."""
        return runnable.config_schema(include=config_keys).schema()

langserve/server.py
langchain-ai/langserve
"""Code to dynamically create pydantic models for validating requests and responses.

Requests share the same basic shape of input, config, and kwargs.

Invoke and Batch responses use an `output` key for the output, other keys may
be added to the response at a later date.

Responses for stream and stream_log are specified as those endpoints use
a streaming response.

Type information for input, config and output can be specified by the user
per runnable. This type information will be used for validation of the input and
output and will appear in the OpenAPI spec for the corresponding endpoint.

Models are created with a namespace to avoid name collisions when hosting
multiple runnables. When present the name collisions prevent fastapi from
generating OpenAPI specs.
"""
from typing import List, Optional, Sequence, Union

try:
    from pydantic.v1 import BaseModel, Field, create_model
except ImportError:
    from pydantic import BaseModel, Field, create_model

from typing_extensions import Type, TypedDict

# Type that is either a python annotation or a pydantic model that can be
# used to validate the input or output of a runnable.
Validator = Union[Type[BaseModel], type]

# PUBLIC API


def create_invoke_request_model(
    namespace: str,
    input_type: Validator,
    config: TypedDict,
) -> Type[BaseModel]:
    """Create a pydantic model for the invoke request."""
    invoke_request_type = create_model(
        f"{namespace}InvokeRequest",
        input=(input_type, Field(..., description="The input to the runnable.")),
        config=(
            config,
            Field(
                default_factory=dict,
                description=(
                    "Subset of RunnableConfig object in LangChain. "
                    "Useful for passing information like tags, metadata etc."
                ),
            ),
        ),
        kwargs=(
            dict,
            Field(
                default_factory=dict,
                description="Keyword arguments to the runnable. Currently ignored.",
            ),
        ),
    )
    invoke_request_type.update_forward_refs()
    return invoke_request_type


def create_stream_request_model(
    namespace: str,
    input_type: Validator,
    config: TypedDict,
) -> Type[BaseModel]:
    """Create a pydantic model for the stream request."""
    stream_request_model = create_model(
        f"{namespace}StreamRequest",
        input=(input_type, Field(..., description="The input to the runnable.")),
        config=(
            config,
            Field(
                default_factory=dict,
                description=(
                    "Subset of RunnableConfig object in LangChain. "
                    "Useful for passing information like tags, metadata etc."
                ),
            ),
        ),
        kwargs=(
            dict,
            Field(
                default_factory=dict,
                description="Keyword arguments to the runnable. Currently ignored.",
            ),
        ),
    )
    stream_request_model.update_forward_refs()
    return stream_request_model


def create_batch_request_model(
    namespace: str,
    input_type: Validator,
    config: TypedDict,
) -> Type[BaseModel]:
    """Create a pydantic model for the batch request."""
    batch_request_type = create_model(
        f"{namespace}BatchRequest",
        inputs=(List[input_type], ...),
        config=(
            Union[config, List[config]],
            Field(
                default_factory=dict,
                description=(
                    "Subset of RunnableConfig object in LangChain. Either specify one "
                    "config for all inputs or a list of configs with one per input. "
                    "Useful for passing information like tags, metadata etc."
                ),
            ),
        ),
        kwargs=(
            dict,
            Field(
                default_factory=dict,
                description="Keyword arguments to the runnable. Currently ignored.",
            ),
        ),
    )
    batch_request_type.update_forward_refs()
    return batch_request_type


def create_stream_log_request_model(
    namespace: str,
    input_type: Validator,
    config: TypedDict,
) -> Type[BaseModel]:
    """Create a pydantic model for the invoke request."""
    stream_log_request = create_model(
        f"{namespace}StreamLogRequest",
        input=(input_type, ...),
        config=(config, Field(default_factory=dict)),
        diff=(Optional[bool], False),
        include_names=(
            Optional[Sequence[str]],
            Field(
                None,
                description="If specified, filter to runnables with matching names",
            ),
        ),
        include_types=(
            Optional[Sequence[str]],
            Field(
                None,
                description="If specified, filter to runnables with matching types",
            ),
        ),
        include_tags=(
            Optional[Sequence[str]],
            Field(
                None,
                description="If specified, filter to runnables with matching tags",
            ),
        ),
        exclude_names=(
            Optional[Sequence[str]],
            Field(
                None,
                description="If specified, exclude runnables with matching names",
            ),
        ),
        exclude_types=(
            Optional[Sequence[str]],
            Field(
                None,
                description="If specified, exclude runnables with matching types",
            ),
        ),
        exclude_tags=(
            Optional[Sequence[str]],
            Field(
                None,
                description="If specified, exclude runnables with matching tags",
            ),
        ),
        kwargs=(dict, Field(default_factory=dict)),
    )
    stream_log_request.update_forward_refs()
    return stream_log_request


def create_invoke_response_model(
    namespace: str,
    output_type: Validator,
) -> Type[BaseModel]:
    """Create a pydantic model for the invoke response."""
    # The invoke response uses a key called `output` for the output, so
    # other information can be added to the response at a later date.
    invoke_response_type = create_model(
        f"{namespace}InvokeResponse",
        output=(output_type, Field(..., description="The output of the invocation.")),
    )
    invoke_response_type.update_forward_refs()
    return invoke_response_type


def create_batch_response_model(
    namespace: str,
    output_type: Validator,
) -> Type[BaseModel]:
    """Create a pydantic model for the batch response."""
    # The response uses a key called `output` for the output, so
    # other information can be added to the response at a later date.
    batch_response_type = create_model(
        f"{namespace}BatchResponse",
        output=(
            List[output_type],
            Field(
                ...,
                description=(
                    "The outputs corresponding to the inputs the batch request."
                ),
            ),
        ),
    )
    batch_response_type.update_forward_refs()
    return batch_response_type

langserve/validation.py
langchain-ai/langserve
"""Main entrypoint into package."""
from importlib import metadata

try:
    __version__ = metadata.version(__package__)
except metadata.PackageNotFoundError:
    # Case where package metadata is not available.
    __version__ = ""
del metadata  # optional, avoids polluting the results of dir(__package__)

langserve/version.py
langchain-ai/langserve
from typing import Any

import pytest
from langchain.schema.messages import (
    HumanMessage,
    HumanMessageChunk,
    SystemMessage,
)

try:
    from pydantic.v1 import BaseModel
except ImportError:
    from pydantic import BaseModel

from langserve.serialization import simple_dumps, simple_loads


@pytest.mark.parametrize(
    "data",
    [
        # Test with python primitives
        1,
        [],
        {},
        {"a": 1},
        {"output": [HumanMessage(content="hello")]},
        # Test with a single message (HumanMessage)
        HumanMessage(content="Hello"),
        # Test with a list containing mixed elements
        [HumanMessage(content="Hello"), SystemMessage(content="Hi"), 42, "world"],
        # Uncomment when langchain 0.0.306 is released
        [HumanMessage(content="Hello"), HumanMessageChunk(content="Hi")],
        # Attention: This test is not correct right now
        # Test with full and chunk messages
        [HumanMessageChunk(content="Hello"), HumanMessage(content="Hi")],
        # Test with a dictionary containing mixed elements
        {
            "message": HumanMessage(content="Greetings"),
            "numbers": [1, 2, 3],
            "boom": "Hello, world!",
        },
    ],
)
def test_serialization(data: Any) -> None:
    """There and back again! :)"""
    # Test encoding
    assert isinstance(simple_dumps(data), str)
    # Test simple equality (does not include pydantic class names)
    assert simple_loads(simple_dumps(data)) == data
    # Test full representation equality (includes pydantic class names)
    assert _get_full_representation(
        simple_loads(simple_dumps(data))
    ) == _get_full_representation(data)


def _get_full_representation(data: Any) -> Any:
    """Get the full representation of the data, replacing pydantic models with schema.

    Pydantic tests two different models for equality based on equality
    of their schema; instead we will rely on the equality of their full
    schema representation. This will make sure that both models have the
    same name (e.g., HumanMessage vs. HumanMessageChunk).

    Args:
        data: python primitives + pydantic models

    Returns:
        data represented entirely with python primitives
    """
    if isinstance(data, dict):
        return {key: _get_full_representation(value) for key, value in data.items()}
    elif isinstance(data, list):
        return [_get_full_representation(value) for value in data]
    elif isinstance(data, BaseModel):
        return data.schema()
    else:
        return data

tests/unit_tests/test_serialization.py
langchain-ai/langserve
"""Test the server and client together."""
import asyncio
from asyncio import AbstractEventLoop
from contextlib import asynccontextmanager
from typing import List, Optional, Union

import httpx
import pytest
import pytest_asyncio
from fastapi import FastAPI
from fastapi.testclient import TestClient
from httpx import AsyncClient
from langchain.callbacks.tracers.log_stream import RunLog, RunLogPatch
from langchain.prompts import PromptTemplate
from langchain.schema.messages import HumanMessage, SystemMessage
from langchain.schema.runnable import RunnablePassthrough
from langchain.schema.runnable.base import RunnableLambda
from langchain.schema.runnable.utils import ConfigurableField
from pytest_mock import MockerFixture

from langserve.client import RemoteRunnable
from langserve.server import add_routes
from tests.unit_tests.utils import FakeListLLM


@pytest.fixture(scope="session")
def event_loop():
    """Create an instance of the default event loop for each test case."""
    loop = asyncio.get_event_loop()
    try:
        yield loop
    finally:
        loop.close()


@pytest.fixture()
def app(event_loop: AbstractEventLoop) -> FastAPI:
    """A simple server that wraps a Runnable and exposes it as an API."""

    async def add_one_or_passthrough(
        x: Union[int, HumanMessage]
    ) -> Union[int, HumanMessage]:
        """Add one to int or passthrough."""
        if isinstance(x, int):
            return x + 1
        else:
            return x

    runnable_lambda = RunnableLambda(func=add_one_or_passthrough)
    app = FastAPI()
    try:
        add_routes(app, runnable_lambda)
        yield app
    finally:
        del app


@pytest.fixture()
def client(app: FastAPI) -> RemoteRunnable:
    """Create a FastAPI app that exposes the Runnable as an API."""
    remote_runnable_client = RemoteRunnable(url="http://localhost:9999")
    sync_client = TestClient(app=app)
    remote_runnable_client.sync_client = sync_client
    yield remote_runnable_client
    sync_client.close()


@asynccontextmanager
async def get_async_client(
    server: FastAPI, path: Optional[str] = None
) -> RemoteRunnable:
    """Get an async client."""
    url = "http://localhost:9999"
    if path:
        url += path
    remote_runnable_client = RemoteRunnable(url=url)
    async_client = AsyncClient(app=server, base_url=url)
    remote_runnable_client.async_client = async_client
    try:
        yield remote_runnable_client
    finally:
        await async_client.aclose()


@pytest_asyncio.fixture()
async def async_client(app: FastAPI) -> RemoteRunnable:
    """Create a FastAPI app that exposes the Runnable as an API."""
    async with get_async_client(app) as client:
        yield client


def test_server(app: FastAPI) -> None:
    """Test the server directly via HTTP requests."""
    sync_client = TestClient(app=app)

    # Test invoke
    response = sync_client.post("/invoke", json={"input": 1})
    assert response.json() == {"output": 2}

    # Test batch
    response = sync_client.post("/batch", json={"inputs": [1]})
    assert response.json() == {
        "output": [2],
    }

    # Test schema
    input_schema = sync_client.get("/input_schema").json()
    assert isinstance(input_schema, dict)
    assert input_schema["title"] == "RunnableLambdaInput"

    output_schema = sync_client.get("/output_schema").json()
    assert isinstance(output_schema, dict)
    assert output_schema["title"] == "RunnableLambdaOutput"

    output_schema = sync_client.get("/config_schema").json()
    assert isinstance(output_schema, dict)
    assert output_schema["title"] == "RunnableLambdaConfig"

    # TODO(Team): Fix test. Issue with eventloops right now when using sync client
    ## Test stream
    # response = sync_client.post("/stream", json={"input": 1})
    # assert response.text == "event: data\r\ndata: 2\r\n\r\nevent: end\r\n\r\n"


@pytest.mark.asyncio
async def test_server_async(app: FastAPI) -> None:
    """Test the server directly via HTTP requests."""
    async_client = AsyncClient(app=app, base_url="http://localhost:9999")

    # Test invoke
    response = await async_client.post("/invoke", json={"input": 1})
    assert response.json() == {"output": 2}

    # Test batch
    response = await async_client.post("/batch", json={"inputs": [1]})
    assert response.json() == {
        "output": [2],
    }

    # Test stream
    response = await async_client.post("/stream", json={"input": 1})
    assert response.text == "event: data\r\ndata: 2\r\n\r\nevent: end\r\n\r\n"


def test_invoke(client: RemoteRunnable) -> None:
    """Test sync invoke."""
    assert client.invoke(1) == 2
    assert client.invoke(HumanMessage(content="hello")) == HumanMessage(content="hello")
    # Test invocation with config
    assert client.invoke(1, config={"tags": ["test"]}) == 2


def test_batch(client: RemoteRunnable) -> None:
    """Test sync batch."""
    assert client.batch([]) == []
    assert client.batch([1, 2, 3]) == [2, 3, 4]
    assert client.batch([HumanMessage(content="hello")]) == [
        HumanMessage(content="hello")
    ]


@pytest.mark.asyncio
async def test_ainvoke(async_client: RemoteRunnable) -> None:
    """Test async invoke."""
    assert await async_client.ainvoke(1) == 2
    assert await async_client.ainvoke(HumanMessage(content="hello")) == HumanMessage(
        content="hello"
    )


@pytest.mark.asyncio
async def test_abatch(async_client: RemoteRunnable) -> None:
    """Test async batch."""
    assert await async_client.abatch([]) == []
    assert await async_client.abatch([1, 2, 3]) == [2, 3, 4]
    assert await async_client.abatch([HumanMessage(content="hello")]) == [
        HumanMessage(content="hello")
    ]


# TODO(Team): Determine how to test
# Some issue with event loops
# def test_stream(client: RemoteRunnable) -> None:
#     """Test stream."""
#     assert list(client.stream(1)) == [2]


@pytest.mark.asyncio
async def test_astream(async_client: RemoteRunnable) -> None:
    """Test async stream."""
    outputs = []

    async for chunk in async_client.astream(1):
        outputs.append(chunk)

    assert outputs == [2]

    outputs = []
    data = HumanMessage(content="hello")

    async for chunk in async_client.astream(data):
        outputs.append(chunk)

    assert outputs == [data]


@pytest.mark.asyncio
async def test_astream_log_no_diff(async_client: RemoteRunnable) -> None:
    """Test async stream."""
    run_logs = []

    async for chunk in async_client.astream_log(1, diff=False):
        run_logs.append(chunk)

    assert len(run_logs) == 3

    op = run_logs[0].ops[0]
    uuid = op["value"]["id"]

    for run_log in run_logs:
        assert isinstance(run_log, RunLog)

    states = [run_log.state for run_log in run_logs]

    assert states == [
        {
            "final_output": None,
            "id": uuid,
            "logs": {},
            "streamed_output": [],
        },
        {
            "final_output": {"output": 2},
            "id": uuid,
            "logs": {},
            "streamed_output": [],
        },
        {
            "final_output": {"output": 2},
            "id": uuid,
            "logs": {},
            "streamed_output": [2],
        },
    ]

    # Check that we're picking up one extra op on each chunk
    assert [len(run_log.ops) for run_log in run_logs] == [1, 2, 3]


@pytest.mark.asyncio
async def test_astream_log(async_client: RemoteRunnable) -> None:
    """Test async stream."""
    run_log_patches = []

    async for chunk in async_client.astream_log(1, diff=True):
        run_log_patches.append(chunk)

    op = run_log_patches[0].ops[0]
    uuid = op["value"]["id"]

    assert [run_log_patch.ops for run_log_patch in run_log_patches] == [
        [
            {
                "op": "replace",
                "path": "",
                "value": {
                    "final_output": {"output": 2},
                    "id": uuid,
                    "logs": {},
                    "streamed_output": [],
                },
            }
        ],
        [{"op": "replace", "path": "/final_output", "value": {"output": 2}}],
        [{"op": "add", "path": "/streamed_output/-", "value": 2}],
    ]


def test_invoke_as_part_of_sequence(client: RemoteRunnable) -> None:
    """Test as part of sequence."""
    runnable = client | RunnableLambda(func=lambda x: x + 1)
    # without config
    assert runnable.invoke(1) == 3
    # with config
    assert runnable.invoke(1, config={"tags": ["test"]}) == 3
    # without config
    assert runnable.batch([1, 2]) == [3, 4]
    # with config
    assert runnable.batch([1, 2], config={"tags": ["test"]}) == [3, 4]
    # TODO(Team): Determine how to test some issues with event loops for testing
    #   set up
    # without config
    # assert list(runnable.stream([1, 2])) == [3, 4]
    # # with config
    # assert list(runnable.stream([1, 2], config={"tags": ["test"]})) == [3, 4]


@pytest.mark.asyncio
async def test_invoke_as_part_of_sequence_async(async_client: RemoteRunnable) -> None:
    """Test as part of a sequence.

    This helps to verify that config is handled properly (e.g., callbacks are not
    passed to the server, but other config is)
    """
    runnable = async_client | RunnableLambda(
        func=lambda x: x + 1 if isinstance(x, int) else x
    ).with_config({"run_name": "hello"})
    # without config
    assert await runnable.ainvoke(1) == 3
    # with config
    assert await runnable.ainvoke(1, config={"tags": ["test"]}) == 3
    # without config
    assert await runnable.abatch([1, 2]) == [3, 4]
    # with config
    assert await runnable.abatch([1, 2], config={"tags": ["test"]}) == [3, 4]

    # Verify can pass many configs to batch
    configs = [{"tags": ["test"]}, {"tags": ["test2"]}]
    assert await runnable.abatch([1, 2], config=configs) == [3, 4]

    # Verify can ValueError on mismatched configs  number
    with pytest.raises(ValueError):
        assert await runnable.abatch([1, 2], config=[configs[0]]) == [3, 4]

    configs = [{"tags": ["test"]}, {"tags": ["test2"]}]
    assert await runnable.abatch([1, 2], config=configs) == [3, 4]

    configs = [
        {"tags": ["test"]},
        {"tags": ["test2"], "other": "test"},
    ]
    assert await runnable.abatch([1, 2], config=configs) == [3, 4]

    # Without config
    assert [x async for x in runnable.astream(1)] == [3]

    # With Config
    assert [x async for x in runnable.astream(1, config={"tags": ["test"]})] == [3]

    # With config and LC input data
    assert [
        x
        async for x in runnable.astream(
            HumanMessage(content="hello"), config={"tags": ["test"]}
        )
    ] == [HumanMessage(content="hello")]

    log_patches = [x async for x in runnable.astream_log(1)]
    for log_patch in log_patches:
        assert isinstance(log_patch, RunLogPatch)
    # Only check the first entry (not validating implementation here)
    first_op = log_patches[0].ops[0]
    assert first_op["op"] == "replace"
    assert first_op["path"] == ""

    # Validate with HumanMessage
    log_patches = [x async for x in runnable.astream_log(HumanMessage(content="hello"))]
    for log_patch in log_patches:
        assert isinstance(log_patch, RunLogPatch)
    # Only check the first entry (not validating implementation here)
    first_op = log_patches[0].ops[0]
    assert first_op == {
        "op": "replace",
        "path": "",
        "value": {
            "final_output": None,
            "id": first_op["value"]["id"],
            "logs": {},
            "streamed_output": [],
        },
    }


@pytest.mark.asyncio
async def test_multiple_runnables(event_loop: AbstractEventLoop) -> None:
    """Test serving multiple runnables."""

    async def add_one(x: int) -> int:
        """Add one to simulate a valid function"""
        return x + 1

    async def mul_2(x: int) -> int:
        """Add one to simulate a valid function"""
        return x * 2

    app = FastAPI()
    add_routes(app, RunnableLambda(add_one), path="/add_one")
    add_routes(
        app,
        RunnableLambda(mul_2),
        input_type=int,
        path="/mul_2",
    )

    async with get_async_client(app, path="/add_one") as runnable:
        async with get_async_client(app, path="/mul_2") as runnable2:
            assert await runnable.ainvoke(1) == 2
            assert await runnable2.ainvoke(4) == 8

            composite_runnable = runnable | runnable2
            assert await composite_runnable.ainvoke(3) == 8

            # Invoke runnable (remote add_one), local add_one, remote mul_2
            composite_runnable_2 = runnable | add_one | runnable2
            assert await composite_runnable_2.ainvoke(3) == 10


@pytest.mark.asyncio
async def test_input_validation(
    event_loop: AbstractEventLoop, mocker: MockerFixture
) -> None:
    """Test client side and server side exceptions."""

    async def add_one(x: int) -> int:
        """Add one to simulate a valid function"""
        return x + 1

    server_runnable = RunnableLambda(func=add_one)
    server_runnable2 = RunnableLambda(func=add_one)

    app = FastAPI()
    add_routes(
        app,
        server_runnable,
        input_type=int,
        path="/add_one",
    )

    add_routes(
        app,
        server_runnable2,
        input_type=int,
        path="/add_one_config",
        config_keys=["tags", "run_name", "metadata"],
    )

    async with get_async_client(app, path="/add_one") as runnable:
        # Verify that can be invoked with valid input
        assert await runnable.ainvoke(1) == 2
        # Verify that the following substring is present in the error message
        with pytest.raises(httpx.HTTPError):
            await runnable.ainvoke("hello")

        with pytest.raises(httpx.HTTPError):
            await runnable.abatch(["hello"])

    config = {"tags": ["test"], "metadata": {"a": 5}}

    invoke_spy_1 = mocker.spy(server_runnable, "ainvoke")
    # Verify config is handled correctly
    async with get_async_client(app, path="/add_one") as runnable1:
        # Verify that can be invoked with valid input
        # Config ignored for runnable1
        assert await runnable1.ainvoke(1, config=config) == 2
        # Config should be ignored but default debug information
        # will still be added
        config_seen = invoke_spy_1.call_args[1]["config"]
        assert "metadata" in config_seen
        assert "__useragent" in config_seen["metadata"]
        assert "__langserve_version" in config_seen["metadata"]

    invoke_spy_2 = mocker.spy(server_runnable2, "ainvoke")
    async with get_async_client(app, path="/add_one_config") as runnable2:
        # Config accepted for runnable2
        assert await runnable2.ainvoke(1, config=config) == 2
        # Config ignored

        config_seen = invoke_spy_2.call_args[1]["config"]
        assert config_seen["tags"] == ["test"]
        assert config_seen["metadata"]["a"] == 5
        assert "__useragent" in config_seen["metadata"]
        assert "__langserve_version" in config_seen["metadata"]


@pytest.mark.asyncio
async def test_input_validation_with_lc_types(event_loop: AbstractEventLoop) -> None:
    """Test client side and server side exceptions."""

    app = FastAPI()
    # Test with langchain objects
    add_routes(
        app, RunnablePassthrough(), input_type=List[HumanMessage], config_keys=["tags"]
    )
    # Invoke request
    async with get_async_client(app) as passthrough_runnable:
        with pytest.raises(httpx.HTTPError):
            await passthrough_runnable.ainvoke("Hello")
        with pytest.raises(httpx.HTTPError):
            await passthrough_runnable.ainvoke(["hello"])
        with pytest.raises(httpx.HTTPError):
            await passthrough_runnable.ainvoke(HumanMessage(content="h"))
        with pytest.raises(httpx.HTTPError):
            await passthrough_runnable.ainvoke([SystemMessage(content="hello")])

        # Valid
        result = await passthrough_runnable.ainvoke([HumanMessage(content="hello")])

        # Valid
        result = await passthrough_runnable.ainvoke(
            [HumanMessage(content="hello")], config={"tags": ["test"]}
        )

        assert isinstance(result, list)
        assert isinstance(result[0], HumanMessage)

    # Batch request
    async with get_async_client(app) as passthrough_runnable:
        # invalid
        with pytest.raises(httpx.HTTPError):
            await passthrough_runnable.abatch("Hello")
        with pytest.raises(httpx.HTTPError):
            await passthrough_runnable.abatch(["hello"])
        with pytest.raises(httpx.HTTPError):
            await passthrough_runnable.abatch([[SystemMessage(content="hello")]])

        # valid
        result = await passthrough_runnable.abatch([[HumanMessage(content="hello")]])
        assert isinstance(result, list)
        assert isinstance(result[0], list)
        assert isinstance(result[0][0], HumanMessage)


def test_client_close() -> None:
    """Test that the client can be automatically."""
    runnable = RemoteRunnable(url="/dev/null", timeout=1)
    sync_client = runnable.sync_client
    async_client = runnable.async_client
    assert async_client.is_closed is False
    assert sync_client.is_closed is False
    del runnable
    assert sync_client.is_closed is True
    assert async_client.is_closed is True


@pytest.mark.asyncio
async def test_async_client_close() -> None:
    """Test that the client can be automatically."""
    runnable = RemoteRunnable(url="/dev/null", timeout=1)
    sync_client = runnable.sync_client
    async_client = runnable.async_client
    assert async_client.is_closed is False
    assert sync_client.is_closed is False
    del runnable
    assert sync_client.is_closed is True
    assert async_client.is_closed is True


@pytest.mark.asyncio
async def test_openapi_docs_with_identical_runnables(
    event_loop: AbstractEventLoop, mocker: MockerFixture
) -> None:
    """Test client side and server side exceptions."""

    async def add_one(x: int) -> int:
        """Add one to simulate a valid function"""
        return x + 1

    server_runnable = RunnableLambda(func=add_one)
    server_runnable2 = RunnableLambda(func=add_one)

    app = FastAPI()
    add_routes(
        app,
        server_runnable,
        path="/a",
    )
    # Add another route that uses the same schema (inferred from runnable input schema)
    add_routes(
        app,
        server_runnable2,
        path="/b",
        config_keys=["tags"],
    )

    async with AsyncClient(app=app, base_url="http://localhost:9999") as async_client:
        response = await async_client.get("/openapi.json")
        assert response.status_code == 200


@pytest.mark.asyncio
async def test_configurable_runnables(event_loop: AbstractEventLoop) -> None:
    """Add tests for using langchain's configurable runnables"""

    template = PromptTemplate.from_template("say {name}").configurable_fields(
        template=ConfigurableField(
            id="template",
            name="Template",
            description="The template to use for the prompt",
        )
    )
    llm = (
        RunnablePassthrough() | RunnableLambda(lambda prompt: prompt.text)
    ).configurable_alternatives(
        ConfigurableField(
            id="llm",
            name="LLM",
        ),
        hardcoded_llm=FakeListLLM(responses=["hello Mr. Kitten!"]),
    )
    chain = template | llm
    # Check server side
    assert chain.invoke({"name": "cat"}) == "say cat"

    app = FastAPI()
    add_routes(app, chain, config_keys=["tags", "configurable"])

    async with get_async_client(app) as remote_runnable:
        # Test with hard-coded LLM
        assert chain.invoke({"name": "cat"}) == "say cat"
        # Test with different prompt

        assert (
            await remote_runnable.ainvoke(
                {"name": "foo"},
                {"configurable": {"template": "hear {name}"}, "tags": ["h"]},
            )
            == "hear foo"
        )
        # Test with alternative passthrough LLM
        assert (
            await remote_runnable.ainvoke(
                {"name": "foo"},
                {"configurable": {"llm": "hardcoded_llm"}, "tags": ["h"]},
            )
            == "hello Mr. Kitten!"
        )

tests/unit_tests/test_server_client.py
langchain-ai/langserve
from typing import List, Optional

import pytest
from langchain.prompts import PromptTemplate
from langchain.schema.runnable.utils import ConfigurableField

from langserve.server import _unpack_config

try:
    from pydantic.v1 import BaseModel, ValidationError
except ImportError:
    from pydantic import BaseModel, ValidationError

from langserve.validation import (
    create_batch_request_model,
    create_invoke_request_model,
)


@pytest.mark.parametrize(
    "test_case",
    [
        {
            "input": {"a": "qqq"},
            "kwargs": {},
            "valid": False,
        },
        {
            "input": {"a": 2},
            "kwargs": "hello",
            "valid": False,
        },
        {
            "input": {"a": 2},
            "config": "hello",
            "valid": False,
        },
        {
            "input": {"b": "hello"},
            "valid": False,
        },
        {
            "input": {"a": 2, "b": "hello"},
            "config": "hello",
            "valid": False,
        },
        {
            "input": {"a": 2, "b": "hello"},
            "valid": True,
        },
        {
            "input": {"a": 2, "b": "hello"},
            "valid": True,
        },
        {
            "input": {"a": 2},
            "valid": True,
        },
    ],
)
def test_create_invoke_and_batch_models(test_case: dict) -> None:
    """Test that the invoke request model is created correctly."""

    class Input(BaseModel):
        """Test input."""

        a: int
        b: Optional[str] = None

    valid = test_case.pop("valid")

    class Config(BaseModel):
        tags: Optional[List[str]] = None

    model = create_invoke_request_model("namespace", Input, Config)

    if valid:
        model(**test_case)
    else:
        with pytest.raises(ValidationError):
            model(**test_case)

    # Validate batch request
    # same structure as input request, but
    # 'input' is a list of inputs and is called 'inputs'
    batch_model = create_batch_request_model("namespace", Input, Config)

    test_case["inputs"] = [test_case.pop("input")]
    if valid:
        batch_model(**test_case)
    else:
        with pytest.raises(ValidationError):
            batch_model(**test_case)


@pytest.mark.parametrize(
    "test_case",
    [
        {
            "type": int,
            "input": 1,
            "valid": True,
        },
        {
            "type": float,
            "input": "name",
            "valid": False,
        },
        {
            "type": float,
            "input": [3.2],
            "valid": False,
        },
        {
            "type": float,
            "input": 1.1,
            "valid": True,
        },
        {
            "type": Optional[float],
            "valid": True,
            "input": None,
        },
    ],
)
def test_validation(test_case) -> None:
    """Test that the invoke request model is created correctly."""

    class Config(BaseModel):
        tags: Optional[List[str]] = None

    model = create_invoke_request_model("namespace", test_case.pop("type"), Config)

    if test_case["valid"]:
        model(**test_case)
    else:
        with pytest.raises(ValidationError):
            model(**test_case)


def test_invoke_request_with_runnables() -> None:
    """Test that the invoke request model is created correctly."""
    runnable = PromptTemplate.from_template("say hello to {name}").configurable_fields(
        template=ConfigurableField(
            id="template",
            name="Template",
            description="The template to use for the prompt",
        )
    )
    config = runnable.config_schema(include=["tags", "run_name", "configurable"])
    Model = create_invoke_request_model("", runnable.input_schema, config)

    assert (
        _unpack_config(
            Model(
                input={"name": "bob"},
            ).config,
            [],
        )
        == {}
    )

    # Test that the config is unpacked correctly
    request = Model(
        input={"name": "bob"},
        config={
            "tags": ["hello"],
            "run_name": "run",
            "configurable": {"template": "goodbye {name}"},
        },
    )
    assert request.input == {"name": "bob"}
    assert request.config.tags == ["hello"]
    assert request.config.run_name == "run"
    assert isinstance(request.config.configurable, BaseModel)
    assert request.config.configurable.dict() == {
        "template": "goodbye {name}",
    }

    assert _unpack_config(request.config, ["configurable"]) == {
        "configurable": {"template": "goodbye {name}"},
    }

tests/unit_tests/test_validation.py
langchain-ai/langserve
def test_version_available() -> None:
    """Test that a version is available."""
    from langserve import __version__  # noqa: F401

tests/unit_tests/test_version.py
langchain-ai/langserve
from typing import Any, List, Mapping, Optional

from langchain.callbacks.manager import (
    AsyncCallbackManagerForLLMRun,
    CallbackManagerForLLMRun,
)
from langchain.llms.base import LLM


class FakeListLLM(LLM):
    """Fake LLM for testing purposes."""

    responses: List[str]
    sleep: Optional[float] = None
    i: int = 0

    @property
    def _llm_type(self) -> str:
        """Return type of llm."""
        return "fake-list"

    def _call(
        self,
        prompt: str,
        stop: Optional[List[str]] = None,
        run_manager: Optional[CallbackManagerForLLMRun] = None,
        **kwargs: Any,
    ) -> str:
        """Return next response"""
        response = self.responses[self.i]
        if self.i < len(self.responses) - 1:
            self.i += 1
        else:
            self.i = 0
        return response

    async def _acall(
        self,
        prompt: str,
        stop: Optional[List[str]] = None,
        run_manager: Optional[AsyncCallbackManagerForLLMRun] = None,
        **kwargs: Any,
    ) -> str:
        """Return next response"""
        response = self.responses[self.i]
        if self.i < len(self.responses) - 1:
            self.i += 1
        else:
            self.i = 0
        return response

    @property
    def _identifying_params(self) -> Mapping[str, Any]:
        return {"responses": self.responses}

tests/unit_tests/utils.py
langchain-ai/langserve