v0.3 release (#767)

Allow users to define a custom serializer

langserve/api_handler.py

@@ -42,6 +42,7 @@ from langsmith import client as ls_client
 from langsmith.schemas import FeedbackIngestToken
 from langsmith.utils import tracing_is_enabled
 from pydantic import BaseModel, Field, RootModel, ValidationError, create_model
+from pydantic.v1 import BaseModel as BaseModelV1
 from starlette.requests import Request
 from starlette.responses import JSONResponse, Response
 from typing_extensions import TypedDict
@@ -61,7 +62,7 @@ from langserve.schema import (
     PublicTraceLink,
     PublicTraceLinkCreateRequest,
 )
-from langserve.serialization import WellKnownLCSerializer
+from langserve.serialization import Serializer, WellKnownLCSerializer
 from langserve.validation import (
     BatchBaseResponse,
     BatchRequestShallowValidator,
@@ -534,6 +535,8 @@ class APIHandler:
         per_req_config_modifier: Optional[PerRequestConfigModifier] = None,
         stream_log_name_allow_list: Optional[Sequence[str]] = None,
         playground_type: Literal["default", "chat"] = "default",
+        astream_events_version: Literal["v1", "v2"] = "v2",
+        serializer: Optional[Serializer] = None,
     ) -> None:
         """Create an API handler for the given runnable.
 
@@ -568,6 +571,7 @@ class APIHandler:
                 If true, the client will be able to show trace information
                 including events that occurred on the server side.
                 Be sure not to include any sensitive information in the callback events.
+                This is a **beta** API.
             enable_feedback_endpoint: Whether to enable an endpoint for logging feedback
                 to LangSmith. Disabled by default. If this flag is disabled or LangSmith
                 tracing is not enabled for the runnable, then 4xx errors will be thrown
@@ -595,6 +599,10 @@ class APIHandler:
                 If not provided, then all logs will be allowed to be streamed.
                 Use to also limit the events that can be streamed by the stream_events.
                 TODO: Introduce deprecation for this parameter to rename it
+            astream_events_version: version of the stream events endpoint to use.
+                By default "v2".
+            serializer: optional serializer to use for serializing the output.
+                If not provided, the default serializer will be used.
         """
         if importlib.util.find_spec("sse_starlette") is None:
             raise ImportError(
@@ -626,12 +634,18 @@ class APIHandler:
         # and when tracing information is logged, we'll be able to see
         # traces for the path /foo/bar.
         self._run_name = self._base_url
+        if include_callback_events:
+            warn_beta(
+                message="Including callback events in the response is in beta. "
+                "This API may change in the future."
+            )
         self._include_callback_events = include_callback_events
         self._per_req_config_modifier = per_req_config_modifier
-        self._serializer = WellKnownLCSerializer()
+        self._serializer = serializer or WellKnownLCSerializer()
         self._enable_feedback_endpoint = enable_feedback_endpoint
         self._enable_public_trace_link_endpoint = enable_public_trace_link_endpoint
         self._names_in_stream_allow_list = stream_log_name_allow_list
+        self._astream_events_version = astream_events_version
 
         if token_feedback_config:
             if len(token_feedback_config["key_configs"]) != 1:
@@ -671,15 +685,57 @@ class APIHandler:
 
         model_namespace = _replace_non_alphanumeric_with_underscores(path.strip("/"))
 
-        input_type_ = _resolve_model(
-            runnable.get_input_schema(), "Input", model_namespace
-        )
+        try:
+            input_type_ = _resolve_model(
+                runnable.get_input_schema(), "Input", model_namespace
+            )
+        except Exception as e:
+            # Attempt to surface a more informative user facing error
+            raise_original_error = True
+            try:
+                if isinstance(runnable.get_input_schema(), BaseModelV1):
+                    raise_original_error = False
+                    raise ValueError(
+                        "Found an input type which is a pydantic v1 model."
+                        "Please use pydantic.BaseModel rather than "
+                        "pydantic.v1.BaseModel."
+                    )
+            finally:  # noqa
+                if raise_original_error:
+                    print(
+                        "Encountered an error while resolving the inputs of "
+                        "the Runnable. Try specifying the input type explicitly "
+                        "using the `with_types` method on the runnable.\n"
+                        "See https://api.python.langchain.com/en/latest/runnables/langchain_core.runnables.base.Runnable.html "  # noqa: E501
+                    )
+                    raise e
 
-        output_type_ = _resolve_model(
-            runnable.get_output_schema(),
-            "Output",
-            model_namespace,
-        )
+        try:
+            output_type_ = _resolve_model(
+                runnable.get_output_schema(),
+                "Output",
+                model_namespace,
+            )
+        except Exception as e:
+            # Attempt to surface a more informative user facing error
+            raise_original_error = True
+            try:
+                if isinstance(runnable.get_output_schema(), BaseModelV1):
+                    raise_original_error = False
+                    raise ValueError(
+                        "Found an output type which is a pydantic v1 model."
+                        "Please use pydantic.BaseModel rather than "
+                        "pydantic.v1.BaseModel."
+                    )
+            finally:  # noqa
+                if raise_original_error:
+                    print(
+                        "Encountered an error while resolving the inputs of "
+                        "the Runnable. Try specifying the output type explicitly "
+                        "using the `with_types` method on the runnable.\n"
+                        "See https://api.python.langchain.com/en/latest/runnables/langchain_core.runnables.base.Runnable.html "  # noqa: E501
+                    )
+                    raise e
 
         self._ConfigPayload = _add_namespace_to_model(
             model_namespace, runnable.config_schema(include=config_keys)
@@ -1343,7 +1399,7 @@ class APIHandler:
                     exclude_names=stream_events_request.exclude_names,
                     exclude_types=stream_events_request.exclude_types,
                     exclude_tags=stream_events_request.exclude_tags,
-                    version="v1",
+                    version=self._astream_events_version,
                 ):
                     if (
                         self._names_in_stream_allow_list is None

langserve/callbacks.py

@@ -48,7 +48,7 @@ class AsyncEventAggregatorCallback(AsyncCallbackHandler):
 
     async def on_chat_model_start(
         self,
-        serialized: Dict[str, Any],
+        serialized: Optional[Dict[str, Any]],
         messages: List[List[BaseMessage]],
         *,
         run_id: UUID,
@@ -73,7 +73,7 @@ class AsyncEventAggregatorCallback(AsyncCallbackHandler):
 
     async def on_chain_start(
         self,
-        serialized: Dict[str, Any],
+        serialized: Optional[Dict[str, Any]],
         inputs: Dict[str, Any],
         *,
         run_id: UUID,
@@ -138,7 +138,7 @@ class AsyncEventAggregatorCallback(AsyncCallbackHandler):
 
     async def on_retriever_start(
         self,
-        serialized: Dict[str, Any],
+        serialized: Optional[Dict[str, Any]],
         query: str,
         *,
         run_id: UUID,
@@ -202,7 +202,7 @@ class AsyncEventAggregatorCallback(AsyncCallbackHandler):
 
     async def on_tool_start(
         self,
-        serialized: Dict[str, Any],
+        serialized: Optional[Dict[str, Any]],
         input_str: str,
         *,
         run_id: UUID,
@@ -306,7 +306,7 @@ class AsyncEventAggregatorCallback(AsyncCallbackHandler):
 
     async def on_llm_start(
         self,
-        serialized: Dict[str, Any],
+        serialized: Optional[Dict[str, Any]],
         prompts: List[str],
         *,
         run_id: UUID,
@@ -445,7 +445,14 @@ async def ahandle_callbacks(
         if event["parent_run_id"] is None:  # How do we make sure it's None!?
             event["parent_run_id"] = callback_manager.run_id
 
-        event_data = {key: value for key, value in event.items() if key != "type"}
+        event_data = {
+            key: value
+            for key, value in event.items()
+            if key != "type" and key != "kwargs"
+        }
+
+        if "kwargs" in event:
+            event_data.update(event["kwargs"])
 
         await ahandle_event(
             # Unpacking like this may not work
@@ -467,7 +474,14 @@ def handle_callbacks(
         if event["parent_run_id"] is None:  # How do we make sure it's None!?
             event["parent_run_id"] = callback_manager.run_id
 
-        event_data = {key: value for key, value in event.items() if key != "type"}
+        event_data = {
+            key: value
+            for key, value in event.items()
+            if key != "type" and key != "kwargs"
+        }
+
+        if "kwargs" in event:
+            event_data.update(event["kwargs"])
 
         handle_event(
             # Unpacking like this may not work

langserve/client.py

@@ -120,6 +120,12 @@ def _log_error_message_once(error_message: str) -> None:
     logger.error(error_message)
 
 
+@lru_cache(maxsize=1_000)  # Will accommodate up to 1_000 different error messages
+def _log_info_message_once(error_message: str) -> None:
+    """Log an error message once."""
+    logger.info(error_message)
+
+
 def _sanitize_request(request: httpx.Request) -> httpx.Request:
     """Remove sensitive headers from the request."""
     accept_headers = {
@@ -279,6 +285,7 @@ class RemoteRunnable(Runnable[Input, Output]):
         cert: Optional[CertTypes] = None,
         client_kwargs: Optional[Dict[str, Any]] = None,
         use_server_callback_events: bool = True,
+        serializer: Optional[Serializer] = None,
     ) -> None:
         """Initialize the client.
 
@@ -294,6 +301,8 @@ class RemoteRunnable(Runnable[Input, Output]):
                 and async httpx clients
             use_server_callback_events: Whether to invoke callbacks on any
                 callback events returned by the server.
+            serializer: The serializer to use for serializing and deserializing
+                data. If not provided, a default serializer will be used.
         """
         _client_kwargs = client_kwargs or {}
         # Enforce trailing slash
@@ -321,7 +330,7 @@ class RemoteRunnable(Runnable[Input, Output]):
 
         # Register cleanup handler once RemoteRunnable is garbage collected
         weakref.finalize(self, _close_clients, self.sync_client, self.async_client)
-        self._lc_serializer = WellKnownLCSerializer()
+        self._lc_serializer = serializer or WellKnownLCSerializer()
         self._use_server_callback_events = use_server_callback_events
 
     def _invoke(
@@ -752,7 +761,7 @@ class RemoteRunnable(Runnable[Input, Output]):
         input: Any,
         config: Optional[RunnableConfig] = None,
         *,
-        version: Literal["v1"],
+        version: Literal["v1", "v2", None] = None,
         include_names: Optional[Sequence[str]] = None,
         include_types: Optional[Sequence[str]] = None,
         include_tags: Optional[Sequence[str]] = None,
@@ -775,7 +784,8 @@ class RemoteRunnable(Runnable[Input, Output]):
             input: The input to the runnable
             config: The config to use for the runnable
             version: The version of the astream_events to use.
-                     Currently only "v1" is supported.
+                Currently, this input is IGNORED on the client.
+                The server will return whatever format it's configured with.
             include_names: The names of the events to include
             include_types: The types of the events to include
             include_tags: The tags of the events to include
@@ -783,13 +793,18 @@ class RemoteRunnable(Runnable[Input, Output]):
             exclude_types: The types of the events to exclude
             exclude_tags: The tags of the events to exclude
         """
-        if version != "v1":
-            raise ValueError(f"Unsupported version: {version}. Use 'v1'")
-
         # Create a stream handler that will emit Log objects
         config = ensure_config(config)
         callback_manager = get_async_callback_manager_for_config(config)
 
+        if version is not None:
+            _log_info_message_once(
+                "Versioning of the astream_events API is not supported on the client "
+                "side currently. The server will return events in whatever format "
+                "it was configured with in add_routes or APIHandler. "
+                "To stop seeing this message, remove the `version` argument."
+            )
+
         events = []
 
         run_manager = await callback_manager.on_chain_start(

langserve/serialization.py

@@ -157,40 +157,50 @@ def _decode_event_data(value: Any) -> Any:
 
 
 class Serializer(abc.ABC):
-    @abc.abstractmethod
     def dumpd(self, obj: Any) -> Any:
         """Convert the given object to a JSON serializable object."""
-
-    @abc.abstractmethod
-    def dumps(self, obj: Any) -> bytes:
-        """Dump the given object as a JSON string."""
-
-    @abc.abstractmethod
-    def loads(self, s: bytes) -> Any:
-        """Load the given JSON string."""
-
-    @abc.abstractmethod
-    def loadd(self, obj: Any) -> Any:
-        """Load the given object."""
-
-
-class WellKnownLCSerializer(Serializer):
-    def dumpd(self, obj: Any) -> Any:
-        """Convert the given object to a JSON serializable object."""
-        return orjson.loads(orjson.dumps(obj, default=default))
-
-    def dumps(self, obj: Any) -> bytes:
-        """Dump the given object as a JSON string."""
-        return orjson.dumps(obj, default=default)
-
-    def loadd(self, obj: Any) -> Any:
-        """Load the given object."""
-        return _decode_lc_objects(obj)
+        return orjson.loads(self.dumps(obj))
 
     def loads(self, s: bytes) -> Any:
         """Load the given JSON string."""
         return self.loadd(orjson.loads(s))
 
+    @abc.abstractmethod
+    def dumps(self, obj: Any) -> bytes:
+        """Dump the given object to a JSON byte string."""
+
+    @abc.abstractmethod
+    def loadd(self, s: bytes) -> Any:
+        """Given a python object, load it into a well known object.
+
+        The obj represents content that was json loaded from a string, but
+        not yet validated or converted into a well known object.
+        """
+
+
+class WellKnownLCSerializer(Serializer):
+    """A pre-defined serializer for well known LangChain objects.
+
+    This is the default serialized used by LangServe for serializing and
+    de-serializing well known LangChain objects.
+
+    If you need to extend the serialization capabilities for your own application,
+    feel free to create a new instance of the Serializer class and implement
+    the abstract methods dumps and loadd.
+    """
+
+    def dumps(self, obj: Any) -> bytes:
+        """Dump the given object to a JSON byte string."""
+        return orjson.dumps(obj, default=default)
+
+    def loadd(self, obj: Any) -> Any:
+        """Given a python object, load it into a well known object.
+
+        The obj represents content that was json loaded from a string, but
+        not yet validated or converted into a well known object.
+        """
+        return _decode_lc_objects(obj)
+
 
 def _project_top_level(model: BaseModel) -> Dict[str, Any]:
     """Project the top level of the model as dict."""

langserve/server.py

@@ -26,6 +26,7 @@ from langserve.api_handler import (
     TokenFeedbackConfig,
     _is_hosted,
 )
+from langserve.serialization import Serializer
 
 try:
     from fastapi import APIRouter, Depends, FastAPI, Request, Response
@@ -262,6 +263,8 @@ def add_routes(
     enabled_endpoints: Optional[Sequence[EndpointName]] = None,
     dependencies: Optional[Sequence[Depends]] = None,
     playground_type: Literal["default", "chat"] = "default",
+    astream_events_version: Literal["v1", "v2"] = "v2",
+    serializer: Optional[Serializer] = None,
 ) -> None:
     """Register the routes on the given FastAPI app or APIRouter.
 
@@ -380,14 +383,18 @@ def add_routes(
             - chat: UX is optimized for chat-like interactions. Please review
               the README in langserve for more details about constraints (e.g.,
               which message types are supported etc.)
+        astream_events_version: version of the stream events endpoint to use.
+            By default "v2".
+        serializer: The serializer to use for serializing the output. If not provided,
+            the default serializer will be used.
     """  # noqa: E501
     if not isinstance(runnable, Runnable):
         raise TypeError(
             f"Expected a Runnable, got {type(runnable)}. "
-            f"The second argument to add_routes should be a Runnable instance."
-            f"add_route(app, runnable, ...) is the correct usage."
-            f"Please make sure that you are using a runnable which is an instance of "
-            f"langchain_core.runnables.Runnable."
+            "The second argument to add_routes should be a Runnable instance."
+            "add_route(app, runnable, ...) is the correct usage."
+            "Please make sure that you are using a runnable which is an instance of "
+            "langchain_core.runnables.Runnable."
         )
 
     endpoint_configuration = _EndpointConfiguration(
@@ -443,7 +450,10 @@ def add_routes(
         per_req_config_modifier=per_req_config_modifier,
         stream_log_name_allow_list=stream_log_name_allow_list,
         playground_type=playground_type,
+        astream_events_version=astream_events_version,
+        serializer=serializer,
     )
+
     namespace = path or ""
 
     route_tags = [path.strip("/")] if path else None

langserve/server_sent_events.py

@@ -1,8 +1,9 @@
 """Adapted from https://github.com/florimondmanca/httpx-sse"""
 from contextlib import asynccontextmanager, contextmanager
-from typing import Any, AsyncIterator, Iterator, List, Optional, TypedDict
+from typing import Any, AsyncIterator, Iterator, List, Optional
 
 import httpx
+from typing_extensions import TypedDict
 
 
 class ServerSentEvent(TypedDict):

langserve/validation.py

@@ -396,27 +396,29 @@ class StreamEventsParameters(BaseModel):
 # status code and a message.
 
 
-class OnChainStart(BaseModel):
-    """On Chain Start Callback Event."""
+class BaseCallback(BaseModel):
+    """Base class for all callback events."""
 
-    serialized: Dict[str, Any]
-    inputs: Any
     run_id: UUID
     parent_run_id: Optional[UUID] = None
     tags: Optional[List[str]] = None
     metadata: Optional[Dict[str, Any]] = None
-    kwargs: Any = None
+
+
+class OnChainStart(BaseCallback):
+    """On Chain Start Callback Event."""
+
+    serialized: Optional[Dict[str, Any]] = None
+    inputs: Any
+    kwargs: Optional[Dict[str, Any]] = None
     type: Literal["on_chain_start"] = "on_chain_start"
 
 
-class OnChainEnd(BaseModel):
+class OnChainEnd(BaseCallback):
     """On Chain End Callback Event."""
 
     outputs: Any
-    run_id: UUID
-    parent_run_id: Optional[UUID] = None
-    tags: Optional[List[str]] = None
-    kwargs: Any = None
+    kwargs: Optional[Dict[str, Any]] = None
     type: Literal["on_chain_end"] = "on_chain_end"
 
 
@@ -428,38 +430,35 @@ class Error(BaseModel):
     type: Literal["error"] = "error"
 
 
-class OnChainError(BaseModel):
+class OnChainError(BaseCallback):
     """On Chain Error Callback Event."""
 
     error: Error
-    run_id: UUID
-    parent_run_id: Optional[UUID] = None
-    tags: Optional[List[str]] = None
-    kwargs: Any = None
+    kwargs: Optional[Dict[str, Any]] = None
     type: Literal["on_chain_error"] = "on_chain_error"
 
 
-class OnToolStart(BaseModel):
+class OnToolStart(BaseCallback):
     """On Tool Start Callback Event."""
 
-    serialized: Dict[str, Any]
+    serialized: Optional[Dict[str, Any]] = None
     input_str: str
     run_id: UUID
     parent_run_id: Optional[UUID] = None
     tags: Optional[List[str]] = None
     metadata: Optional[Dict[str, Any]] = None
-    kwargs: Any = None
+    kwargs: Optional[Dict[str, Any]] = None
     type: Literal["on_tool_start"] = "on_tool_start"
 
 
-class OnToolEnd(BaseModel):
+class OnToolEnd(BaseCallback):
     """On Tool End Callback Event."""
 
     output: str
     run_id: UUID
     parent_run_id: Optional[UUID] = None
     tags: Optional[List[str]] = None
-    kwargs: Any = None
+    kwargs: Optional[Dict[str, Any]] = None
     type: Literal["on_tool_end"] = "on_tool_end"
 
 
@@ -467,36 +466,29 @@ class OnToolError(BaseModel):
     """On Tool Error Callback Event."""
 
     error: Error
-    run_id: UUID
-    parent_run_id: Optional[UUID] = None
-    tags: Optional[List[str]] = None
-    kwargs: Any = None
+    kwargs: Optional[Dict[str, Any]] = None
     type: Literal["on_tool_error"] = "on_tool_error"
 
 
-class OnChatModelStart(BaseModel):
+class OnChatModelStart(BaseCallback):
     """On Chat Model Start Callback Event."""
 
-    serialized: Dict[str, Any]
+    serialized: Optional[Dict[str, Any]] = None
     messages: List[List[BaseMessage]]
-    run_id: UUID
-    parent_run_id: Optional[UUID] = None
-    tags: Optional[List[str]] = None
-    metadata: Optional[Dict[str, Any]] = None
-    kwargs: Any = None
+    kwargs: Optional[Dict[str, Any]] = None
     type: Literal["on_chat_model_start"] = "on_chat_model_start"
 
 
-class OnLLMStart(BaseModel):
+class OnLLMStart(BaseCallback):
     """On LLM Start Callback Event."""
 
-    serialized: Dict[str, Any]
+    serialized: Optional[Dict[str, Any]] = None
     prompts: List[str]
     run_id: UUID
     parent_run_id: Optional[UUID] = None
     tags: Optional[List[str]] = None
     metadata: Optional[Dict[str, Any]] = None
-    kwargs: Any = None
+    kwargs: Optional[Dict[str, Any]] = None
     type: Literal["on_llm_start"] = "on_llm_start"
 
 
@@ -515,49 +507,39 @@ class LLMResult(BaseModel):
     """List of metadata info for model call for each input."""
 
 
-class OnLLMEnd(BaseModel):
+class OnLLMEnd(BaseCallback):
     """On LLM End Callback Event."""
 
     response: LLMResult
-    run_id: UUID
-    parent_run_id: Optional[UUID] = None
-    tags: Optional[List[str]] = None
-    kwargs: Any = None
+    kwargs: Optional[Dict[str, Any]] = None
     type: Literal["on_llm_end"] = "on_llm_end"
 
 
-class OnRetrieverStart(BaseModel):
+class OnRetrieverStart(BaseCallback):
     """On Retriever Start Callback Event."""
 
-    serialized: Dict[str, Any]
+    serialized: Optional[Dict[str, Any]] = None
     query: str
-    run_id: UUID
-    parent_run_id: Optional[UUID] = None
-    tags: Optional[List[str]] = None
-    metadata: Optional[Dict[str, Any]] = None
-    kwargs: Any = None
+    kwargs: Optional[Dict[str, Any]] = None
     type: Literal["on_retriever_start"] = "on_retriever_start"
 
 
-class OnRetrieverError(BaseModel):
+class OnRetrieverError(BaseCallback):
     """On Retriever Error Callback Event."""
 
     error: Error
     run_id: UUID
     parent_run_id: Optional[UUID] = None
     tags: Optional[List[str]] = None
-    kwargs: Any = None
+    kwargs: Optional[Dict[str, Any]] = None
     type: Literal["on_retriever_error"] = "on_retriever_error"
 
 
-class OnRetrieverEnd(BaseModel):
+class OnRetrieverEnd(BaseCallback):
     """On Retriever End Callback Event."""
 
     documents: Sequence[Document]
-    run_id: UUID
-    parent_run_id: Optional[UUID] = None
-    tags: Optional[List[str]] = None
-    kwargs: Any = None
+    kwargs: Optional[Dict[str, Any]] = None
     type: Literal["on_retriever_end"] = "on_retriever_end"
 
 

poetry.lock

@@ -826,13 +826,13 @@ tests = ["asttokens (>=2.1.0)", "coverage", "coverage-enable-subprocess", "ipyth
 
 [[package]]
 name = "fastapi"
-version = "0.114.1"
+version = "0.114.2"
 description = "FastAPI framework, high performance, easy to learn, fast to code, ready for production"
 optional = false
 python-versions = ">=3.8"
 files = [
-    {file = "fastapi-0.114.1-py3-none-any.whl", hash = "sha256:5d4746f6e4b7dff0b4f6b6c6d5445645285f662fe75886e99af7ee2d6b58bb3e"},
-    {file = "fastapi-0.114.1.tar.gz", hash = "sha256:1d7bbbeabbaae0acb0c22f0ab0b040f642d3093ca3645f8c876b6f91391861d8"},
+    {file = "fastapi-0.114.2-py3-none-any.whl", hash = "sha256:44474a22913057b1acb973ab90f4b671ba5200482e7622816d79105dcece1ac5"},
+    {file = "fastapi-0.114.2.tar.gz", hash = "sha256:0adb148b62edb09e8c6eeefa3ea934e8f276dabc038c5a82989ea6346050c3da"},
 ]
 
 [package.dependencies]
@@ -1062,15 +1062,18 @@ zstd = ["zstandard (>=0.18.0)"]
 
 [[package]]
 name = "idna"
-version = "3.8"
+version = "3.9"
 description = "Internationalized Domain Names in Applications (IDNA)"
 optional = false
 python-versions = ">=3.6"
 files = [
-    {file = "idna-3.8-py3-none-any.whl", hash = "sha256:050b4e5baadcd44d760cedbd2b8e639f2ff89bbc7a5730fcc662954303377aac"},
-    {file = "idna-3.8.tar.gz", hash = "sha256:d838c2c0ed6fced7693d5e8ab8e734d5f8fda53a039c0164afb0b82e771e3603"},
+    {file = "idna-3.9-py3-none-any.whl", hash = "sha256:69297d5da0cc9281c77efffb4e730254dd45943f45bbfb461de5991713989b1e"},
+    {file = "idna-3.9.tar.gz", hash = "sha256:e5c5dafde284f26e9e0f28f6ea2d6400abd5ca099864a67f576f3981c6476124"},
 ]
 
+[package.extras]
+all = ["flake8 (>=7.1.1)", "mypy (>=1.11.2)", "pytest (>=8.3.2)", "ruff (>=0.6.2)"]
+
 [[package]]
 name = "importlib-metadata"
 version = "8.5.0"
@@ -1564,41 +1567,44 @@ test = ["hatch", "ipykernel", "openapi-core (>=0.18.0,<0.19.0)", "openapi-spec-v
 
 [[package]]
 name = "langchain-core"
-version = "0.3.0.dev5"
+version = "0.3.0"
 description = "Building applications with LLMs through composability"
 optional = false
 python-versions = "<4.0,>=3.9"
 files = [
-    {file = "langchain_core-0.3.0.dev5-py3-none-any.whl", hash = "sha256:8de38ea9dd15f9705d20ee44af7160216052467d47893d5473e0e878861685be"},
-    {file = "langchain_core-0.3.0.dev5.tar.gz", hash = "sha256:d0986b3e2810522d90eb09d048ee4a92b079a334b313f4b7429624b4a062adff"},
+    {file = "langchain_core-0.3.0-py3-none-any.whl", hash = "sha256:bee6dae2366d037ef0c5b87401fed14b5497cad26f97724e8c9ca7bc9239e847"},
+    {file = "langchain_core-0.3.0.tar.gz", hash = "sha256:1249149ea3ba24c9c761011483c14091573a5eb1a773aa0db9c8ad155dd4a69d"},
 ]
 
 [package.dependencies]
 jsonpatch = ">=1.33,<2.0"
 langsmith = ">=0.1.117,<0.2.0"
 packaging = ">=23.2,<25"
-pydantic = ">=2.7.4,<3.0.0"
+pydantic = [
+    {version = ">=2.5.2,<3.0.0", markers = "python_full_version < \"3.12.4\""},
+    {version = ">=2.7.4,<3.0.0", markers = "python_full_version >= \"3.12.4\""},
+]
 PyYAML = ">=5.3"
 tenacity = ">=8.1.0,<8.4.0 || >8.4.0,<9.0.0"
 typing-extensions = ">=4.7"
 
 [[package]]
 name = "langsmith"
-version = "0.1.117"
+version = "0.1.120"
 description = "Client library to connect to the LangSmith LLM Tracing and Evaluation Platform."
 optional = false
 python-versions = "<4.0,>=3.8.1"
 files = [
-    {file = "langsmith-0.1.117-py3-none-any.whl", hash = "sha256:e936ee9bcf8293b0496df7ba462a3702179fbe51f9dc28744b0fbec0dbf206ae"},
-    {file = "langsmith-0.1.117.tar.gz", hash = "sha256:a1b532f49968b9339bcaff9118d141846d52ed3d803f342902e7448edf1d662b"},
+    {file = "langsmith-0.1.120-py3-none-any.whl", hash = "sha256:54d2785e301646c0988e0a69ebe4d976488c87b41928b358cb153b6ddd8db62b"},
+    {file = "langsmith-0.1.120.tar.gz", hash = "sha256:25499ca187b41bd89d784b272b97a8d76f60e0e21bdf20336e8a2aa6a9b23ac9"},
 ]
 
 [package.dependencies]
 httpx = ">=0.23.0,<1"
 orjson = ">=3.9.14,<4.0.0"
 pydantic = [
-    {version = ">=2.7.4,<3.0.0", markers = "python_full_version >= \"3.12.4\""},
     {version = ">=1,<3", markers = "python_full_version < \"3.12.4\""},
+    {version = ">=2.7.4,<3.0.0", markers = "python_full_version >= \"3.12.4\""},
 ]
 requests = ">=2,<3"
 
@@ -2117,13 +2123,13 @@ ptyprocess = ">=0.5"
 
 [[package]]
 name = "platformdirs"
-version = "4.3.2"
+version = "4.3.3"
 description = "A small Python package for determining appropriate platform-specific dirs, e.g. a `user data dir`."
 optional = false
 python-versions = ">=3.8"
 files = [
-    {file = "platformdirs-4.3.2-py3-none-any.whl", hash = "sha256:eb1c8582560b34ed4ba105009a4badf7f6f85768b30126f351328507b2beb617"},
-    {file = "platformdirs-4.3.2.tar.gz", hash = "sha256:9e5e27a08aa095dd127b9f2e764d74254f482fef22b0970773bfba79d091ab8c"},
+    {file = "platformdirs-4.3.3-py3-none-any.whl", hash = "sha256:50a5450e2e84f44539718293cbb1da0a0885c9d14adf21b77bae4e66fc99d9b5"},
+    {file = "platformdirs-4.3.3.tar.gz", hash = "sha256:d4e0b7d8ec176b341fb03cb11ca12d0276faa8c485f9cd218f613840463fc2c0"},
 ]
 
 [package.extras]
@@ -2254,8 +2260,8 @@ files = [
 annotated-types = ">=0.6.0"
 pydantic-core = "2.23.3"
 typing-extensions = [
-    {version = ">=4.12.2", markers = "python_version >= \"3.13\""},
     {version = ">=4.6.1", markers = "python_version < \"3.13\""},
+    {version = ">=4.12.2", markers = "python_version >= \"3.13\""},
 ]
 
 [package.extras]
@@ -2377,23 +2383,6 @@ files = [
 [package.extras]
 windows-terminal = ["colorama (>=0.4.6)"]
 
-[[package]]
-name = "pyproject-toml"
-version = "0.0.10"
-description = "Project intend to implement PEP 517, 518, 621, 631 and so on."
-optional = false
-python-versions = "*"
-files = [
-    {file = "pyproject-toml-0.0.10.tar.gz", hash = "sha256:f0ce0e9934ecb00c0e529b4a1c380edd3034c4be65516769c5f080bdb23dfcb3"},
-    {file = "pyproject_toml-0.0.10-py3-none-any.whl", hash = "sha256:257a7070617e1a0bcfd8f790817b30bd9193876023a9b9e7a6b4fc976acf4c3e"},
-]
-
-[package.dependencies]
-jsonschema = "*"
-setuptools = ">=42"
-toml = "*"
-wheel = "*"
-
 [[package]]
 name = "pytest"
 version = "7.4.4"
@@ -2988,26 +2977,6 @@ nativelib = ["pyobjc-framework-Cocoa", "pywin32"]
 objc = ["pyobjc-framework-Cocoa"]
 win32 = ["pywin32"]
 
-[[package]]
-name = "setuptools"
-version = "74.1.2"
-description = "Easily download, build, install, upgrade, and uninstall Python packages"
-optional = false
-python-versions = ">=3.8"
-files = [
-    {file = "setuptools-74.1.2-py3-none-any.whl", hash = "sha256:5f4c08aa4d3ebcb57a50c33b1b07e94315d7fc7230f7115e47fc99776c8ce308"},
-    {file = "setuptools-74.1.2.tar.gz", hash = "sha256:95b40ed940a1c67eb70fc099094bd6e99c6ee7c23aa2306f4d2697ba7916f9c6"},
-]
-
-[package.extras]
-check = ["pytest-checkdocs (>=2.4)", "pytest-ruff (>=0.2.1)", "ruff (>=0.5.2)"]
-core = ["importlib-metadata (>=6)", "importlib-resources (>=5.10.2)", "jaraco.text (>=3.7)", "more-itertools (>=8.8)", "packaging (>=24)", "platformdirs (>=2.6.2)", "tomli (>=2.0.1)", "wheel (>=0.43.0)"]
-cover = ["pytest-cov"]
-doc = ["furo", "jaraco.packaging (>=9.3)", "jaraco.tidelift (>=1.4)", "pygments-github-lexers (==0.0.5)", "pyproject-hooks (!=1.1)", "rst.linker (>=1.9)", "sphinx (>=3.5)", "sphinx-favicon", "sphinx-inline-tabs", "sphinx-lint", "sphinx-notfound-page (>=1,<2)", "sphinx-reredirects", "sphinxcontrib-towncrier", "towncrier (<24.7)"]
-enabler = ["pytest-enabler (>=2.2)"]
-test = ["build[virtualenv] (>=1.0.3)", "filelock (>=3.4.0)", "ini2toml[lite] (>=0.14)", "jaraco.develop (>=7.21)", "jaraco.envs (>=2.2)", "jaraco.path (>=3.2.0)", "jaraco.test", "packaging (>=23.2)", "pip (>=19.1)", "pyproject-hooks (!=1.1)", "pytest (>=6,!=8.1.*)", "pytest-home (>=0.5)", "pytest-perf", "pytest-subprocess", "pytest-timeout", "pytest-xdist (>=3)", "tomli-w (>=1.0.0)", "virtualenv (>=13.0.0)", "wheel (>=0.44.0)"]
-type = ["importlib-metadata (>=7.0.2)", "jaraco.develop (>=7.21)", "mypy (==1.11.*)", "pytest-mypy"]
-
 [[package]]
 name = "six"
 version = "1.16.0"
@@ -3149,17 +3118,6 @@ webencodings = ">=0.4"
 doc = ["sphinx", "sphinx_rtd_theme"]
 test = ["pytest", "ruff"]
 
-[[package]]
-name = "toml"
-version = "0.10.2"
-description = "Python Library for Tom's Obvious, Minimal Language"
-optional = false
-python-versions = ">=2.6, !=3.0.*, !=3.1.*, !=3.2.*"
-files = [
-    {file = "toml-0.10.2-py2.py3-none-any.whl", hash = "sha256:806143ae5bfb6a3c6e736a764057db0e6a0e05e338b5630894a5f779cabb4f9b"},
-    {file = "toml-0.10.2.tar.gz", hash = "sha256:b3bda1d108d5dd99f4a20d24d9c348e91c4db7ab1b749200bded2f839ccbe68f"},
-]
-
 [[package]]
 name = "tomli"
 version = "2.0.1"
@@ -3264,13 +3222,13 @@ dev = ["flake8", "flake8-annotations", "flake8-bandit", "flake8-bugbear", "flake
 
 [[package]]
 name = "urllib3"
-version = "2.2.2"
+version = "2.2.3"
 description = "HTTP library with thread-safe connection pooling, file post, and more."
 optional = false
 python-versions = ">=3.8"
 files = [
-    {file = "urllib3-2.2.2-py3-none-any.whl", hash = "sha256:a448b2f64d686155468037e1ace9f2d2199776e17f0a46610480d311f73e3472"},
-    {file = "urllib3-2.2.2.tar.gz", hash = "sha256:dd505485549a7a552833da5e6063639d0d177c04f23bc3864e41e5dc5f612168"},
+    {file = "urllib3-2.2.3-py3-none-any.whl", hash = "sha256:ca899ca043dcb1bafa3e262d73aa25c465bfb49e0bd9dd5d59f1d0acba2f8fac"},
+    {file = "urllib3-2.2.3.tar.gz", hash = "sha256:e7d814a81dad81e6caf2ec9fdedb284ecc9c73076b62654547cc64ccdcae26e9"},
 ]
 
 [package.extras]
@@ -3634,20 +3592,6 @@ files = [
     {file = "websockets-13.0.1.tar.gz", hash = "sha256:4d6ece65099411cfd9a48d13701d7438d9c34f479046b34c50ff60bb8834e43e"},
 ]
 
-[[package]]
-name = "wheel"
-version = "0.44.0"
-description = "A built-package format for Python"
-optional = false
-python-versions = ">=3.8"
-files = [
-    {file = "wheel-0.44.0-py3-none-any.whl", hash = "sha256:2376a90c98cc337d18623527a97c31797bd02bad0033d41547043a1cbfbe448f"},
-    {file = "wheel-0.44.0.tar.gz", hash = "sha256:a29c3f2817e95ab89aa4660681ad547c0e9547f20e75b0562fe7723c9a2a9d49"},
-]
-
-[package.extras]
-test = ["pytest (>=6.0.0)", "setuptools (>=65)"]
-
 [[package]]
 name = "y-py"
 version = "0.6.2"
@@ -3857,13 +3801,13 @@ test = ["mypy", "pre-commit", "pytest", "pytest-asyncio", "websockets (>=10.0)"]
 
 [[package]]
 name = "zipp"
-version = "3.20.1"
+version = "3.20.2"
 description = "Backport of pathlib-compatible object wrapper for zip files"
 optional = false
 python-versions = ">=3.8"
 files = [
-    {file = "zipp-3.20.1-py3-none-any.whl", hash = "sha256:9960cd8967c8f85a56f920d5d507274e74f9ff813a0ab8889a5b5be2daf44064"},
-    {file = "zipp-3.20.1.tar.gz", hash = "sha256:c22b14cc4763c5a5b04134207736c107db42e9d3ef2d9779d465f5f1bcba572b"},
+    {file = "zipp-3.20.2-py3-none-any.whl", hash = "sha256:a817ac80d6cf4b23bf7f2828b7cabf326f15a001bea8b1f9b49631780ba28350"},
+    {file = "zipp-3.20.2.tar.gz", hash = "sha256:bc9eb26f4506fda01b81bcde0ca78103b6e62f991b381fec825435c836edbc29"},
 ]
 
 [package.extras]
@@ -3882,4 +3826,4 @@ server = ["fastapi", "sse-starlette"]
 [metadata]
 lock-version = "2.0"
 python-versions = "^3.9"
-content-hash = "bee76de01384e265c807d785959226cdf9c788b4d5dfc3772dcce5d70a8f6121"
+content-hash = "8bb5ca68199c071d2d95d1bec5d4cf3e36a0e0535853b2370cfb594ba0808c0d"

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "langserve"
-version = "0.3.0dev2"
+version = "0.3.0"
 description = ""
 readme = "README.md"
 authors = ["LangChain"]
@@ -11,12 +11,11 @@ include = ["langserve/playground/dist/**/*", "langserve/chat_playground/dist/**/
 
 [tool.poetry.dependencies]
 python = "^3.9"
-httpx = ">=0.23.0" # May be able to decrease this version
+httpx = ">=0.23.0,<1.0"
 fastapi = {version = ">=0.90.1,<1", optional = true}
 sse-starlette = {version = "^1.3.0", optional = true}
-langchain-core = "0.3.0dev5"
-orjson = ">=2"
-pyproject-toml = "^0.0.10"
+langchain-core = "^0.3"
+orjson = ">=2,<4"
 pydantic = "^2.7"
 
 [tool.poetry.group.dev.dependencies]

tests/unit_tests/test_serialization.py

@@ -189,9 +189,3 @@ def test_encoding_of_well_known_types(obj: Any, expected: str) -> None:
     """
     lc_serializer = WellKnownLCSerializer()
     assert lc_serializer.dumpd(obj) == expected
-
-
-@pytest.mark.xfail(reason="0.3")
-def test_fail_03() -> None:
-    """This test will fail on purposes. It contains a TODO list for 0.3 release."""
-    assert "CHatGeneration_Deserialized correct" == "UNcomment test above"

tests/unit_tests/test_server_client.py

@@ -56,6 +56,7 @@ from langchain_core.tracers import RunLog, RunLogPatch
 from langsmith import schemas as ls_schemas
 from langsmith.client import Client
 from langsmith.schemas import FeedbackIngestToken
+from orjson import orjson
 from pydantic import BaseModel, Field, __version__
 from pytest import MonkeyPatch
 from pytest_mock import MockerFixture
@@ -122,6 +123,19 @@ def _replace_run_id_in_stream_resp(streamed_resp: str) -> str:
     return streamed_resp.replace(uuid, "<REPLACED>")
 
 
+def _null_run_id_and_metadata_recursively(decoded_response: Any) -> None:
+    """Recursively traverse the object and delete any keys called run_id"""
+    if isinstance(decoded_response, dict):
+        for key, value in decoded_response.items():
+            if key in {"run_id", "__langserve_version"}:
+                decoded_response[key] = None
+            else:
+                _null_run_id_and_metadata_recursively(value)
+    elif isinstance(decoded_response, list):
+        for item in decoded_response:
+            _null_run_id_and_metadata_recursively(item)
+
+
 @pytest.fixture(scope="module")
 def event_loop():
     """Create an instance of the default event loop for each test case."""
@@ -231,13 +245,17 @@ async def get_async_test_client(
 
 @asynccontextmanager
 async def get_async_remote_runnable(
-    server: FastAPI, *, path: Optional[str] = None, raise_app_exceptions: bool = True
+    server: FastAPI,
+    *,
+    path: Optional[str] = None,
+    raise_app_exceptions: bool = True,
+    **kwargs: Any,
 ) -> RemoteRunnable:
     """Get an async client."""
     url = "http://localhost:9999"
     if path:
         url += path
-    remote_runnable_client = RemoteRunnable(url=url)
+    remote_runnable_client = RemoteRunnable(url=url, **kwargs)
 
     async with get_async_test_client(
         server, path=path, raise_app_exceptions=raise_app_exceptions
@@ -523,6 +541,15 @@ def test_invoke(sync_remote_runnable: RemoteRunnable) -> None:
     assert remote_runnable_run.child_runs[0].name == "add_one_or_passthrough"
 
 
+def test_batch_tracer_with_single_input(sync_remote_runnable: RemoteRunnable) -> None:
+    """Test passing a single tracer to batch."""
+    tracer = FakeTracer()
+    assert sync_remote_runnable.batch([1], config={"callbacks": [tracer]}) == [2]
+    assert len(tracer.runs) == 1
+    assert len(tracer.runs[0].child_runs) == 1
+    assert tracer.runs[0].child_runs[0].name == "add_one_or_passthrough"
+
+
 def test_batch(sync_remote_runnable: RemoteRunnable) -> None:
     """Test sync batch."""
     assert sync_remote_runnable.batch([]) == []
@@ -536,17 +563,9 @@ def test_batch(sync_remote_runnable: RemoteRunnable) -> None:
     tracer = FakeTracer()
     assert sync_remote_runnable.batch([1, 2], config={"callbacks": [tracer]}) == [2, 3]
     assert len(tracer.runs) == 2
-    # Light test to verify that we're picking up information about the server side
-    # function being invoked via a callback.
-    assert tracer.runs[0].child_runs[0].name == "RunnableLambda"
-    assert (
-        tracer.runs[0].child_runs[0].extra["kwargs"]["name"] == "add_one_or_passthrough"
-    )
 
-    assert tracer.runs[1].child_runs[0].name == "RunnableLambda"
-    assert (
-        tracer.runs[1].child_runs[0].extra["kwargs"]["name"] == "add_one_or_passthrough"
-    )
+    assert tracer.runs[0].child_runs[0].name == "add_one_or_passthrough"
+    assert tracer.runs[1].child_runs[0].name == "add_one_or_passthrough"
 
     # Verify that each tracer gets its own run
     tracer1 = FakeTracer()
@@ -558,17 +577,8 @@ def test_batch(sync_remote_runnable: RemoteRunnable) -> None:
     assert len(tracer2.runs) == 1
     # Light test to verify that we're picking up information about the server side
     # function being invoked via a callback.
-    assert tracer1.runs[0].child_runs[0].name == "RunnableLambda"
-    assert (
-        tracer1.runs[0].child_runs[0].extra["kwargs"]["name"]
-        == "add_one_or_passthrough"
-    )
-
-    assert tracer2.runs[0].child_runs[0].name == "RunnableLambda"
-    assert (
-        tracer2.runs[0].child_runs[0].extra["kwargs"]["name"]
-        == "add_one_or_passthrough"
-    )
+    assert tracer1.runs[0].child_runs[0].name == "add_one_or_passthrough"
+    assert tracer2.runs[0].child_runs[0].name == "add_one_or_passthrough"
 
 
 async def test_ainvoke(async_remote_runnable: RemoteRunnable) -> None:
@@ -601,10 +611,7 @@ async def test_ainvoke(async_remote_runnable: RemoteRunnable) -> None:
     elif sys.version_info < (3, 11):
         assert len(tracer.runs) == 1, "Failed for python < 3.11"
         remote_runnable = tracer.runs[0]
-        assert (
-            remote_runnable.child_runs[0].extra["kwargs"]["name"]
-            == "add_one_or_passthrough"
-        )
+        assert remote_runnable.name == "RemoteRunnable"
     else:
         raise AssertionError(f"Unsupported python version {sys.version_info}")
 
@@ -624,17 +631,9 @@ async def test_abatch(async_remote_runnable: RemoteRunnable) -> None:
         [1, 2], config={"callbacks": [tracer]}
     ) == [2, 3]
     assert len(tracer.runs) == 2
-    # Light test to verify that we're picking up information about the server side
-    # function being invoked via a callback.
-    assert tracer.runs[0].child_runs[0].name == "RunnableLambda"
-    assert (
-        tracer.runs[0].child_runs[0].extra["kwargs"]["name"] == "add_one_or_passthrough"
-    )
 
-    assert tracer.runs[1].child_runs[0].name == "RunnableLambda"
-    assert (
-        tracer.runs[1].child_runs[0].extra["kwargs"]["name"] == "add_one_or_passthrough"
-    )
+    assert tracer.runs[0].child_runs[0].name == "add_one_or_passthrough"
+    assert tracer.runs[1].child_runs[0].name == "add_one_or_passthrough"
 
     # Verify that each tracer gets its own run
     tracer1 = FakeTracer()
@@ -644,19 +643,9 @@ async def test_abatch(async_remote_runnable: RemoteRunnable) -> None:
     ) == [2, 3]
     assert len(tracer1.runs) == 1
     assert len(tracer2.runs) == 1
-    # Light test to verify that we're picking up information about the server side
-    # function being invoked via a callback.
-    assert tracer1.runs[0].child_runs[0].name == "RunnableLambda"
-    assert (
-        tracer1.runs[0].child_runs[0].extra["kwargs"]["name"]
-        == "add_one_or_passthrough"
-    )
 
-    assert tracer2.runs[0].child_runs[0].name == "RunnableLambda"
-    assert (
-        tracer2.runs[0].child_runs[0].extra["kwargs"]["name"]
-        == "add_one_or_passthrough"
-    )
+    assert tracer1.runs[0].child_runs[0].name == "add_one_or_passthrough"
+    assert tracer2.runs[0].child_runs[0].name == "add_one_or_passthrough"
 
 
 async def test_astream(async_remote_runnable: RemoteRunnable) -> None:
@@ -1128,6 +1117,156 @@ async def test_config_keys_validation(mocker: MockerFixture) -> None:
         assert config_seen["metadata"]["__langserve_endpoint"] == "invoke"
 
 
+async def test_include_callback_events(mocker: MockerFixture) -> None:
+    """This test should not use a RemoteRunnable.
+
+    Check if callback events are being sent back from the server.
+
+    Do so using the raw client.
+    """
+
+    async def add_one(x: int) -> int:
+        """Add one to simulate a valid function"""
+        return x + 1
+
+    server_runnable = RunnableLambda(func=add_one)
+
+    app = FastAPI()
+    add_routes(app, server_runnable, input_type=int, include_callback_events=True)
+    async with AsyncClient(
+        base_url="http://localhost:9999", transport=httpx.ASGITransport(app=app)
+    ) as async_client:
+        response = await async_client.post("/invoke", json={"input": 1})
+        # Config should be ignored but default debug information
+        # will still be added
+        assert response.status_code == 200
+        decoded_response = response.json()
+        # Remove any run_id from the response recursively
+        _null_run_id_and_metadata_recursively(decoded_response)
+        assert decoded_response == {
+            "callback_events": [
+                {
+                    "inputs": 1,
+                    "kwargs": {"name": "add_one", "run_type": None},
+                    "metadata": {
+                        "__langserve_endpoint": "invoke",
+                        "__langserve_version": None,
+                        "__useragent": "python-httpx/0.27.2",
+                    },
+                    "parent_run_id": None,
+                    "serialized": None,
+                    "run_id": None,
+                    "tags": [],
+                    "type": "on_chain_start",
+                },
+                {
+                    "kwargs": {},
+                    "outputs": 2,
+                    "metadata": None,
+                    "parent_run_id": None,
+                    "tags": [],
+                    "run_id": None,
+                    "type": "on_chain_end",
+                },
+            ],
+            "metadata": {
+                "feedback_tokens": [],
+                "run_id": None,
+            },
+            "output": 2,
+        }
+
+
+async def test_include_callback_events_batch() -> None:
+    """This test should not use a RemoteRunnable.
+
+    Check if callback events are being sent back from the server.
+
+    Do so using the raw client.
+    """
+
+    async def add_one(x: int) -> int:
+        """Add one to simulate a valid function"""
+        return x + 1
+
+    server_runnable = RunnableLambda(func=add_one)
+
+    app = FastAPI()
+    add_routes(app, server_runnable, input_type=int, include_callback_events=True)
+    async with AsyncClient(
+        base_url="http://localhost:9999", transport=httpx.ASGITransport(app=app)
+    ) as async_client:
+        response = await async_client.post("/batch", json={"inputs": [1, 2]})
+        # Config should be ignored but default debug information
+        # will still be added
+        assert response.status_code == 200
+        decoded_response = response.json()
+        # Remove any run_id from the response recursively
+        _null_run_id_and_metadata_recursively(decoded_response)
+        del decoded_response["metadata"]["run_ids"]
+        assert decoded_response == {
+            "callback_events": [
+                [
+                    {
+                        "inputs": 1,
+                        "kwargs": {"name": "add_one", "run_type": None},
+                        "metadata": {
+                            "__langserve_endpoint": "batch",
+                            "__langserve_version": None,
+                            "__useragent": "python-httpx/0.27.2",
+                        },
+                        "parent_run_id": None,
+                        "run_id": None,
+                        "serialized": None,
+                        "tags": [],
+                        "type": "on_chain_start",
+                    },
+                    {
+                        "kwargs": {},
+                        "outputs": 2,
+                        "parent_run_id": None,
+                        "metadata": None,
+                        "run_id": None,
+                        "tags": [],
+                        "type": "on_chain_end",
+                    },
+                ],
+                [
+                    {
+                        "inputs": 2,
+                        "kwargs": {"name": "add_one", "run_type": None},
+                        "metadata": {
+                            "__langserve_endpoint": "batch",
+                            "__langserve_version": None,
+                            "__useragent": "python-httpx/0.27.2",
+                        },
+                        "parent_run_id": None,
+                        "run_id": None,
+                        "serialized": None,
+                        "tags": [],
+                        "type": "on_chain_start",
+                    },
+                    {
+                        "kwargs": {},
+                        "outputs": 3,
+                        "parent_run_id": None,
+                        "metadata": None,
+                        "run_id": None,
+                        "tags": [],
+                        "type": "on_chain_end",
+                    },
+                ],
+            ],
+            "metadata": {
+                "responses": [
+                    {"feedback_tokens": [], "run_id": None},
+                    {"feedback_tokens": [], "run_id": None},
+                ],
+            },
+            "output": [2, 3],
+        }
+
+
 async def test_input_validation(mocker: MockerFixture) -> None:
     """Test client side and server side exceptions."""
 
@@ -2146,6 +2285,49 @@ async def test_uuid_serialization() -> None:
         )
 
 
+async def test_custom_serialization() -> None:
+    """Test updating the config based on the raw request object."""
+    from langserve.serialization import Serializer
+
+    class CustomObject:
+        def __init__(self, x: int) -> None:
+            self.x = x
+
+        def __eq__(self, other) -> bool:
+            return self.x == other.x
+
+    class CustomSerializer(Serializer):
+        def dumps(self, obj: Any) -> bytes:
+            """Dump the given object as a JSON string."""
+            if isinstance(obj, CustomObject):
+                return orjson.dumps({"x": obj.x})
+            else:
+                return orjson.dumps(obj)
+
+        def loadd(self, obj: Any) -> Any:
+            """Load the given object."""
+            if isinstance(obj, bytes):
+                obj = obj.decode("utf-8")
+            if obj.get("x"):
+                return CustomObject(x=obj["x"])
+            return obj
+
+    def foo(x: int) -> Any:
+        """Add one to simulate a valid function."""
+        return CustomObject(x=5)
+
+    app = FastAPI()
+    server_runnable = RunnableLambda(foo)
+    add_routes(app, server_runnable, serializer=CustomSerializer())
+
+    async with get_async_remote_runnable(
+        app, raise_app_exceptions=True, serializer=CustomSerializer()
+    ) as runnable:
+        result = await runnable.ainvoke(5)
+        assert isinstance(result, CustomObject)
+        assert result == CustomObject(x=5)
+
+
 async def test_endpoint_configurations() -> None:
     """Test enabling/disabling endpoints."""
     app = FastAPI()
@@ -2306,7 +2488,7 @@ async def test_astream_events_simple(async_remote_runnable: RemoteRunnable) -> N
         # test client side error
         with pytest.raises(httpx.HTTPStatusError) as cb:
             # Invalid input type (expected string but got int)
-            async for _ in runnable.astream_events("foo", version="v1"):
+            async for _ in runnable.astream_events("foo", version="v2"):
                 pass
 
         # Verify that this is a 422 error
@@ -2315,7 +2497,7 @@ async def test_astream_events_simple(async_remote_runnable: RemoteRunnable) -> N
         with pytest.raises(httpx.HTTPStatusError) as cb:
             # Invalid input type (expected string but got int)
             # include names should not be a list of lists
-            async for _ in runnable.astream_events(1, include_names=[[]], version="v1"):
+            async for _ in runnable.astream_events(1, include_names=[[]], version="v2"):
                 pass
 
         # Verify that this is a 422 error
@@ -2324,7 +2506,7 @@ async def test_astream_events_simple(async_remote_runnable: RemoteRunnable) -> N
         # Test good requests
         events = []
 
-        async for event in runnable.astream_events(1, version="v1"):
+        async for event in runnable.astream_events(1, version="v2"):
             events.append(event)
 
         # validate events
@@ -2337,6 +2519,7 @@ async def test_astream_events_simple(async_remote_runnable: RemoteRunnable) -> N
                 assert not k.startswith("__")
             assert "metadata" in event
             del event["metadata"]
+            event["parent_ids"] = []
 
         assert events == [
             {
@@ -2416,6 +2599,7 @@ def _clean_up_events(events: List[Dict[str, Any]]) -> None:
             assert not k.startswith("__")
         assert "metadata" in event
         del event["metadata"]
+        event["parent_ids"] = []
 
 
 async def test_astream_events_with_serialization(
@@ -2488,7 +2672,7 @@ async def test_astream_events_with_serialization(
         app, raise_app_exceptions=False, path="/doc_types"
     ) as runnable:
         # Test good requests
-        events = [event async for event in runnable.astream_events("foo", version="v1")]
+        events = [event async for event in runnable.astream_events("foo", version="v2")]
         _clean_up_events(events)
 
         assert events == [
@@ -2578,7 +2762,7 @@ async def test_astream_events_with_serialization(
         app, raise_app_exceptions=False, path="/get_pets"
     ) as runnable:
         # Test good requests
-        events = [event async for event in runnable.astream_events("foo", version="v1")]
+        events = [event async for event in runnable.astream_events("foo", version="v2")]
         _clean_up_events(events)
         assert events == [
             {
@@ -2613,7 +2797,7 @@ async def test_astream_events_with_serialization(
     ) as runnable:
         # Test good requests
         with pytest.raises(httpx.HTTPStatusError) as cb:
-            async for event in runnable.astream_events("foo", version="v1"):
+            async for event in runnable.astream_events("foo", version="v2"):
                 pass
         assert cb.value.response.status_code == 500
 
@@ -2641,7 +2825,7 @@ async def test_astream_events_with_prompt_model_parser_chain(
         events = [
             event
             async for event in runnable.astream_events(
-                {"question": "hello"}, version="v1"
+                {"question": "hello"}, version="v2"
             )
         ]
         _clean_up_events(events)
@@ -2850,25 +3034,16 @@ async def test_astream_events_with_prompt_model_parser_chain(
                         ]
                     },
                     "output": {
-                        "generations": [
-                            [
-                                {
-                                    "generation_info": None,
-                                    "message": {
-                                        "additional_kwargs": {},
-                                        "content": "Hello World!",
-                                        "name": None,
-                                        "response_metadata": {},
-                                        "type": "AIMessageChunk",
-                                    },
-                                    "text": "Hello World!",
-                                    "type": "ChatGenerationChunk",
-                                }
-                            ]
-                        ],
-                        "llm_output": None,
-                        "run": None,
-                        "type": "LLMResult",
+                        "additional_kwargs": {},
+                        "content": "Hello World!",
+                        "example": False,
+                        "invalid_tool_calls": [],
+                        "name": None,
+                        "response_metadata": {},
+                        "tool_call_chunks": [],
+                        "tool_calls": [],
+                        "type": "AIMessageChunk",
+                        "usage_metadata": None,
                     },
                 },
                 "event": "on_chat_model_end",

tests/unit_tests/utils/tracer.py

@@ -1,4 +1,4 @@
-from typing import Dict, List
+from typing import Any, Dict, List, Optional
 from uuid import UUID
 
 from langchain_core.tracers import BaseTracer
@@ -39,6 +39,34 @@ class FakeTracer(BaseTracer):
             }
         )
 
+    def _create_chain_run(
+        self,
+        serialized: Dict[str, Any],
+        inputs: Dict[str, Any],
+        run_id: UUID,
+        tags: Optional[List[str]] = None,
+        parent_run_id: Optional[UUID] = None,
+        metadata: Optional[Dict[str, Any]] = None,
+        run_type: Optional[str] = None,
+        name: Optional[str] = None,
+        **kwargs: Any,
+    ) -> Run:
+        if name is None:
+            # can't raise an exception from here, but can get a breakpoint
+            # import pdb; pdb.set_trace()
+            pass
+        return super()._create_chain_run(
+            serialized,
+            inputs,
+            run_id,
+            tags,
+            parent_run_id,
+            metadata,
+            run_type,
+            name,
+            **kwargs,
+        )
+
     def _persist_run(self, run: Run) -> None:
         """Persist a run."""
         self.runs.append(self._copy_run(run))