跳到主要内容

🦜️🏓 LangServe

Release Notes Downloads Open Issues

[!WARNING] 我们建议新项目使用 LangGraph Platform 而不是 LangServe。

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

我们将继续接受社区对 LangServe 的错误修复；但是，我们将不接受新的功能贡献。

概述​

LangServe 帮助开发人员将 LangChain runnables 和 chains 部署为 REST API。

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

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

功能​

  • 输入和输出模式从您的 LangChain 对象自动推断，并在每个 API 调用上强制执行，并提供丰富的错误消息
  • 带有 JSONSchema 和 Swagger 的 API 文档页面（插入示例链接）
  • 高效的 /invoke、/batch 和 /stream 端点，支持单个服务器上的多个并发请求
  • /stream_log 端点，用于流式传输来自您的 chain/agent 的所有（或部分）中间步骤
  • 新功能 自 0.0.40 起，支持 /stream_events，以便更容易进行流式传输，而无需解析 /stream_log 的输出。
  • /playground/ 上的 Playground 页面，具有流式输出和中间步骤
  • 内置（可选）跟踪到 LangSmith，只需添加您的 API 密钥（请参阅 说明）
  • 全部使用经过实战检验的开源 Python 库构建，如 FastAPI、Pydantic、uvloop 和 asyncio。
  • 使用客户端 SDK 调用 LangServe 服务器，就像它是本地运行的 Runnable 一样（或直接调用 HTTP API）
  • LangServe Hub

⚠️ LangGraph 兼容性​

LangServe 主要设计用于部署简单的 Runnables，并与 langchain-core 中的知名原语一起工作。

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

局限性​

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

安全性​

  • 0.0.13 - 0.0.15 版本中的漏洞 -- playground 端点允许访问服务器上的任意文件。已在 0.0.16 中解决。

安装​

对于客户端和服务器

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 中定义 runnable。转到 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 聊天模型。使用异步，支持批量处理和流式传输。服务器， 客户端
Retriever 简单的服务器，将 retriever 作为 runnable 公开。服务器， 客户端
Conversational Retriever 通过 LangServe 公开的 Conversational Retriever服务器， 客户端
基于 OpenAI tools 的Agent，没有对话历史记录服务器， 客户端
基于 OpenAI tools 的Agent，带有对话历史记录服务器， 客户端
RunnableWithMessageHistory 用于实现持久化在后端的聊天，通过客户端提供的 session_id 键控。服务器， 客户端
RunnableWithMessageHistory 用于实现持久化在后端的聊天，通过客户端提供的 conversation_id 和 user_id 键控（有关正确实现 user_id，请参阅 Auth）。服务器， 客户端
Configurable Runnable 用于创建支持运行时配置索引名称的 retriever。服务器， 客户端
Configurable Runnable，展示可配置字段和可配置替代方案。服务器， 客户端
APIHandler 展示如何使用 APIHandler 而不是 add_routes。这为开发人员定义端点提供了更大的灵活性。与所有 FastAPI 模式配合良好，但需要付出更多努力。服务器
LCEL Example 使用 LCEL 操作字典输入的示例。服务器， 客户端
带有 add_routes 的 Auth：简单的身份验证，可以应用于与应用关联的所有端点。（对于实现每个用户的逻辑本身没有用处。）服务器
带有 add_routes 的 Auth：基于路径依赖的简单身份验证机制。（对于实现每个用户的逻辑本身没有用处。）服务器
带有 add_routes 的 Auth：为使用每个请求配置修改器的端点实现每个用户逻辑和身份验证。（注意：目前，不与 OpenAPI 文档集成。）服务器， 客户端
带有 APIHandler 的 Auth：实现每个用户逻辑和身份验证，展示如何仅在用户拥有的文档中搜索。服务器， 客户端
Widgets 可以与 playground 一起使用的不同小部件（文件上传和聊天）服务器
Widgets 用于 LangServe playground 的文件上传小部件。服务器， 客户端

示例应用程序​

服务器​

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

#!/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://127.0.0.1:8000/openai/")
anthropic = RemoteRunnable("https://127.0.0.1:8000/anthropic/")
joke_chain = RemoteRunnable("https://127.0.0.1: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://127.0.0.1:8000/joke/`,
});
const result = await chain.invoke({
  topic: "cats",
});

使用 requests 的 Python

import requests

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

您也可以使用 curl

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

端点​

以下代码

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

将这些端点添加到服务器

  • POST /my_runnable/invoke - 对单个输入调用 runnable
  • POST /my_runnable/batch - 对一批输入调用 runnable
  • POST /my_runnable/stream - 对单个输入调用并流式传输输出
  • POST /my_runnable/stream_log - 对单个输入调用并流式传输输出，包括生成时中间步骤的输出
  • POST /my_runnable/astream_events - 对单个输入调用并流式传输生成的事件，包括来自中间步骤的事件。
  • GET /my_runnable/input_schema - runnable 输入的 json schema
  • GET /my_runnable/output_schema - runnable 输出的 json schema
  • GET /my_runnable/config_schema - runnable 配置的 json schema

这些端点与 LangChain Expression Language 接口 匹配 -- 请参考此文档了解更多详细信息。

Playground​

您可以在 /my_runnable/playground/ 找到 runnable 的 playground 页面。这公开了一个简单的 UI，用于 配置 和调用您的 runnable，并具有流式输出和中间步骤。

Widgets​

playground 支持 widgets，可用于使用不同的输入测试您的 runnable。有关更多详细信息，请参阅下面的 widgets 部分。

共享​

此外，对于可配置的 runnables，playground 将允许您配置 runnable 并共享带有配置的链接

聊天 playground​

LangServe 还支持一个以聊天为中心的 playground，该 playground 选择加入并在 /my_runnable/playground/ 下使用。与通用 playground 不同，仅支持某些类型的 runnables - runnable 的输入模式必须是 dict，其中包含：

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

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

runnable 还必须返回 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"

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

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

旧版 Chains​

LangServe 既适用于 Runnables（通过 LangChain Expression Language 构建）也适用于旧版 chains（继承自 Chain）。但是，旧版 chains 的某些输入模式可能不完整/不正确，从而导致错误。这可以通过更新 LangChain 中这些 chains 的 input_schema 属性来修复。如果您遇到任何错误，请在此 repo 上打开一个 issue，我们将努力解决它。

部署​

部署到 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 Repo

Deploy on Railway

Pydantic​

LangServe>=0.3 完全支持 Pydantic 2。

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

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

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

高级​

处理身份验证​

如果您需要为您的服务器添加身份验证，请阅读 Fast API 关于 dependencies 和 security 的文档。

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

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

如果您不确定自己在做什么，可以尝试使用现有的解决方案 Auth0。

使用 add_routes​

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

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

或者，您可以使用 FastAPI 的 middleware。

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

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

每个用户

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

使用 APIHandler​

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

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

这需要更多的工作，但可以让您完全控制端点定义，因此您可以执行所需的任何自定义逻辑进行身份验证。

文件​

LLM 应用程序通常处理文件。可以构建不同的架构来实现文件处理；在较高层面上

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

您应该确定适合您应用程序的架构。

目前，要按值将文件上传到 runnable，请使用 base64 编码文件（尚不支持 multipart/form-data）。

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

请记住，您始终可以通过引用（例如，s3 url）上传文件，或将它们作为 multipart/form-data 上传到专用端点。

自定义输入和输出类型​

输入和输出类型在所有 runnables 上定义。

您可以通过 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 模型而不是等效的 dict 表示形式，请从 CustomUserType 继承。

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

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

Playground Widgets​

playground 允许您从后端为您的 runnable 定义自定义 widgets。

这里有一些示例

描述链接
Widgets 可以与 playground 一起使用的不同小部件（文件上传和聊天）服务器， 客户端
Widgets 用于 LangServe playground 的文件上传小部件。服务器， 客户端

Schema​

  • widget 在字段级别指定，并作为输入类型的 JSON schema 的一部分提供
  • widget 必须包含一个名为 type 的键，其值是众所周知的 widgets 列表之一
  • 其他 widget 键将与描述 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;
};

可用 Widgets​

现在用户只能手动指定两种小部件

  1. 文件上传小部件
  2. 聊天记录小部件

请参阅下文以获取关于这些小部件的更多信息。

游乐场 UI 上的所有其他小部件都由 UI 基于 Runnable 的配置模式自动创建和管理。当您创建可配置的 Runnables 时，游乐场应创建适当的小部件供您控制其行为。

文件上传小部件​

允许在 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")

此页是否对您有帮助？