path bugfix

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

README.md

@@ -1,4 +1,4 @@
-# LangServe 🦜️🔗 
+# LangServe 🦜️🔗
 
 ## Overview
 
@@ -15,6 +15,7 @@ A javascript client is available in [LangChainJS](https://js.langchain.com/docs/
 - API docs page with JSONSchema and Swagger (insert example link)
 - Efficient `/invoke`, `/batch` and `/stream` endpoints with support for many concurrent requests on a single server
 - `/stream_log` endpoint for streaming all (or some) intermediate steps from your chain/agent
+- Playground page at `/playground` with streaming output and intermediate steps
 - Built-in (optional) tracing to [LangSmith](https://www.langchain.com/langsmith), just add your API key (see [Instructions](https://docs.smith.langchain.com/)])
 - All built with battle-tested open-source Python libraries like FastAPI, Pydantic, uvloop and asyncio.
 - Use the client SDK to call a LangServe server as if it was a Runnable running locally (or call the HTTP API directly)
@@ -24,7 +25,7 @@ A javascript client is available in [LangChainJS](https://js.langchain.com/docs/
 - Client callbacks are not yet supported for events that originate on the server
 - Does not work with [pydantic v2 yet](https://github.com/tiangolo/fastapi/issues/10360)
 
-## LangChain CLI  🛠️
+## LangChain CLI 🛠️
 
 Use the `LangChain` CLI to bootstrap a `LangServe` project quickly.
 
@@ -41,7 +42,6 @@ And follow the instructions...
 
 For more examples, see the [examples](./examples) directory.
 
-
 ### Server
 
 Here's a server that deploys an OpenAI chat model, an Anthropic chat model, and a chain that uses
@@ -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
@@ -116,18 +124,18 @@ joke_chain.invoke({"topic": "parrots"})
 await joke_chain.ainvoke({"topic": "parrots"})
 
 prompt = [
-    SystemMessage(content='Act like either a cat or a parrot.'), 
+    SystemMessage(content='Act like either a cat or a parrot.'),
     HumanMessage(content='Hello!')
 ]
 
 # Supports astream
 async for msg in anthropic.astream(prompt):
     print(msg, end="", flush=True)
-    
+
 prompt = ChatPromptTemplate.from_messages(
     [("system", "Tell me a long story about {topic}")]
 )
-    
+
 # Can define custom chains
 chain = prompt | RunnableMap({
     "openai": openai,
@@ -142,9 +150,11 @@ In TypeScript (requires LangChain.js version 0.0.166 or later):
 ```typescript
 import { RemoteRunnable } from "langchain/runnables/remote";
 
-const chain = new RemoteRunnable({ url: `http://localhost:8000/chain/invoke/` });
+const chain = new RemoteRunnable({
+  url: `http://localhost:8000/chain/invoke/`,
+});
 const result = await chain.invoke({
-  "topic": "cats", 
+  topic: "cats",
 });
 ```
 
@@ -171,8 +181,7 @@ curl --location --request POST 'http://localhost:8000/chain/invoke/' \
     }'
 ```
 
-
-## Endpoints 
+## Endpoints
 
 The following code:
 
@@ -195,6 +204,10 @@ adds of these endpoints to the server:
 - `GET /my_runnable/output_schema` - json schema for output of the runnable
 - `GET /my_runnable/config_schema` - json schema for config of the runnable
 
+## Playground
+
+You can find a playground page for your runnable at `/my_runnable/playground`. This exposes a simple UI to [configure](https://python.langchain.com/docs/expression_language/how_to/configure) and invoke your runnable with streaming output and intermediate steps.
+
 ## Installation
 
 For both client and server:
@@ -212,10 +225,9 @@ However, some of the input schemas for legacy chains may be incomplete/incorrect
 This can be fixed by updating the `input_schema` property of those chains in LangChain.
 If you encounter any errors, please open an issue on THIS repo, and we will work to address it.
 
-
 ## Handling Authentication
 
-If you need to add authentication to your server, 
+If you need to add authentication to your server,
 please reference FastAPI's [security documentation](https://fastapi.tiangolo.com/tutorial/security/)
 and [middleware documentation](https://fastapi.tiangolo.com/tutorial/middleware/).
 

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/playground/dist/index.html

@@ -5,7 +5,7 @@
     <link rel="icon" href="/____LANGSERVE_BASE_URL/favicon.ico" />
     <meta name="viewport" content="width=device-width, initial-scale=1.0" />
     <title>Playground</title>
-    <script type="module" crossorigin src="/____LANGSERVE_BASE_URL/assets/index-6c8f83bb.js"></script>
+    <script type="module" crossorigin src="/____LANGSERVE_BASE_URL/assets/index-9fe7f71c.js"></script>
     <link rel="stylesheet" href="/____LANGSERVE_BASE_URL/assets/index-5008c8a8.css">
   </head>
   <body>

langserve/playground/package.json

@@ -29,6 +29,7 @@
     "react": "^18.2.0",
     "react-dom": "^18.2.0",
     "tailwind-merge": "^1.14.0",
+    "use-debounce": "^9.0.4",
     "vaul": "^0.7.3"
   },
   "devDependencies": {

langserve/playground/src/App.tsx

@@ -164,14 +164,14 @@ function App() {
 
   // store form state
   const [configData, setConfigData] = useState<
-    Pick<JsonFormsCore, "data" | "errors">
-  >({ data: {}, errors: [] });
+    Pick<JsonFormsCore, "data" | "errors"> & { defaults: boolean }
+  >({ data: {}, errors: [], defaults: true });
 
   const [inputData, setInputData] = useState<
     Pick<JsonFormsCore, "data" | "errors">
   >({ data: null, errors: [] });
   // fetch input and config schemas from the server
-  const schemas = useSchemas();
+  const schemas = useSchemas(configData);
   // apply defaults defined in each schema
   useEffect(() => {
     if (schemas.config) {
@@ -182,6 +182,7 @@ function App() {
           initConfigData.current ??
           defaults(schemas.config),
         errors: [],
+        defaults: true,
       });
       setInputData({ data: null, errors: [] });
     }
@@ -204,7 +205,11 @@ function App() {
               const value: { config: JsonFormsCore["data"] } = message.value;
               if (Object.keys(value.config).length > 0) {
                 initConfigData.current = value.config;
-                setConfigData({ data: value.config, errors: [] });
+                setConfigData({
+                  data: value.config,
+                  errors: [],
+                  defaults: false,
+                });
                 break;
               }
             }
@@ -232,7 +237,9 @@ function App() {
             renderers={renderers}
             cells={cells}
             onChange={({ data, errors }) =>
-              data ? setConfigData({ data, errors }) : undefined
+              data
+                ? setConfigData({ data, errors, defaults: false })
+                : undefined
             }
           />
           {!!configData.errors?.length && configData.data && (

langserve/playground/src/useSchemas.tsx

@@ -1,6 +1,9 @@
 import { useEffect, useState } from "react";
 import { resolveApiUrl } from "./utils/url";
 import { simplifySchema } from "./utils/simplifySchema";
+import { JsonFormsCore } from "@jsonforms/core";
+import { compressToEncodedURIComponent } from "lz-string";
+import { useDebounce } from "use-debounce";
 
 declare global {
   interface Window {
@@ -11,7 +14,9 @@ declare global {
   }
 }
 
-export function useSchemas() {
+export function useSchemas(
+  configData: Pick<JsonFormsCore, "data" | "errors"> & { defaults: boolean }
+) {
   const [schemas, setSchemas] = useState({
     config: null,
     input: null,
@@ -44,5 +49,23 @@ export function useSchemas() {
     save();
   }, []);
 
+  const [debouncedConfigData] = useDebounce(configData, 500);
+
+  useEffect(() => {
+    if (!debouncedConfigData.defaults) {
+      fetch(
+        resolveApiUrl(
+          `c/${compressToEncodedURIComponent(
+            JSON.stringify(debouncedConfigData.data)
+          )}/input_schema`
+        )
+      )
+        .then((r) => r.json())
+        .then(simplifySchema)
+        .then((input) => setSchemas((current) => ({ ...current, input })))
+        .catch(() => {}); // ignore errors, eg. due to incomplete config
+    }
+  }, [debouncedConfigData]);
+
   return schemas;
 }

langserve/playground/yarn.lock

@@ -2699,6 +2699,11 @@ use-callback-ref@^1.3.0:
   dependencies:
     tslib "^2.0.0"
 
+use-debounce@^9.0.4:
+  version "9.0.4"
+  resolved "https://registry.yarnpkg.com/use-debounce/-/use-debounce-9.0.4.tgz#51d25d856fbdfeb537553972ce3943b897f1ac85"
+  integrity sha512-6X8H/mikbrt0XE8e+JXRtZ8yYVvKkdYRfmIhWZYsP8rcNs9hk3APV8Ua2mFkKRLcJKVdnX2/Vwrmg2GWKUQEaQ==
+
 use-sidecar@^1.1.2:
   version "1.1.2"
   resolved "https://registry.yarnpkg.com/use-sidecar/-/use-sidecar-1.1.2.tgz#2f43126ba2d7d7e117aa5855e5d8f0276dfe73c2"

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")

poetry.lock

@@ -827,20 +827,20 @@ tests = ["asttokens (>=2.1.0)", "coverage", "coverage-enable-subprocess", "ipyth
 
 [[package]]
 name = "fastapi"
-version = "0.103.2"
+version = "0.104.0"
 description = "FastAPI framework, high performance, easy to learn, fast to code, ready for production"
 optional = false
-python-versions = ">=3.7"
+python-versions = ">=3.8"
 files = [
-    {file = "fastapi-0.103.2-py3-none-any.whl", hash = "sha256:3270de872f0fe9ec809d4bd3d4d890c6d5cc7b9611d721d6438f9dacc8c4ef2e"},
-    {file = "fastapi-0.103.2.tar.gz", hash = "sha256:75a11f6bfb8fc4d2bec0bd710c2d5f2829659c0e8c0afd5560fdda6ce25ec653"},
+    {file = "fastapi-0.104.0-py3-none-any.whl", hash = "sha256:456482c1178fb7beb2814b88e1885bc49f9a81f079665016feffe3e1c6a7663e"},
+    {file = "fastapi-0.104.0.tar.gz", hash = "sha256:9c44de45693ae037b0c6914727a29c49a40668432b67c859a87851fc6a7b74c6"},
 ]
 
 [package.dependencies]
 anyio = ">=3.7.1,<4.0.0"
 pydantic = ">=1.7.4,<1.8 || >1.8,<1.8.1 || >1.8.1,<2.0.0 || >2.0.0,<2.0.1 || >2.0.1,<2.1.0 || >2.1.0,<3.0.0"
 starlette = ">=0.27.0,<0.28.0"
-typing-extensions = ">=4.5.0"
+typing-extensions = ">=4.8.0"
 
 [package.extras]
 all = ["email-validator (>=2.0.0)", "httpx (>=0.23.0)", "itsdangerous (>=1.1.0)", "jinja2 (>=2.11.2)", "orjson (>=3.2.1)", "pydantic-extra-types (>=2.0.0)", "pydantic-settings (>=2.0.0)", "python-multipart (>=0.0.5)", "pyyaml (>=5.3.1)", "ujson (>=4.0.1,!=4.0.2,!=4.1.0,!=4.2.0,!=4.3.0,!=5.0.0,!=5.1.0)", "uvicorn[standard] (>=0.12.0)"]
@@ -1654,13 +1654,13 @@ test = ["hatch", "ipykernel", "openapi-core (>=0.18.0,<0.19.0)", "openapi-spec-v
 
 [[package]]
 name = "langchain"
-version = "0.0.316"
+version = "0.0.319"
 description = "Building applications with LLMs through composability"
 optional = false
 python-versions = ">=3.8.1,<4.0"
 files = [
-    {file = "langchain-0.0.316-py3-none-any.whl", hash = "sha256:997d2ecdaf481517c90dcea86f93b98327546c357ac8f581f06d7d782a2eda2d"},
-    {file = "langchain-0.0.316.tar.gz", hash = "sha256:732c958d41e22a5bb55f8c0cb70914793f083d175739d476e5cbbc7715703153"},
+    {file = "langchain-0.0.319-py3-none-any.whl", hash = "sha256:a61448fd418ff9478f2be3477c9c92acbf6b6acd8163c58c994a6158d4d116f8"},
+    {file = "langchain-0.0.319.tar.gz", hash = "sha256:4fe5025e5fd48dcf8e02107fefe173ba997af3c8960871cc4a4467e24bb89375"},
 ]
 
 [package.dependencies]
@@ -1685,7 +1685,7 @@ cli = ["typer (>=0.9.0,<0.10.0)"]
 cohere = ["cohere (>=4,<5)"]
 docarray = ["docarray[hnswlib] (>=0.32.0,<0.33.0)"]
 embeddings = ["sentence-transformers (>=2,<3)"]
-extended-testing = ["aiosqlite (>=0.19.0,<0.20.0)", "amazon-textract-caller (<2)", "anthropic (>=0.3.11,<0.4.0)", "arxiv (>=1.4,<2.0)", "assemblyai (>=0.17.0,<0.18.0)", "atlassian-python-api (>=3.36.0,<4.0.0)", "beautifulsoup4 (>=4,<5)", "bibtexparser (>=1.4.0,<2.0.0)", "cassio (>=0.1.0,<0.2.0)", "chardet (>=5.1.0,<6.0.0)", "dashvector (>=1.0.1,<2.0.0)", "esprima (>=4.0.1,<5.0.0)", "faiss-cpu (>=1,<2)", "feedparser (>=6.0.10,<7.0.0)", "geopandas (>=0.13.1,<0.14.0)", "gitpython (>=3.1.32,<4.0.0)", "gql (>=3.4.1,<4.0.0)", "html2text (>=2020.1.16,<2021.0.0)", "jinja2 (>=3,<4)", "jq (>=1.4.1,<2.0.0)", "lxml (>=4.9.2,<5.0.0)", "markdownify (>=0.11.6,<0.12.0)", "motor (>=3.3.1,<4.0.0)", "mwparserfromhell (>=0.6.4,<0.7.0)", "mwxml (>=0.3.3,<0.4.0)", "newspaper3k (>=0.2.8,<0.3.0)", "numexpr (>=2.8.6,<3.0.0)", "openai (>=0,<1)", "openapi-schema-pydantic (>=1.2,<2.0)", "pandas (>=2.0.1,<3.0.0)", "pdfminer-six (>=20221105,<20221106)", "pgvector (>=0.1.6,<0.2.0)", "psychicapi (>=0.8.0,<0.9.0)", "py-trello (>=0.19.0,<0.20.0)", "pymupdf (>=1.22.3,<2.0.0)", "pypdf (>=3.4.0,<4.0.0)", "pypdfium2 (>=4.10.0,<5.0.0)", "pyspark (>=3.4.0,<4.0.0)", "rank-bm25 (>=0.2.2,<0.3.0)", "rapidfuzz (>=3.1.1,<4.0.0)", "rapidocr-onnxruntime (>=1.3.2,<2.0.0)", "requests-toolbelt (>=1.0.0,<2.0.0)", "rspace_client (>=2.5.0,<3.0.0)", "scikit-learn (>=1.2.2,<2.0.0)", "sqlite-vss (>=0.1.2,<0.2.0)", "streamlit (>=1.18.0,<2.0.0)", "sympy (>=1.12,<2.0)", "telethon (>=1.28.5,<2.0.0)", "timescale-vector (>=0.0.1,<0.0.2)", "tqdm (>=4.48.0)", "upstash-redis (>=0.15.0,<0.16.0)", "xata (>=1.0.0a7,<2.0.0)", "xmltodict (>=0.13.0,<0.14.0)"]
+extended-testing = ["aiosqlite (>=0.19.0,<0.20.0)", "amazon-textract-caller (<2)", "anthropic (>=0.3.11,<0.4.0)", "arxiv (>=1.4,<2.0)", "assemblyai (>=0.17.0,<0.18.0)", "atlassian-python-api (>=3.36.0,<4.0.0)", "beautifulsoup4 (>=4,<5)", "bibtexparser (>=1.4.0,<2.0.0)", "cassio (>=0.1.0,<0.2.0)", "chardet (>=5.1.0,<6.0.0)", "dashvector (>=1.0.1,<2.0.0)", "esprima (>=4.0.1,<5.0.0)", "faiss-cpu (>=1,<2)", "feedparser (>=6.0.10,<7.0.0)", "geopandas (>=0.13.1,<0.14.0)", "gitpython (>=3.1.32,<4.0.0)", "gql (>=3.4.1,<4.0.0)", "html2text (>=2020.1.16,<2021.0.0)", "jinja2 (>=3,<4)", "jq (>=1.4.1,<2.0.0)", "lxml (>=4.9.2,<5.0.0)", "markdownify (>=0.11.6,<0.12.0)", "motor (>=3.3.1,<4.0.0)", "mwparserfromhell (>=0.6.4,<0.7.0)", "mwxml (>=0.3.3,<0.4.0)", "newspaper3k (>=0.2.8,<0.3.0)", "numexpr (>=2.8.6,<3.0.0)", "openai (>=0,<1)", "openapi-pydantic (>=0.3.2,<0.4.0)", "pandas (>=2.0.1,<3.0.0)", "pdfminer-six (>=20221105,<20221106)", "pgvector (>=0.1.6,<0.2.0)", "psychicapi (>=0.8.0,<0.9.0)", "py-trello (>=0.19.0,<0.20.0)", "pymupdf (>=1.22.3,<2.0.0)", "pypdf (>=3.4.0,<4.0.0)", "pypdfium2 (>=4.10.0,<5.0.0)", "pyspark (>=3.4.0,<4.0.0)", "rank-bm25 (>=0.2.2,<0.3.0)", "rapidfuzz (>=3.1.1,<4.0.0)", "rapidocr-onnxruntime (>=1.3.2,<2.0.0)", "requests-toolbelt (>=1.0.0,<2.0.0)", "rspace_client (>=2.5.0,<3.0.0)", "scikit-learn (>=1.2.2,<2.0.0)", "sqlite-vss (>=0.1.2,<0.2.0)", "streamlit (>=1.18.0,<2.0.0)", "sympy (>=1.12,<2.0)", "telethon (>=1.28.5,<2.0.0)", "timescale-vector (>=0.0.1,<0.0.2)", "tqdm (>=4.48.0)", "upstash-redis (>=0.15.0,<0.16.0)", "xata (>=1.0.0a7,<2.0.0)", "xmltodict (>=0.13.0,<0.14.0)"]
 javascript = ["esprima (>=4.0.1,<5.0.0)"]
 llms = ["clarifai (>=9.1.0)", "cohere (>=4,<5)", "huggingface_hub (>=0,<1)", "manifest-ml (>=0.0.1,<0.0.2)", "nlpcloud (>=1,<2)", "openai (>=0,<1)", "openlm (>=0.0.5,<0.0.6)", "torch (>=1,<3)", "transformers (>=4,<5)"]
 openai = ["openai (>=0,<1)", "tiktoken (>=0.3.2,<0.6.0)"]
@@ -1694,13 +1694,13 @@ text-helpers = ["chardet (>=5.1.0,<6.0.0)"]
 
 [[package]]
 name = "langsmith"
-version = "0.0.44"
+version = "0.0.47"
 description = "Client library to connect to the LangSmith LLM Tracing and Evaluation Platform."
 optional = false
 python-versions = ">=3.8.1,<4.0"
 files = [
-    {file = "langsmith-0.0.44-py3-none-any.whl", hash = "sha256:5e7e5b45360ce89a2d5d6066a3b9fdd31b1f874a0cf19b1666c9792fecef0a1b"},
-    {file = "langsmith-0.0.44.tar.gz", hash = "sha256:74a262ba23a958ca1a4863d5386c151be462e40ccfcb8b39d0a5d8c9eaa40164"},
+    {file = "langsmith-0.0.47-py3-none-any.whl", hash = "sha256:90279d4888e4513f83703becf38a6d92b7c7d696b120872fa1fb28c5c67bfd91"},
+    {file = "langsmith-0.0.47.tar.gz", hash = "sha256:04b8373cbfcf7b9c28ba53174206095c50dc09ae1131a99b501f92776c8ad0c8"},
 ]
 
 [package.dependencies]
@@ -2496,13 +2496,13 @@ testing = ["fields", "hunter", "process-tests", "pytest-xdist", "six", "virtuale
 
 [[package]]
 name = "pytest-mock"
-version = "3.11.1"
+version = "3.12.0"
 description = "Thin-wrapper around the mock package for easier use with pytest"
 optional = false
-python-versions = ">=3.7"
+python-versions = ">=3.8"
 files = [
-    {file = "pytest-mock-3.11.1.tar.gz", hash = "sha256:7f6b125602ac6d743e523ae0bfa71e1a697a2f5534064528c6ff84c2f7c2fc7f"},
-    {file = "pytest_mock-3.11.1-py3-none-any.whl", hash = "sha256:21c279fff83d70763b05f8874cc9cfb3fcacd6d354247a976f9529d19f9acf39"},
+    {file = "pytest-mock-3.12.0.tar.gz", hash = "sha256:31a40f038c22cad32287bb43932054451ff5583ff094bca6f675df2f8bc1a6e9"},
+    {file = "pytest_mock-3.12.0-py3-none-any.whl", hash = "sha256:0972719a7263072da3a21c7f4773069bcc7486027d7e8e1f81d98a47e701bc4f"},
 ]
 
 [package.dependencies]
@@ -3328,13 +3328,13 @@ dev = ["flake8", "flake8-annotations", "flake8-bandit", "flake8-bugbear", "flake
 
 [[package]]
 name = "urllib3"
-version = "2.0.6"
+version = "2.0.7"
 description = "HTTP library with thread-safe connection pooling, file post, and more."
 optional = false
 python-versions = ">=3.7"
 files = [
-    {file = "urllib3-2.0.6-py3-none-any.whl", hash = "sha256:7a7c7003b000adf9e7ca2a377c9688bbc54ed41b985789ed576570342a375cd2"},
-    {file = "urllib3-2.0.6.tar.gz", hash = "sha256:b19e1a85d206b56d7df1d5e683df4a7725252a964e3993648dd0fb5a1c157564"},
+    {file = "urllib3-2.0.7-py3-none-any.whl", hash = "sha256:fdb6d215c776278489906c2f8916e6e7d4f5a9b602ccbcfdf7f016fc8da0596e"},
+    {file = "urllib3-2.0.7.tar.gz", hash = "sha256:c97dfde1f7bd43a71c8d2a58e369e9b2bf692d1334ea9f9cae55add7d0dd0f84"},
 ]
 
 [package.extras]

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "langserve"
-version = "0.0.13"
+version = "0.0.15"
 description = ""
 readme = "README.md"
 authors = ["LangChain"]

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"