如何在LangChain Runnable中向前端返回FastAPI HTTPException？

后端基于LangChain Runnable配置FastAPI路由，期望向前端返回404 Not found这类HTTPException，但直接在Runnable逻辑中抛出raise HTTPException(status_code=404, detail="Item not found")时，会触发CancelledException，最终前端收到"内部服务器错误"的500响应。由于Runnable直接传入add_routes函数，无法直接在路由层捕获异常，需要特殊处理。

报错栈 #

ERROR:    Exception in ASGI application
Traceback (most recent call last):
  File "/my-code-path/.venv/lib/python3.10/site-packages/sse_starlette/sse.py", line 269, in __call__
    await wrap(partial(self.listen_for_disconnect, receive))
  File "/my-code-path/.venv/lib/python3.10/site-packages/sse_starlette/sse.py", line 258, in wrap
    await func()
  File "/my-code-path/.venv/lib/python3.10/site-packages/sse_starlette/sse.py", line 215, in listen_for_disconnect
    message = await receive()
  File "/my-code-path/.venv/lib/python3.10/site-packages/uvicorn/protocols/http/httptools_impl.py", line 596, in receive
    await self.message_event.wait()
  File "/usr/lib/python3.10/asyncio/locks.py", line 214, in wait
    await fut
asyncio.exceptions.CancelledError: Cancelled by cancel scope 7fba499a8490

During handling of the above exception, another exception occurred:

  + Exception Group Traceback (most recent call last):
  |   File "/my-code-path/.venv/lib/python3.10/site-packages/uvicorn/protocols/http/httptools_impl.py", line 435, in run_asgi
  |     result = await app(  # type: ignore[func-returns-value]
  |   File "/my-code-path/.venv/lib/python3.10/site-packages/uvicorn/middleware/proxy_headers.py", line 78, in __call__
  |     return await self.app(scope, receive, send)
  |   File "/my-code-path/.venv/lib/python3.10/site-packages/fastapi/applications.py", line 276, in __call__
  |     await super().__call__(scope, receive, send)
  |   File "/my-code-path/.venv/lib/python3.10/site-packages/starlette/applications.py", line 122, in __call__
  |     await self.middleware_stack(scope, receive, send)
  |   File "/my-code-path/.venv/lib/python3.10/site-packages/starlette/middleware/errors.py", line 184, in __call__
  |     raise exc
  |   File "/my-code-path/.venv/lib/python3.10/site-packages/starlette/middleware/errors.py", line 162, in __call__
  |     await self.app(scope, receive, _send)
  |   File "/my-code-path/.venv/lib/python3.10/site-packages/starlette/middleware/cors.py", line 91, in __call__
  |     await self.simple_response(scope, receive, send, request_headers=headers)
  |   File "/my-code-path/.venv/lib/python3.10/site-packages/starlette/middleware/cors.py", line 146, in simple_response
  |     await self.app(scope, receive, send)
  |   File "/my-code-path/.venv/lib/python3.10/site-packages/starlette/middleware/exceptions.py", line 79, in __call__
  |     raise exc
  |   File "/my-code-path/.venv/lib/python3.10/site-packages/starlette/middleware/exceptions.py", line 68, in __call__
  |     await self.app(scope, receive, sender)
  |   File "/my-code-path/.venv/lib/python3.10/site-packages/fastapi/middleware/asyncexitstack.py", line 21, in __call__
  |     raise e
  |   File "/my-code-path/.venv/lib/python3.10/site-packages/fastapi/middleware/asyncexitstack.py", line 18, in __call__
  |     await self.app(scope, receive, send)
  |   File "/my-code-path/.venv/lib/python3.10/site-packages/starlette/routing.py", line 718, in __call__
  |     await route.handle(scope, receive, send)
  |   File "/my-code-path/.venv/lib/python3.10/site-packages/starlette/routing.py", line 276, in handle
  |     await self.app(scope, receive, send)
  |   File "/my-code-path/.venv/lib/python3.10/site-packages/starlette/routing.py", line 69, in app
  |     await response(scope, receive, send)
  |   File "/my-code-path/.venv/lib/python3.10/site-packages/sse_starlette/sse.py", line 255, in __call__
  |     async with anyio.create_task_group() as task_group:
  |   File "/my-code-path/.venv/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 680, in __aexit__
  |     raise BaseExceptionGroup(
  | exceptiongroup.ExceptionGroup: unhandled errors in a TaskGroup (1 sub-exception)
  +-+---------------- 1 ----------------
    | Traceback (most recent call last):
    |   File "/my-code-path/.venv/lib/python3.10/site-packages/sse_starlette/sse.py", line 258, in wrap
    |     await func()
    |   File "/my-code-path/.venv/lib/python3.10/site-packages/sse_starlette/sse.py", line 245, in stream_response
    |     async for data in self.body_iterator:
    |   File "/my-code-path/.venv/lib/python3.10/site-packages/langserve/api_handler.py", line 1121, in _stream
    |     async for chunk in self._runnable.astream(
    |   File "/my-code-path/.venv/lib/python3.10/site-packages/langchain_core/runnables/base.py", line 4745, in astream
    |     async for item in self.bound.astream(
    |   File "/my-code-path/.venv/lib/python3.10/site-packages/langchain_core/runnables/base.py", line 2898, in astream
    |     async for chunk in self.atransform(input_aiter(), config, **kwargs):
    |   File "/my-code-path/.venv/lib/python3.10/site-packages/langchain_core/runnables/base.py", line 2881, in atransform
    |     async for chunk in self._atransform_stream_with_config(
    |   File "/my-code-path/.venv/lib/python3.10/site-packages/langchain_core/runnables/base.py", line 1972, in _atransform_stream_with_config
    |     chunk = cast(Output, await py_anext(iterator))
    |   File "/my-code-path/.venv/lib/python3.10/site-packages/langchain_core/runnables/base.py", line 2851, in _atransform
    |     async for output in final_pipeline:
    |   File "/my-code-path/.venv/lib/python3.10/site-packages/langchain_core/runnables/base.py", line 4178, in atransform
    |     async for output in self._atransform_stream_with_config(
    |   File "/my-code-path/.venv/lib/python3.10/site-packages/langchain_core/runnables/base.py", line 1972, in _atransform_stream_with_config
    |     chunk = cast(Output, await py_anext(iterator))
    |   File "/my-code-path/.venv/lib/python3.10/site-packages/langchain_core/runnables/base.py", line 4147, in _atransform
    |     output = await acall_func_with_variable_args(
    |   File "/my-code-path/.venv/lib/python3.10/site-packages/langchain_core/runnables/base.py", line 4122, in f
    |     return await run_in_executor(config, func, *args, **kwargs)
    |   File "/my-code-path/.venv/lib/python3.10/site-packages/langchain_core/runnables/config.py", line 547, in run_in_executor
    |     return await asyncio.get_running_loop().run_in_executor(
    |   File "/usr/lib/python3.10/concurrent/futures/thread.py", line 58, in run
    |     result = self.fn(*self.args, **self.kwargs)
    |   File "/my-code-path/.venv/lib/python3.10/site-packages/langchain_core/runnables/base.py", line 4116, in func
    |     return call_func_with_variable_args(
    |   File "/my-code-path/.venv/lib/python3.10/site-packages/langchain_core/runnables/config.py", line 380, in call_func_with_variable_args
    |     return func(input, **kwargs)  # type: ignore[call-arg]
    |   File "/my-code-path/src/chains/forms/refine_chain.py", line 65, in get_refined_runnable_map
    |     raise HTTPException(status_code=404, detail="Item not found")
    | fastapi.exceptions.HTTPException
    +------------------------------------

当前代码示例 #

refine_chain = RefineChain(custom_app, tool_name)
add_routes(
  app,
  refine_chain.get_runnable().with_config({"tags": [custom_app]}),
  enabled_endpoints=["stream"],
  path=f"/{custom_app}/{tool_name}/refine",
  config_keys=["configurable"],
  dependencies=[Depends(get_current_active_username)],
)   

解决方案 #

方案1：用RunnableLambda包装异常捕获逻辑 #

将抛出HTTPException的逻辑包裹在RunnableLambda中，捕获异常并转换为可识别的标记，再通过FastAPI中间件处理：

from langchain_core.runnables import RunnableLambda
from fastapi import HTTPException

def wrap_runnable_with_exception_handler(runnable):
    async def handle_exception(input_data):
        try:
            return await runnable.ainvoke(input_data)
        except HTTPException as e:
            # 用特殊标记封装异常信息
            raise RuntimeError(f"HTTP_EXCEPTION:{e.status_code}:{e.detail}") from e
        except Exception as e:
            raise e
    
    return RunnableLambda(handle_exception)

# 使用包装后的runnable
refine_chain = RefineChain(custom_app, tool_name)
wrapped_runnable = wrap_runnable_with_exception_handler(refine_chain.get_runnable())

add_routes(
  app,
  wrapped_runnable.with_config({"tags": [custom_app]}),
  enabled_endpoints=["stream"],
  path=f"/{custom_app}/{tool_name}/refine",
  config_keys=["configurable"],
  dependencies=[Depends(get_current_active_username)],
)

添加FastAPI异常处理中间件：

from fastapi import Request
from fastapi.responses import JSONResponse

@app.exception_handler(RuntimeError)
async def handle_http_exception_from_runnable(request: Request, exc: RuntimeError):
    if exc.args[0].startswith("HTTP_EXCEPTION:"):
        _, status_code, detail = exc.args[0].split(":", 2)
        return JSONResponse(
            status_code=int(status_code),
            content={"detail": detail}
        )
    # 其他异常按默认逻辑返回
    return JSONResponse(
        status_code=500,
        content={"detail": "Internal Server Error"}
    )

方案2：直接捕获ExceptionGroup中的HTTPException #

针对LangServe抛出的ExceptionGroup，提取其中的HTTPException并返回正确响应：

from exceptiongroup import BaseExceptionGroup
from fastapi import HTTPException
from fastapi.responses import JSONResponse

@app.exception_handler(BaseExceptionGroup)
async def handle_exception_group(request: Request, exc: BaseExceptionGroup):
    # 遍历异常组查找HTTPException
    for sub_exc in exc.exceptions:
        if isinstance(sub_exc.__cause__, HTTPException):
            http_exc = sub_exc.__cause__
            return JSONResponse(
                status_code=http_exc.status_code,
                content={"detail": http_exc.detail}
            )
    # 无匹配异常返回默认500
    return JSONResponse(
        status_code=500,
        content={"detail": "Internal Server Error"}
    )

方案3：在Runnable中返回错误格式而非抛出异常 #

如果允许修改返回结构，直接在Runnable中返回错误信息，由前端自行处理状态：

# 修改RefineChain的get_refined_runnable_map方法
def get_refined_runnable_map(self):
    def process_input(input_data):
        if not self._check_item_exists(input_data):
            # 返回错误结构
            return {"error": True, "status_code": 404, "detail": "Item not found"}
        # 正常业务逻辑
        return self._process_refine(input_data)
    
    return RunnableLambda(process_input)

内容的提问来源于stack exchange，提问作者Carles Roch Arnau