🦜️🏓 LangServe

🚩 我们将发布 LangServe 的托管版本，以便一键部署 LangChain 应用程序。在此注册加入等待列表。

概述

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

该库集成了 FastAPI 并使用 pydantic 进行数据验证。

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

特性

从 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

限制

目前不支持服务器发起的事件的客户端回调
当使用 Pydantic V2 时，将不会生成 OpenAPI 文档。FastAPI 不支持混合使用 pydantic v1 和 v2 命名空间。更多细节请参见下面的章节。

托管的 LangServe

我们将发布 LangServe 的托管版本，以便一键部署 LangChain 应用程序。在此注册加入等待列表。

安全性

版本 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 中定义可运行对象。转到 server.py 并编辑

add_routes(app. NotImplemented)

3. 使用 `poetry` 添加第三方包（例如 langchain-openai、langchain-anthropic、langchain-mistral 等）

poetry add [package-name] // 例如 `poetry add langchain-openai`

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

export OPENAI_API_KEY="sk-..."

5. 启动您的应用

poetry run langchain serve --port=8100

示例

使用LangChain 模板快速启动您的 LangServe 实例。

有关更多示例，请参见模板index 或 examples 目录。

描述	链接
LLMs 最小示例，保留了 OpenAI 和 Anthropic 聊天模型。使用异步，支持批处理和流式处理。	server, client
Retriever 简单的服务器，将检索器公开为可运行的程序。	server, client
Conversational Retriever 通过 LangServe 公开的对话检索器	server, client
Agent 基于OpenAI 工具的无对话历史代理	server, client
Agent 基于OpenAI 工具的有对话历史代理	server, client
带消息历史的可运行程序用于在后端实现持久化的聊天，由客户端提供的 `session_id` 作为键。	server, client
带消息历史的可运行程序用于在后端实现持久化的聊天，由客户端提供的 `conversation_id` 和 `user_id`（有关实现 `user_id` 的身份验证，请参见 Auth）作为键。	server, client
可配置的可运行程序用于创建支持运行时配置索引名称的检索器。	server, client
可配置的可运行展示了可配置字段和可配置的替代方案。	服务器端, 客户端
APIHandler 展示了如何使用 `APIHandler` 而不是 `add_routes`。这为开发人员提供了更多灵活性来定义端点。与所有 FastAPI 模式兼容，但需要更多的工作。	服务器端
LCEL 示例使用 LCEL 操作字典输入的示例。	服务器端, 客户端
使用 `add_routes` 进行身份验证：可以应用于与应用相关的所有端点的简单身份验证。(单独使用时不适用于实现每个用户逻辑。)	服务器端
使用 `add_routes` 进行身份验证：基于路径依赖的简单身份验证机制。(单独使用时不适用于实现每个用户逻辑。)	服务器端
使用 `add_routes` 进行身份验证：为使用每个请求配置修改器的端点实现每个用户逻辑和身份验证。(注意：目前不与 OpenAPI 文档集成。)	服务器端, 客户端
使用 `APIHandler` 进行身份验证：实现每个用户逻辑和身份验证，展示如何仅在用户拥有的文档中搜索。	服务器端, 客户端
小部件不同的小部件，可与 playground（文件上传和聊天）一起使用。	服务器端
小部件用于 LangServe playground 的文件上传小部件。	服务器端, 客户端

示例应用

服务器

以下是一个部署 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 服务器",
    version="1.0",
    description="使用 Langchain 的 Runnable 接口的简单 API 服务器",
)
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("告诉我一个关于 {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
# 设置所有启用 CORS 的来源
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
    expose_headers=["*"],
)

文档

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

⚠️ 如果使用 pydantic v2，将不会为 invoke、batch、stream、stream_log 生成文档。请参阅下面的 Pydantic 部分获取更多详细信息。

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("http://localhost:8000/openai/")
anthropic = RemoteRunnable("http://localhost:8000/anthropic/")
joke_chain = RemoteRunnable("http://localhost:8000/joke/")
joke_chain.invoke({"topic": "parrots"})
# 或者异步
await joke_chain.ainvoke({"topic": "parrots"})
prompt = [
    SystemMessage(content='Act like either a cat or a parrot.'),
    HumanMessage(content='Hello!')
]
# 支持 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}")]
)
# 可以定义自定义链
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: `http://localhost:8000/joke/`,
});
const result = await chain.invoke({
  topic: "cats",
});

使用 requests 的 Python 代码：

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

您也可以使用 curl：

curl --location --request POST 'http://localhost: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 表达式语言接口相匹配 --

请参考此文档以获取更多详细信息。

Playground

您可以在 /my_runnable/playground/ 找到一个可运行项的游乐场页面。这

提供了一个简单的 UI

来配置

并调用您的可运行项，具有流式输出和中间步骤。

小部件

游乐场支持小部件，可用于使用不同输入测试您的

可运行项。有关更多详细信息，请参阅下面的小部件部分。

共享

此外，对于可配置的可运行项，游乐场将允许您配置

可运行项并共享带有配置的链接：

聊天游乐场

LangServe 还支持一个聊天重点的游乐场，可以选择并在 /my_runnable/playground/ 下使用。

与一般游乐场不同，仅支持某些类型的可运行项 - 可运行项的输入模式必须为

一个 dict，其中包含以下内容之一：

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

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

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

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

# 声明一个对话链
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "你是一个乐于助人、专业的助手，名叫 Cob。"),
        MessagesPlaceholder(variable_name="messages"),
    ]
)
chain = prompt | ChatAnthropic(model="claude-2")
class InputChat(BaseModel):
    """聊天端点的输入。"""
    messages: List[Union[HumanMessage, AIMessage, SystemMessage]] = Field(
        ...,
        description="表示当前对话的聊天消息。",
    )
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 可以与 Runnables（通过 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 Container Apps（无服务器）部署到 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 存储库

Pydantic

LangServe 对 Pydantic 2 提供支持，但有一些限制。

使用 Pydantic V2 时，invoke/batch/stream/stream_log 的 OpenAPI 文档将不会生成。
Fast API 不支持 [混合使用 pydantic v1 和 v2 命名空间]。
LangChain 在 Pydantic v2 中使用 v1 命名空间。请阅读
以下指南以确保与 LangChain 的兼容性

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

高级

处理身份验证

如果需要为服务器添加身份验证，请阅读 Fast API 的文档

有关依赖项

和安全性。

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

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

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

使用 add_routes

如果您使用 add_routes，请参阅

此处的示例。

描述	链接
使用 `add_routes` 进行身份验证：简单的身份验证机制，可应用于与应用程序相关的所有端点。（单独使用时无法实现每个用户的逻辑。）	server
使用 `add_routes` 进行身份验证：基于路径依赖的简单身份验证机制。（单独使用时无法实现每个用户的逻辑。）	server
使用 `add_routes` 进行身份验证：为使用每个请求配置修饰符的端点实现每个用户的逻辑和身份验证。（注意：目前不与 OpenAPI 文档集成。）	server, client

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

使用全局依赖项和路径依赖项的优点是身份验证将在 OpenAPI 文档页面中得到正确支持，但是

这些不足以实现每个用户的逻辑（例如，创建一个只能在用户拥有的文档中搜索的应用程序）。

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

每个用户

如果您需要与用户相关的授权或逻辑，

在使用 add_routes 时指定 per_req_config_modifier。使用一个可调用对象来接收

原始的 Request 对象，并从中提取相关信息进行身份验证和

授权目的。

使用 APIHandler

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

描述	链接
使用 `APIHandler` 进行身份验证：实现每个用户的逻辑和身份验证，演示如何仅在用户拥有的文档中进行搜索。	server, client
APIHandler 演示如何使用 `APIHandler` 而不是 `add_routes`。这为开发人员更灵活地定义端点提供了更多选择。与所有 FastAPI 模式配合使用效果很好，但需要更多的工作。	server, client

这需要更多的工作，但可以完全控制端点定义，因此

您可以根据需要进行自定义身份验证逻辑。

文件

LLM 应用程序通常涉及文件处理。可以使用不同的架构

来实现文件处理；在高层次上：

文件可以通过专用端点上传到服务器，并使用
单独的端点进行处理
文件可以通过值（文件的字节）或引用（例如，指向文件内容的 s3 URL）
进行上传
处理端点可以是阻塞或非阻塞的
如果需要进行重要的处理，可以将处理工作转移到专用的

进程池

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

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

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

请记住，您始终可以通过引用（例如s3网址）上传文件，或将它们作为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:
    """应接受整数但接受任何类型的错误函数。"""
    return x + 1
runnable = RunnableLambda(func).with_types(
    input_type=int,
)
add_routes(app, runnable)

自定义用户类型

如果希望数据反序列化为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:
    """期望一个Foo类型（pydantic模型）的示例函数。"""
    assert isinstance(foo, Foo)
    return foo.bar
add_routes(app, RunnableLambda(func), path="/foo")

游乐场小部件

游乐场允许您从后端为可运行文件定义自定义小部件。

以下是一些示例：

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

模式

小部件在字段级别指定，并作为输入类型的JSON模式的一部分进行传送
小部件必须包含一个名为type的键，其值是众所周知的小部件列表之一
其他小部件键将与描述JSON对象中路径的值相关联

type JsonPath = number | string | (number | string)[];
type NameSpacedPath = { title: string; path: JsonPath }; // 使用title模拟json模式，但可以使用命名空间
type OneOfPath = { oneOf: JsonPath[] };
type Widget = {
  type: string; // 一些众所周知的类型（例如，base64file，chat等）
  [key: string]: JsonPath | NameSpacedPath | OneOfPath;
};

可用小部件

目前用户可以手动指定的小部件仅有两种：

文件上传小部件
聊天历史小部件

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

游乐场UI上的所有其他小部件都是根据可运行文件的配置模式自动创建和管理的。当您创建可配置的可运行文件时，游乐场应为您创建适当的小部件以控制行为。

文件上传小部件

允许在UI游乐场中创建文件上传输入。

这些文件以 base64 编码字符串的形式上传。这里是完整的示例。

代码片段：

try:
    from pydantic.v1 import Field
except ImportError:
    from pydantic import Field
from langserve import CustomUserType
# 注意：继承自 CustomUserType 而不是 BaseModel，否则服务器会将其解码为字典而不是 pydantic 模型。
class FileProcessingRequest(CustomUserType):
    """包含 base64 编码文件的请求。"""
    # extra 字段用于为 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]:
    """将输入格式化为消息列表。"""
    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")
class MessageListInput(BaseModel):
    """聊天端点的输入。"""
    messages: List[Union[HumanMessage, AIMessage]] = Field(
        ...,
        description="代表当前对话的聊天消息。",
        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")

禁用：下面的代码将禁用链路的 playground。

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

🦜️🏓 LangServe

概述​

特性​

限制​

托管的 LangServe​

安全性​

安装​

LangChain CLI 🛠️​

设置​

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

2. 在 add_routes 中定义可运行对象。转到 server.py 并编辑​

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

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

5. 启动您的应用​

示例​

示例应用​

服务器​

文档​

客户端​

端点​

Playground​

小部件​

共享​

聊天游乐场​

旧版链​

部署​

部署到 AWS​

部署到 Azure​

部署到 GCP​

社区贡献​

部署到 Railway​

Pydantic​

高级​

处理身份验证​

使用 add_routes​

使用 APIHandler​

文件​

进程池​

自定义输入和输出类型​

自定义用户类型​

游乐场小部件​

模式​

可用小部件​

文件上传小部件​

聊天小部件​

启用/禁用端点（LangServe >=0.0.33）​

Was this page helpful?

You can leave detailed feedback on GitHub.

概述

特性

限制