跳到主要内容

🦜️🏓 LangServe

Release Notes Downloads Open Issues

[!警告] 对于新项目，我们建议使用 LangGraph Platform 而不是 LangServe。

更多信息请参阅 LangGraph Platform 迁移指南。

我们将继续接受社区为 LangServe 提供的错误修复；但我们将不再接受新功能贡献。

概述​

LangServe 帮助开发者将 LangChain 的 可运行对象和链 部署为 REST API。

此库与 FastAPI 集成，并使用 pydantic 进行数据验证。

此外，它还提供了一个客户端，可用于调用部署在服务器上的可运行对象。LangChain.js 中提供了一个 JavaScript 客户端。

功能​

  • 输入和输出模式从您的 LangChain 对象中自动推断，并在每次 API 调用时强制执行，并附带丰富的错误消息
  • 带有 JSONSchema 和 Swagger 的 API 文档页面（插入示例链接）
  • 高效的 /invoke、/batch 和 /stream 端点，支持单个服务器上的多个并发请求
  • /stream_log 端点，用于流式传输链/代理的所有（或部分）中间步骤
  • 0.0.40 版本**新增**，支持 /stream_events，使得流式传输更容易，无需解析 /stream_log 的输出。
  • 在 /playground/ 的演练场页面，包含流式输出和中间步骤
  • 内置（可选）到 LangSmith 的追踪功能，只需添加您的 API 密钥（参见 说明）
  • 所有功能均使用经过实战检验的开源 Python 库构建，例如 FastAPI、Pydantic、uvloop 和 asyncio。
  • 使用客户端 SDK 调用 LangServe 服务器，就像调用本地运行的可运行对象一样（或者直接调用 HTTP API）
  • LangServe Hub

⚠️ LangGraph 兼容性​

LangServe 主要设计用于部署简单的可运行对象，并与 langchain-core 中众所周知的基本类型配合使用。

如果您需要 LangGraph 的部署选项，您应该考虑 LangGraph Cloud (beta)，它将更适合部署 LangGraph 应用程序。

限制​

  • 尚不支持源自服务器事件的客户端回调
  • LangServe <= 0.2.0 版本在使用 Pydantic V2 时无法正确生成 OpenAPI 文档，因为 FastAPI 不支持混合 pydantic v1 和 v2 命名空间。有关更多详细信息，请参阅以下部分。请升级到 LangServe>=0.3.0 或将 Pydantic 降级到 pydantic 1。

安全性​

安装​

同时适用于客户端和服务器

pip install "langserve[all]"

或者针对客户端代码使用 pip install "langserve[client]"，针对服务器代码使用 pip install "langserve[server]"。

LangChain CLI 🛠️​

使用 LangChain CLI 快速启动一个 LangServe 项目。

要使用 langchain CLI，请确保您已安装最新版本的 langchain-cli。您可以通过 pip install -U langchain-cli 安装它。

设置​

**注意**：我们使用 poetry 进行依赖管理。请遵循 poetry 文档 了解更多信息。

1. 使用 langchain CLI 命令创建新应用​

langchain app new my-app

2. 在 add_routes 中定义可运行对象。前往 server.py 进行编辑​

add_routes(app. NotImplemented)

3. 使用 poetry 添加第三方包（例如，langchain-openai, langchain-anthropic, langchain-mistral 等）。​

poetry add [package-name] // e.g `poetry add langchain-openai`

4. 设置相关环境变量。例如，​

export OPENAI_API_KEY="sk-..."

5. 运行您的应用​

poetry run langchain serve --port=8100

示例​

通过 examples 目录快速启动您的 LangServe 实例。

描述链接
**LLMs** 预留 OpenAI 和 Anthropic 聊天模型的最小示例。使用异步，支持批处理和流式传输。服务器, 客户端
**检索器** 暴露检索器作为可运行对象的简单服务器。服务器, 客户端
**对话式检索器** 通过 LangServe 暴露的 对话式检索器服务器, 客户端
**代理**（无**对话历史**）基于 OpenAI 工具服务器, 客户端
**代理**（带**对话历史**）基于 OpenAI 工具服务器, 客户端
RunnableWithMessageHistory 用于实现后端持久化聊天，以客户端提供的 session_id 为键。服务器, 客户端
RunnableWithMessageHistory 用于实现后端持久化聊天，以客户端提供的 conversation_id 和 user_id 为键（有关正确实现 user_id，请参阅身份验证部分）。服务器, 客户端
可配置可运行对象 用于创建一个支持运行时配置索引名称的检索器。服务器, 客户端
可配置可运行对象，显示可配置字段和可配置替代方案。服务器, 客户端
**APIHandler** 展示了如何使用 APIHandler 而不是 add_routes。这为开发者定义端点提供了更大的灵活性。与所有 FastAPI 模式配合良好，但需要更多一些工作。服务器
**LCEL 示例** 使用 LCEL 操作字典输入的示例。服务器, 客户端
使用 add_routes 的**身份验证**：可应用于与应用关联的所有端点的简单身份验证。（单独使用对于实现每个用户的逻辑没有用处。）服务器
使用 add_routes 的**身份验证**：基于路径依赖的简单身份验证机制。（单独使用对于实现每个用户的逻辑没有用处。）服务器
使用 add_routes 的**身份验证**：为使用按请求配置修改器的端点实现每个用户的逻辑和身份验证。（**注意**：目前，未与 OpenAPI 文档集成。）服务器, 客户端
使用 APIHandler 的**身份验证**：实现每个用户的逻辑和身份验证，展示如何仅在用户拥有的文档中进行搜索。服务器, 客户端
**小部件** 可用于演练场的不同小部件（文件上传和聊天）服务器
**小部件** 用于 LangServe 演练场的文件上传小部件。服务器, 客户端

示例应用​

服务器​

这是一个部署了 OpenAI 聊天模型、Anthropic 聊天模型以及一个使用 Anthropic 模型讲述关于某个主题笑话的链的服务器。

#!/usr/bin/env python
from fastapi import FastAPI
from langchain.prompts import ChatPromptTemplate
from langchain.chat_models import ChatAnthropic, ChatOpenAI
from langserve import add_routes

app = FastAPI(
    title="LangChain Server",
    version="1.0",
    description="A simple api server using Langchain's Runnable interfaces",
)

add_routes(
    app,
    ChatOpenAI(model="gpt-3.5-turbo-0125"),
    path="/openai",
)

add_routes(
    app,
    ChatAnthropic(model="claude-3-haiku-20240307"),
    path="/anthropic",
)

model = ChatAnthropic(model="claude-3-haiku-20240307")
prompt = ChatPromptTemplate.from_template("tell me a joke about {topic}")
add_routes(
    app,
    prompt | model,
    path="/joke",
)

if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="localhost", port=8000)

如果您打算从浏览器调用您的端点，您还需要设置 CORS 头。您可以使用 FastAPI 内置的中间件来实现这一点

from fastapi.middleware.cors import CORSMiddleware

# Set all CORS enabled origins
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
    expose_headers=["*"],
)

文档​

如果您已部署上述服务器，您可以使用以下方式查看生成的 OpenAPI 文档

⚠️ 如果使用 LangServe <= 0.2.0 和 pydantic v2，则不会为 invoke、batch、stream、stream_log 生成文档。有关更多详细信息，请参阅下面的 Pydantic 部分。要解决此问题，请升级到 LangServe 0.3.0。

curl localhost:8000/docs

请务必**添加** /docs 后缀。

⚠️ 索引页 / **设计**上未定义，因此 curl localhost:8000 或访问该 URL 将返回 404。如果您想在 / 处显示内容，请定义一个 @app.get("/") 端点。

客户端​

Python SDK


from langchain.schema import SystemMessage, HumanMessage
from langchain.prompts import ChatPromptTemplate
from langchain.schema.runnable import RunnableMap
from langserve import RemoteRunnable

openai = RemoteRunnable("https://:8000/openai/")
anthropic = RemoteRunnable("https://:8000/anthropic/")
joke_chain = RemoteRunnable("https://:8000/joke/")

joke_chain.invoke({"topic": "parrots"})

# or async
await joke_chain.ainvoke({"topic": "parrots"})

prompt = [
    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,
    "anthropic": anthropic,
})

chain.batch([{"topic": "parrots"}, {"topic": "cats"}])

在 TypeScript 中（需要 LangChain.js 0.0.166 或更高版本）

import { RemoteRunnable } from "@langchain/core/runnables/remote";

const chain = new RemoteRunnable({
  url: `https://:8000/joke/`,
});
const result = await chain.invoke({
  topic: "cats",
});

使用 requests 的 Python

import requests

response = requests.post(
    "https://:8000/joke/invoke",
    json={'input': {'topic': 'cats'}}
)
response.json()

您也可以使用 curl

curl --location --request POST 'https://:8000/joke/invoke' \
    --header 'Content-Type: application/json' \
    --data-raw '{
        "input": {
            "topic": "cats"
        }
    }'

端点​

以下代码

...
add_routes(
    app,
    runnable,
    path="/my_runnable",
)

将这些端点添加到服务器

  • POST /my_runnable/invoke - 对单个输入调用可运行对象
  • POST /my_runnable/batch - 对批量输入调用可运行对象
  • POST /my_runnable/stream - 对单个输入进行调用并流式传输输出
  • POST /my_runnable/stream_log - 对单个输入进行调用并流式传输输出，包括生成时中间步骤的输出
  • POST /my_runnable/astream_events - 对单个输入进行调用并流式传输事件，包括来自中间步骤的事件。
  • GET /my_runnable/input_schema - 可运行对象输入的 JSON 模式
  • GET /my_runnable/output_schema - 可运行对象输出的 JSON 模式
  • GET /my_runnable/config_schema - 可运行对象配置的 JSON 模式

这些端点与 LangChain 表达式语言接口 匹配——请参考此文档以获取更多详细信息。

演练场​

您可以在 /my_runnable/playground/ 找到您的可运行对象的演练场页面。这提供了一个简单的 UI，用于配置和调用您的可运行对象，并提供流式输出和中间步骤。

小部件​

演练场支持小部件，可用于测试您的可运行对象与不同输入。有关更多详细信息，请参阅下面的小部件部分。

分享​

此外，对于可配置的可运行对象，演练场将允许您配置可运行对象并分享带有配置的链接

聊天演练场​

LangServe 还支持一个以聊天为中心的演练场，您可以在 /my_runnable/playground/ 下选择并使用它。与通用演练场不同，只支持某些类型的可运行对象——可运行对象的输入模式必须是一个 dict，其中包含

  • 单个键，且该键的值必须是聊天消息列表。
  • 两个键，一个键的值是消息列表，另一个键代表最新消息。

我们建议您使用第一种格式。

可运行对象还必须返回 AIMessage 或字符串。

要启用它，您在添加路由时必须设置 playground_type="chat",。这是一个示例

# Declare a chain
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a helpful, professional assistant named Cob."),
        MessagesPlaceholder(variable_name="messages"),
    ]
)

chain = prompt | ChatAnthropic(model="claude-2.1")


class InputChat(BaseModel):
    """Input for the chat endpoint."""

    messages: List[Union[HumanMessage, AIMessage, SystemMessage]] = Field(
        ...,
        description="The chat messages representing the current conversation.",
    )


add_routes(
    app,
    chain.with_types(input_type=InputChat),
    enable_feedback_endpoint=True,
    enable_public_trace_link_endpoint=True,
    playground_type="chat",
)

如果您正在使用 LangSmith，您还可以在路由上设置 enable_feedback_endpoint=True，以在每条消息后启用点赞/点踩按钮，并设置 enable_public_trace_link_endpoint=True 以添加一个创建公共追踪链接的按钮。请注意，您还需要设置以下环境变量

export LANGCHAIN_TRACING_V2="true"
export LANGCHAIN_PROJECT="YOUR_PROJECT_NAME"
export LANGCHAIN_API_KEY="YOUR_API_KEY"

这是上述两个选项开启的示例

注意：如果您启用公共追踪链接，您的链的内部结构将暴露。我们建议仅将此设置用于演示或测试。

旧版链​

LangServe 可与可运行对象（通过LangChain 表达式语言构建）和旧版链（继承自 Chain）一起使用。但是，某些旧版链的输入模式可能不完整/不正确，导致错误。这可以通过更新 LangChain 中这些链的 input_schema 属性来解决。如果您遇到任何错误，请在此仓库上提出问题，我们将努力解决。

部署​

部署到 AWS​

您可以使用 AWS Copilot CLI 部署到 AWS

copilot init --app [application-name] --name [service-name] --type 'Load Balanced Web Service' --dockerfile './Dockerfile' --deploy

点击此处了解更多信息。

部署到 Azure​

您可以使用 Azure 容器应用（无服务器）部署到 Azure

az containerapp up --name [container-app-name] --source . --resource-group [resource-group-name] --environment  [environment-name] --ingress external --target-port 8001 --env-vars=OPENAI_API_KEY=your_key

您可以在此处找到更多信息

部署到 GCP​

您可以使用以下命令部署到 GCP Cloud Run

gcloud run deploy [your-service-name] --source . --port 8001 --allow-unauthenticated --region us-central1 --set-env-vars=OPENAI_API_KEY=your_key

社区贡献​

部署到 Railway​

Railway 仓库示例

Deploy on Railway

Pydantic​

LangServe>=0.3 完全支持 Pydantic 2。

如果您使用的是早期版本的 LangServe (<= 0.2)，请注意 Pydantic 2 的支持存在以下限制

  1. 当使用 Pydantic V2 时，OpenAPI 文档将不会为 invoke/batch/stream/stream_log 生成。Fast API 不支持混合 pydantic v1 和 v2 命名空间。要解决此问题，请使用 pip install pydantic==1.10.17。
  2. LangChain 在 Pydantic v2 中使用 v1 命名空间。请阅读以下指南以确保与 LangChain 的兼容性

除了这些限制，我们预计 API 端点、演练场和任何其他功能都能按预期工作。

高级​

处理身份验证​

如果您需要为服务器添加身份验证，请阅读 Fast API 关于依赖关系和安全性的文档。

以下示例展示了如何使用 FastAPI 原语将身份验证逻辑连接到 LangServe 端点。

您负责提供实际的身份验证逻辑、用户表等。

如果您不确定如何操作，可以尝试使用现有解决方案 Auth0。

使用 add_routes​

如果您正在使用 add_routes，请参阅此处的示例。

描述链接
使用 add_routes 的**身份验证**：可应用于与应用关联的所有端点的简单身份验证。（单独使用对于实现每个用户的逻辑没有用处。）服务器
使用 add_routes 的**身份验证**：基于路径依赖的简单身份验证机制。（单独使用对于实现每个用户的逻辑没有用处。）服务器
使用 add_routes 的**身份验证**：为使用按请求配置修改器的端点实现每个用户的逻辑和身份验证。（**注意**：目前，未与 OpenAPI 文档集成。）服务器, 客户端

或者，您可以使用 FastAPI 的中间件。

使用全局依赖和路径依赖的优点是身份验证将在 OpenAPI 文档页面中得到良好支持，但这些不足以实现每个用户的逻辑（例如，创建一个只能在用户拥有的文档中搜索的应用程序）。

如果您需要实现每个用户的逻辑，可以使用 per_req_config_modifier 或 APIHandler（见下文）来实现此逻辑。

每个用户

如果您需要依赖于用户的授权或逻辑，在使用 add_routes 时请指定 per_req_config_modifier。使用一个可调用对象，该对象接收原始 Request 对象，并可从中提取相关信息用于身份验证和授权目的。

使用 APIHandler​

如果您熟悉 FastAPI 和 Python，可以使用 LangServe 的APIHandler。

描述链接
使用 APIHandler 的**身份验证**：实现每个用户的逻辑和身份验证，展示如何仅在用户拥有的文档中进行搜索。服务器, 客户端
**APIHandler** 展示了如何使用 APIHandler 而不是 add_routes。这为开发者定义端点提供了更大的灵活性。与所有 FastAPI 模式配合良好，但需要更多一些工作。服务器, 客户端

这会多一些工作，但它让您可以完全控制端点定义，从而可以实现您所需的任何自定义身份验证逻辑。

文件​

LLM 应用通常处理文件。可以采用不同的架构来实现文件处理；从高层次看

  1. 文件可以通过专用端点上传到服务器，并使用单独的端点进行处理
  2. 文件可以通过值（文件字节）或引用（例如，s3 文件内容的 URL）上传
  3. 处理端点可以是阻塞的或非阻塞的
  4. 如果需要大量处理，处理可能会被卸载到专用的进程池

您应该确定哪种架构适合您的应用。

目前，要通过值将文件上传到可运行对象，请使用 base64 编码文件（multipart/form-data 暂不支持）。

这是一个示例，展示了如何使用 base64 编码将文件发送到远程可运行对象。

请记住，您始终可以通过引用（例如 s3 URL）上传文件，或者以 multipart/form-data 格式将其上传到专用端点。

自定义输入和输出类型​

输入和输出类型在所有可运行对象上定义。

您可以通过 input_schema 和 output_schema 属性访问它们。

LangServe 使用这些类型进行验证和文档编写。

如果您想覆盖默认推断的类型，可以使用 with_types 方法。

这是一个说明该思想的示例

from typing import Any

from fastapi import FastAPI
from langchain.schema.runnable import RunnableLambda

app = FastAPI()


def func(x: Any) -> int:
    """Mistyped function that should accept an int but accepts anything."""
    return x + 1


runnable = RunnableLambda(func).with_types(
    input_type=int,
)

add_routes(app, runnable)
API 参考:RunnableLambda

自定义用户类型​

如果您希望数据反序列化为 pydantic 模型而不是等效的字典表示，请继承自 CustomUserType。

目前，此类型仅在服务器端有效，并用于指定所需的解码行为。如果继承此类型，服务器将保留解码后的类型为 pydantic 模型，而不是将其转换为字典。

from fastapi import FastAPI
from langchain.schema.runnable import RunnableLambda

from langserve import add_routes
from langserve.schema import CustomUserType

app = FastAPI()


class Foo(CustomUserType):
    bar: int


def func(foo: Foo) -> int:
    """Sample function that expects a Foo type which is a pydantic model"""
    assert isinstance(foo, Foo)
    return foo.bar


# Note that the input and output type are automatically inferred!
# You do not need to specify them.
# runnable = RunnableLambda(func).with_types( # <-- Not needed in this case
#     input_type=Foo,
#     output_type=int,
#
add_routes(app, RunnableLambda(func), path="/foo")
API 参考:RunnableLambda

演练场小部件​

演练场允许您从后端为可运行对象定义自定义小部件。

以下是一些示例

描述链接
**小部件** 可用于演练场的不同小部件（文件上传和聊天）服务器, 客户端
**小部件** 用于 LangServe 演练场的文件上传小部件。服务器, 客户端

模式​

  • 小部件在字段级别指定，并作为输入类型 JSON 模式的一部分进行分发
  • 小部件必须包含一个名为 type 的键，其值是已知小部件列表中的一个
  • 其他小部件键将与描述 JSON 对象中路径的值相关联
type JsonPath = number | string | (number | string)[];
type NameSpacedPath = { title: string; path: JsonPath }; // Using title to mimick json schema, but can use namespace
type OneOfPath = { oneOf: JsonPath[] };

type Widget = {
  type: string; // Some well known type (e.g., base64file, chat etc.)
  [key: string]: JsonPath | NameSpacedPath | OneOfPath;
};

可用小部件​

目前用户只能手动指定两种小部件

  1. 文件上传小部件
  2. 聊天历史小部件

有关这些小部件的更多信息，请参阅下文。

演练场 UI 上的所有其他小部件均由 UI 根据可运行对象的配置模式自动创建和管理。当您创建可配置可运行对象时，演练场应为您创建适当的小部件以控制其行为。

文件上传小部件​

允许在 UI 演练场中创建文件上传输入，用于上传为 base64 编码字符串的文件。这是完整示例。

代码片段

try:
    from pydantic.v1 import Field
except ImportError:
    from pydantic import Field

from langserve import CustomUserType


# ATTENTION: Inherit from CustomUserType instead of BaseModel otherwise
#            the server will decode it into a dict instead of a pydantic model.
class FileProcessingRequest(CustomUserType):
    """Request including a base64 encoded file."""

    # The extra field is used to specify a widget for the playground UI.
    file: str = Field(..., extra={"widget": {"type": "base64file"}})
    num_chars: int = 100

示例小部件

聊天小部件​

查看小部件示例。

要定义聊天小部件，请确保传递 "type": "chat"。

  • "input" 是 Request 中包含新输入消息的字段的 JSONPath。
  • "output" 是 Response 中包含新输出消息的字段的 JSONPath。
  • 如果整个输入或输出应按原样使用（例如，如果输出是聊天消息列表），则无需指定这些字段。

这是一个代码片段

class ChatHistory(CustomUserType):
    chat_history: List[Tuple[str, str]] = Field(
        ...,
        examples=[[("human input", "ai response")]],
        extra={"widget": {"type": "chat", "input": "question", "output": "answer"}},
    )
    question: str


def _format_to_messages(input: ChatHistory) -> List[BaseMessage]:
    """Format the input to a list of messages."""
    history = input.chat_history
    user_input = input.question

    messages = []

    for human, ai in history:
        messages.append(HumanMessage(content=human))
        messages.append(AIMessage(content=ai))
    messages.append(HumanMessage(content=user_input))
    return messages


model = ChatOpenAI()
chat_model = RunnableParallel({"answer": (RunnableLambda(_format_to_messages) | model)})
add_routes(
    app,
    chat_model.with_types(input_type=ChatHistory),
    config_keys=["configurable"],
    path="/chat",
)

示例小部件

您也可以直接将消息列表指定为参数，如本代码片段所示

prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a helpful assisstant named Cob."),
        MessagesPlaceholder(variable_name="messages"),
    ]
)

chain = prompt | ChatAnthropic(model="claude-2.1")


class MessageListInput(BaseModel):
    """Input for the chat endpoint."""
    messages: List[Union[HumanMessage, AIMessage]] = Field(
        ...,
        description="The chat messages representing the current conversation.",
        extra={"widget": {"type": "chat", "input": "messages"}},
    )


add_routes(
    app,
    chain.with_types(input_type=MessageListInput),
    path="/chat",
)

请参阅此示例文件以获取示例。

启用/禁用端点 (LangServe >=0.0.33)​

您可以启用/禁用在为给定链添加路由时暴露哪些端点。

如果您想确保在将 LangServe 升级到新版本时绝不会获得新的端点，请使用 enabled_endpoints。

启用：以下代码将仅启用 invoke、batch 以及相应的 config_hash 端点变体。

add_routes(app, chain, enabled_endpoints=["invoke", "batch", "config_hashes"], path="/mychain")

禁用：以下代码将禁用该链的演练场

add_routes(app, chain, disabled_endpoints=["playground"], path="/mychain")