path bugfix

Co-authored-by: Eugene Yurtsev <eyurtsev@gmail.com>

README.md

@@ -95,6 +95,14 @@ If you've deployed the server above, you can view the generated OpenAPI docs usi
 curl localhost:8000/docs
 ```
 
+make sure to **add** the `/docs` suffix. 
+
+Below will return a 404 until you define a `@app.get("/")`
+
+```sh
+localhost:8000
+```
+
 ### Client
 
 Python SDK

examples/llm/client.ipynb

@@ -229,7 +229,7 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "from langchain.schema.runnable import PutLocalVar, GetLocalVar"
+    "from langchain.schema.runnable import RunnablePassthrough"
    ]
   },
   {
@@ -266,13 +266,8 @@
     ")\n",
     "\n",
     "\n",
-    "chain = (\n",
-    "    comedian_chain\n",
-    "    | PutLocalVar(\"joke\")\n",
-    "    | {\"joke\": GetLocalVar(\"joke\")}\n",
-    "    | joke_classifier_chain\n",
-    "    | PutLocalVar(\"classification\")\n",
-    "    | {\"joke\": GetLocalVar(\"joke\"), \"classification\": GetLocalVar(\"classification\")}\n",
+    "chain = {\"joke\": comedian_chain} | RunnablePassthrough.assign(\n",
+    "    classification=joke_classifier_chain\n",
     ")"
    ]
   },

langserve/client.py

@@ -1,8 +1,11 @@
 from __future__ import annotations
 
 import asyncio
+import json
+import logging
 import weakref
 from concurrent.futures import ThreadPoolExecutor
+from functools import lru_cache
 from typing import (
     Any,
     AsyncIterator,
@@ -30,6 +33,8 @@ from langchain.schema.runnable.utils import Input, Output
 
 from langserve.serialization import simple_dumpd, simple_loads
 
+logger = logging.getLogger(__name__)
+
 
 def _without_callbacks(config: Optional[RunnableConfig]) -> RunnableConfig:
     """Evict callbacks from the config since those are definitely not supported."""
@@ -37,6 +42,12 @@ def _without_callbacks(config: Optional[RunnableConfig]) -> RunnableConfig:
     return {k: v for k, v in _config.items() if k != "callbacks"}
 
 
+@lru_cache(maxsize=1_000)  # Will accommodate up to 100 different error messages
+def _log_error_message_once(error_message: str) -> None:
+    """Log an error message once."""
+    logger.error(error_message)
+
+
 def _raise_for_status(response: httpx.Response) -> None:
     """Re-raise with a more informative message.
 
@@ -94,6 +105,26 @@ def _close_clients(sync_client: httpx.Client, async_client: httpx.AsyncClient) -
         asyncio.run(async_client.aclose())
 
 
+def _raise_exception_from_data(data: str, request: httpx.Request) -> None:
+    """Raise an httpx exception from the given error event data."""
+    try:
+        decoded_data = json.loads(data)
+    except json.JSONDecodeError:
+        raise httpx.HTTPStatusError(
+            message="invalid json in error event sent from server",
+            request=request,
+            response=httpx.Response(status_code=500, text=data),
+        )
+    raise httpx.HTTPStatusError(
+        message=decoded_data["message"],
+        request=request,
+        response=httpx.Response(
+            status_code=decoded_data["status_code"],
+            text=decoded_data["message"],
+        ),
+    )
+
+
 class RemoteRunnable(Runnable[Input, Output]):
     """A RemoteRunnable is a runnable that is executed on a remote server.
 
@@ -334,10 +365,19 @@ class RemoteRunnable(Runnable[Input, Output]):
                             final_output += chunk
                         else:
                             final_output = chunk
+                    elif sse.event == "error":
+                        # This can only be a server side error
+                        _raise_exception_from_data(
+                            sse.data, httpx.Request(method="POST", url=endpoint)
+                        )
                     elif sse.event == "end":
                         break
                     else:
-                        raise NotImplementedError(f"Unknown event {sse.event}")
+                        logger.error(
+                            f"Encountered an unsupported event type: `{sse.event}`. "
+                            f"Try upgrading the remote client to the latest version."
+                            f"Ignoring events of type `{sse.event}`."
+                        )
         except BaseException as e:
             run_manager.on_chain_error(e)
             raise
@@ -385,11 +425,20 @@ class RemoteRunnable(Runnable[Input, Output]):
                             final_output += chunk
                         else:
                             final_output = chunk
+
+                    elif sse.event == "error":
+                        # This can only be a server side error
+                        _raise_exception_from_data(
+                            sse.data, httpx.Request(method="POST", url=endpoint)
+                        )
                     elif sse.event == "end":
                         break
-
                     else:
-                        raise NotImplementedError(f"Unknown event {sse.event}")
+                        logger.error(
+                            f"Encountered an unsupported event type: `{sse.event}`. "
+                            f"Try upgrading the remote client to the latest version."
+                            f"Ignoring events of type `{sse.event}`."
+                        )
         except BaseException as e:
             await run_manager.on_chain_error(e)
             raise
@@ -460,15 +509,23 @@ class RemoteRunnable(Runnable[Input, Output]):
                         chunk = RunLogPatch(*data["ops"])
 
                         yield chunk
-
                         if final_output:
                             final_output += chunk
                         else:
                             final_output = chunk
+                    elif sse.event == "error":
+                        # This can only be a server side error
+                        _raise_exception_from_data(
+                            sse.data, httpx.Request(method="POST", url=endpoint)
+                        )
                     elif sse.event == "end":
                         break
                     else:
-                        raise NotImplementedError(f"Unknown event {sse.event}")
+                        _log_error_message_once(
+                            f"Encountered an unsupported event type: `{sse.event}`. "
+                            f"Try upgrading the remote client to the latest version."
+                            f"Ignoring events of type `{sse.event}`."
+                        )
         except BaseException as e:
             await run_manager.on_chain_error(e)
             raise

langserve/packages.py

@@ -0,0 +1,104 @@
+import importlib
+import logging
+from pathlib import Path
+from typing import Generator, TypedDict, Union
+
+from fastapi import APIRouter, FastAPI
+from tomli import load
+
+from langserve.server import add_routes
+
+
+class LangServeExport(TypedDict):
+    """
+    Fields from pyproject.toml that are relevant to LangServe
+
+    Attributes:
+        module: The module to import from, tool.langserve.export_module
+        attr: The attribute to import from the module, tool.langserve.export_attr
+        package_name: The name of the package, tool.poetry.name
+    """
+
+    module: str
+    attr: str
+    package_name: str
+
+
+def get_langserve_export(filepath: Path) -> LangServeExport:
+    with open(filepath, "rb") as f:
+        data = load(f)
+    try:
+        module = data["tool"]["langserve"]["export_module"]
+        attr = data["tool"]["langserve"]["export_attr"]
+        package_name = data["tool"]["poetry"]["name"]
+    except KeyError as e:
+        raise KeyError("Invalid LangServe PyProject.toml") from e
+    return LangServeExport(module=module, attr=attr, package_name=package_name)
+
+
+EXCLUDE_PATHS = set(["__pycache__", ".venv", ".git", ".github"])
+
+
+def _include_path(path: Path) -> bool:
+    """
+    Skip paths that are in EXCLUDE_PATHS or start with an underscore.
+    """
+    for part in path.parts:
+        if part in EXCLUDE_PATHS:
+            return False
+        if part.startswith("_"):
+            return False
+    return True
+
+
+def list_packages(path: str = "packages") -> Generator[Path, None, None]:
+    """
+    Yields Path objects for each folder that contains a pyproject.toml file within a
+    path. Use this to find packages to add to the server.
+
+    See `add_package_routes` below for an example of how to use this.
+    """
+    # traverse path for routes to host (any directory holding a pyproject.toml file)
+    package_root = Path(path)
+    for pyproject_path in package_root.glob("**/pyproject.toml"):
+        if not _include_path(pyproject_path):
+            continue
+        yield pyproject_path.parent
+
+
+def add_package_route(
+    app: Union[FastAPI, APIRouter], package_path: Path, mount_path: str
+) -> None:
+    try:
+        pyproject_path = package_path / "pyproject.toml"
+
+        # get langserve export
+        package = get_langserve_export(pyproject_path)
+        package_name = package["package_name"]
+        # import module
+        mod = importlib.import_module(package["module"])
+    except KeyError:
+        logging.warning(
+            f"Skipping {package_path} because it is not a valid LangServe "
+            "package (see pyproject.toml)"
+        )
+        return
+    except ImportError as e:
+        logging.warning(f"Error: {e}")
+        logging.warning(f"Try fixing with `poetry add --editable {package_path}`")
+        logging.warning(
+            "To remove packages, use `poe` instead of `poetry`: "
+            f"`poe remove {package_name}`"
+        )
+        return
+    # get attr
+    chain = getattr(mod, package["attr"])
+    add_routes(app, chain, path=mount_path)
+
+
+def add_package_routes(app: Union[FastAPI, APIRouter], path: str = "packages") -> None:
+    # traverse path for routes to host (any directory holding a pyproject.toml file)
+    for package_path in list_packages(path):
+        mount_path_relative = package_path.relative_to(Path(path))
+        mount_path = f"/{mount_path_relative}"
+        add_package_route(app, package_path, mount_path)

langserve/server.py

@@ -382,6 +382,16 @@ def add_routes(
                 }
             }
 
+        * error - for signaling an error in the stream, also ends the stream.
+
+        {
+            "event": "error",
+            "data": {
+                "status_code": 500,
+                "message": "Internal Server Error"
+            }
+        }
+
         * end - for signaling the end of the stream.
 
             This helps the client to know when to stop listening for events and
@@ -402,12 +412,24 @@ def add_routes(
 
         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"}
+            try:
+                async for chunk in runnable.astream(
+                    input_,
+                    config=config,
+                ):
+                    yield {"data": simple_dumps(chunk), "event": "data"}
+                yield {"event": "end"}
+            except BaseException:
+                yield {
+                    "event": "error",
+                    # Do not expose the error message to the client since
+                    # the message may contain sensitive information.
+                    # We'll add client side errors for validation as well.
+                    "data": json.dumps(
+                        {"status_code": 500, "message": "Internal Server Error"}
+                    ),
+                }
+                raise
 
         return EventSourceResponse(_stream())
 
@@ -441,6 +463,16 @@ def add_routes(
                 }
             }
 
+        * error - for signaling an error in the stream, also ends the stream.
+
+        {
+            "event": "error",
+            "data": {
+                "status_code": 500,
+                "message": "Internal Server Error"
+            }
+        }
+
         * end - for signaling the end of the stream.
 
             This helps the client to know when to stop listening for events and
@@ -464,31 +496,43 @@ def add_routes(
 
         async def _stream_log() -> AsyncIterator[dict]:
             """Stream the output of the runnable."""
-            async for chunk in runnable.astream_log(
-                input_,
-                config=config,
-                diff=True,
-                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 not isinstance(chunk, RunLogPatch):
-                    raise AssertionError(
-                        f"Expected a RunLog instance got {type(chunk)}"
-                    )
-                data = {
-                    "ops": chunk.ops,
-                }
+            try:
+                async for chunk in runnable.astream_log(
+                    input_,
+                    config=config,
+                    diff=True,
+                    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 not isinstance(chunk, RunLogPatch):
+                        raise AssertionError(
+                            f"Expected a RunLog instance got {type(chunk)}"
+                        )
+                    data = {
+                        "ops": chunk.ops,
+                    }
 
-                # Temporary adapter
+                    # Temporary adapter
+                    yield {
+                        "data": simple_dumps(data),
+                        "event": "data",
+                    }
+                yield {"event": "end"}
+            except BaseException:
                 yield {
-                    "data": simple_dumps(data),
-                    "event": "data",
+                    "event": "error",
+                    # Do not expose the error message to the client since
+                    # the message may contain sensitive information.
+                    # We'll add client side errors for validation as well.
+                    "data": json.dumps(
+                        {"status_code": 500, "message": "Internal Server Error"}
+                    ),
                 }
-            yield {"event": "end"}
+                raise
 
         return EventSourceResponse(_stream_log())
 
@@ -508,9 +552,13 @@ def add_routes(
     @app.get(f"{namespace}/output_schema")
     async def output_schema(config_hash: str = "") -> Any:
         """Return the output schema of the runnable."""
-        return runnable.with_config(
-            _unpack_config(config_hash, keys=config_keys, model=ConfigPayload)
-        ).output_schema.schema()
+        return (
+            runnable.with_config(
+                _unpack_config(config_hash, keys=config_keys, model=ConfigPayload)
+            ).output_schema.schema()
+            if output_type_ == "auto"
+            else output_type_.schema()
+        )
 
     @app.get(namespace + "/c/{config_hash}/config_schema", tags=["config"])
     @app.get(f"{namespace}/config_schema")

tests/unit_tests/test_server_client.py

@@ -2,8 +2,8 @@
 import asyncio
 import json
 from asyncio import AbstractEventLoop
-from contextlib import asynccontextmanager
-from typing import Any, Dict, List, Optional, Union
+from contextlib import asynccontextmanager, contextmanager
+from typing import Any, Dict, Iterator, List, Optional, Sequence, Union
 
 import httpx
 import pytest
@@ -15,12 +15,13 @@ from langchain.callbacks.tracers.log_stream import RunLogPatch
 from langchain.prompts import PromptTemplate
 from langchain.prompts.base import StringPromptValue
 from langchain.schema.messages import HumanMessage, SystemMessage
-from langchain.schema.runnable import RunnableConfig, RunnablePassthrough
+from langchain.schema.runnable import Runnable, RunnableConfig, RunnablePassthrough
 from langchain.schema.runnable.base import RunnableLambda
-from langchain.schema.runnable.utils import ConfigurableField
+from langchain.schema.runnable.utils import ConfigurableField, Input, Output
 from pytest_mock import MockerFixture
 from typing_extensions import TypedDict
 
+from langserve import server
 from langserve.client import RemoteRunnable
 from langserve.lzstring import LZString
 from langserve.server import (
@@ -103,14 +104,19 @@ def client(app: FastAPI) -> RemoteRunnable:
 
 @asynccontextmanager
 async def get_async_client(
-    server: FastAPI, path: Optional[str] = None
+    server: FastAPI, *, path: Optional[str] = None, raise_app_exceptions: bool = True
 ) -> 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)
+
+    transport = httpx.ASGITransport(
+        app=server,
+        raise_app_exceptions=raise_app_exceptions,
+    )
+    async_client = AsyncClient(app=server, base_url=url, transport=transport)
     remote_runnable_client.async_client = async_client
     try:
         yield remote_runnable_client
@@ -118,6 +124,25 @@ async def get_async_client(
         await async_client.aclose()
 
 
+@contextmanager
+def get_sync_client(
+    server: FastAPI, *, path: Optional[str] = None, raise_server_exceptions: bool = True
+) -> RemoteRunnable:
+    """Get an async client."""
+    url = "http://localhost:9999"
+    if path:
+        url += path
+    remote_runnable_client = RemoteRunnable(url=url)
+    sync_client = TestClient(
+        app=server, base_url=url, raise_server_exceptions=raise_server_exceptions
+    )
+    remote_runnable_client.sync_client = sync_client
+    try:
+        yield remote_runnable_client
+    finally:
+        sync_client.close()
+
+
 @pytest_asyncio.fixture()
 async def async_client(app: FastAPI) -> RemoteRunnable:
     """Create a FastAPI app that exposes the Runnable as an API."""
@@ -749,6 +774,132 @@ def test_rename_pydantic_model() -> None:
     assert Model.__name__ == "Bar"
 
 
+@pytest.mark.asyncio
+async def test_input_config_output_schemas(event_loop: AbstractEventLoop) -> None:
+    """Test schemas returned for different configurations."""
+    # TODO(Fix me): need to fix handling of global state -- we get problems
+    # gives inconsistent results when running multiple tests / results
+    # depending on ordering
+    server._SEEN_NAMES = set()
+    server._MODEL_REGISTRY = {}
+
+    async def add_one(x: int) -> int:
+        """Add one to simulate a valid function"""
+        return x + 1
+
+    async def add_two(y: int) -> int:
+        """Add one to simulate a valid function"""
+        return y + 2
+
+    app = FastAPI()
+
+    add_routes(app, RunnableLambda(add_one), path="/add_one")
+    # Custom input type
+    add_routes(
+        app,
+        RunnableLambda(add_two),
+        path="/add_two_custom",
+        input_type=float,
+        output_type=Sequence[float],
+        config_keys=["tags", "configurable"],
+    )
+    add_routes(app, PromptTemplate.from_template("{question}"), path="/prompt_1")
+
+    template = PromptTemplate.from_template("say {name}").configurable_fields(
+        template=ConfigurableField(
+            id="template",
+            name="Template",
+            description="The template to use for the prompt",
+        )
+    )
+    add_routes(app, template, path="/prompt_2", config_keys=["tags", "configurable"])
+
+    async with AsyncClient(app=app, base_url="http://localhost:9999") as async_client:
+        # input schema
+        response = await async_client.get("/add_one/input_schema")
+        assert response.json() == {"title": "RunnableLambdaInput", "type": "integer"}
+
+        response = await async_client.get("/add_two_custom/input_schema")
+        assert response.json() == {"title": "Input", "type": "number"}
+
+        response = await async_client.get("/prompt_1/input_schema")
+        assert response.json() == {
+            "properties": {"question": {"title": "Question", "type": "string"}},
+            "title": "PromptInput",
+            "type": "object",
+        }
+
+        response = await async_client.get("/prompt_2/input_schema")
+        assert response.json() == {
+            "properties": {"name": {"title": "Name", "type": "string"}},
+            "title": "PromptInput",
+            "type": "object",
+        }
+
+        # output schema
+        response = await async_client.get("/add_one/output_schema")
+        assert response.json() == {
+            "title": "RunnableLambdaOutput",
+            "type": "integer",
+        }
+
+        response = await async_client.get("/add_two_custom/output_schema")
+        assert response.json() == {
+            "items": {"type": "number"},
+            "title": "Output",
+            "type": "array",
+        }
+
+        # Just verify that the schema is not empty (it's pretty long)
+        # and the actual value should be tested in LangChain
+        response = await async_client.get("/prompt_1/output_schema")
+        assert response.json() != {}  # Long string
+
+        response = await async_client.get("/prompt_2/output_schema")
+        assert response.json() != {}  # Long string
+
+        ## Config schema
+        response = await async_client.get("/add_one/config_schema")
+        assert response.json() == {
+            "properties": {},
+            "title": "RunnableLambdaConfig",
+            "type": "object",
+        }
+
+        response = await async_client.get("/add_two_custom/config_schema")
+        assert response.json() == {
+            "properties": {
+                "tags": {"items": {"type": "string"}, "title": "Tags", "type": "array"}
+            },
+            "title": "RunnableLambdaConfig",
+            "type": "object",
+        }
+
+        response = await async_client.get("/prompt_2/config_schema")
+        assert response.json() == {
+            "definitions": {
+                "Configurable": {
+                    "properties": {
+                        "template": {
+                            "default": "say {name}",
+                            "description": "The template to use for the prompt",
+                            "title": "Template",
+                            "type": "string",
+                        }
+                    },
+                    "title": "Configurable",
+                    "type": "object",
+                }
+            },
+            "properties": {
+                "configurable": {"$ref": "#/definitions/Configurable"},
+                "tags": {"items": {"type": "string"}, "title": "Tags", "type": "array"},
+            },
+            "title": "RunnableConfigurableFieldsConfig",
+            "type": "object",
+        }
+
+
 @pytest.mark.asyncio
 async def test_input_schema_typed_dict() -> None:
     class InputType(TypedDict):
@@ -783,3 +934,97 @@ async def test_input_schema_typed_dict() -> None:
                 }
             },
         }
+
+
+class ErroringRunnable(Runnable):
+    """A custom runnable for testing errors are raised server side."""
+
+    def invoke(self, input: Input, config: Optional[RunnableConfig] = None) -> Output:
+        """Invoke the runnable."""
+        raise ValueError("Server side error")
+
+    def stream(
+        self,
+        input: Input,
+        config: Optional[RunnableConfig] = None,
+        **kwargs: Optional[Any],
+    ) -> Iterator[Output]:
+        yield 1
+        yield 2
+        raise ValueError("An exception occurred")
+
+    async def astream(
+        self,
+        input: Iterator[Input],
+        config: Optional[RunnableConfig] = None,
+        **kwargs: Optional[Any],
+    ) -> Iterator[Output]:
+        yield 1
+        yield 2
+        raise ValueError("An exception occurred")
+
+
+@pytest.mark.asyncio
+async def test_server_side_error() -> None:
+    """Test server side error handling."""
+
+    app = FastAPI()
+    add_routes(app, ErroringRunnable())
+
+    # Invoke request
+    async with get_async_client(app, raise_app_exceptions=False) as runnable:
+        with pytest.raises(httpx.HTTPStatusError) as cm:
+            assert await runnable.ainvoke(1)
+        assert isinstance(cm.value, httpx.HTTPStatusError)
+
+        with pytest.raises(httpx.HTTPStatusError) as cm:
+            assert await runnable.abatch([1, 2])
+        assert isinstance(cm.value, httpx.HTTPStatusError)
+
+        # Test astream
+        chunks = []
+        try:
+            async for chunk in runnable.astream({"a": 1}):
+                chunks.append(chunk)
+        except httpx.HTTPStatusError as e:
+            assert chunks == [1, 2]
+            assert e.response.status_code == 500
+            assert e.response.text == "Internal Server Error"
+
+        # # Failing right now, can uncomment or add callbacks
+        # # Test astream_log
+        # chunks = []
+        # try:
+        #     async for chunk in runnable.astream_log({"a": 1}):
+        #         chunks.append(chunk)
+        # except httpx.HTTPStatusError as e:
+        #     assert chunks == []
+        #     assert e.response.status_code == 500
+        #     assert e.response.text == "Internal Server Error"
+
+
+def test_server_side_error_sync() -> None:
+    """Test server side error handling."""
+
+    app = FastAPI()
+    add_routes(app, ErroringRunnable())
+
+    # Invoke request
+    with get_sync_client(app, raise_server_exceptions=False) as runnable:
+        with pytest.raises(httpx.HTTPStatusError) as cm:
+            assert runnable.invoke(1)
+        assert isinstance(cm.value, httpx.HTTPStatusError)
+
+        with pytest.raises(httpx.HTTPStatusError) as cm:
+            assert runnable.batch([1, 2])
+        assert isinstance(cm.value, httpx.HTTPStatusError)
+
+        # Test astream
+        chunks = []
+        try:
+            for chunk in runnable.stream({"a": 1}):
+                chunks.append(chunk)
+        except httpx.HTTPStatusError as e:
+            assert chunks == [1, 2]
+            assert e.response.status_code == 500
+            assert e.response.text == "Internal Server Error"