xx

langserve/api_handler.py

@@ -534,6 +534,7 @@ def __init__(
         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",
     ) -> None:
         """Create an API handler for the given runnable.
 
@@ -595,6 +596,8 @@ def __init__(
                 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".
         """
         if importlib.util.find_spec("sse_starlette") is None:
             raise ImportError(
@@ -632,6 +635,7 @@ def __init__(
         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:
@@ -1343,7 +1347,7 @@ async def _stream_events() -> AsyncIterator[dict]:
                     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/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 = {
@@ -752,7 +758,7 @@ async def astream_events(
         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,21 +781,27 @@ async def astream_events(
             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
             exclude_names: The names of the events to exclude
             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/server.py

@@ -262,6 +262,7 @@ 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",
 ) -> None:
     """Register the routes on the given FastAPI app or APIRouter.
 
@@ -380,14 +381,16 @@ 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".
     """  # 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(