跳到主要內容

🦜️🏓 LangServe

Release Notes Downloads Open Issues

[!WARNING] 我們建議新專案使用 LangGraph Platform 而非 LangServe。

請參閱 LangGraph Platform 遷移指南 以取得更多資訊。

我們將持續接受社群針對 LangServe 提交的錯誤修正；然而，我們將不再接受新的功能貢獻。

概觀​

LangServe 協助開發人員將 LangChain 可執行物件和鏈 部署為 REST API。

此函式庫與 FastAPI 整合，並使用 pydantic 進行資料驗證。

此外，它提供了一個客戶端，可用於調用部署在伺服器上的可執行物件。JavaScript 客戶端可在 LangChain.js 中取得。

功能特色​

  • 輸入和輸出 schema 從您的 LangChain 物件自動推斷，並在每次 API 調用時強制執行，並提供豐富的錯誤訊息
  • API 文件頁面，包含 JSONSchema 和 Swagger (插入範例連結)
  • 高效的 /invoke、/batch 和 /stream 端點，支援單一伺服器上的多個並行請求
  • /stream_log 端點，用於串流來自您的鏈/代理程式的所有（或部分）中繼步驟
  • 新功能 自 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 中眾所周知的 primitives 協同工作。

如果您需要 LangGraph 的部署選項，您應該考慮 LangGraph Cloud (beta)，它將更適合部署 LangGraph 應用程式。

限制​

  • 尚不支援來自伺服器的事件的客戶端回調
  • LangServe <= 0.2.0 版本，當使用 Pydantic V2 時，將無法正確生成 OpenAPI 文件，因為 Fast API 不支援 混合 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 中定義 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 聊天模型。使用 async，支援批次處理和串流。伺服器、客戶端
Retriever 簡單的伺服器，將 retriever 作為 runnable 公開。伺服器、客戶端
Conversational Retriever 透過 LangServe 公開的 Conversational Retriever伺服器、客戶端
基於 OpenAI 工具 的Agent，不含對話歷史記錄伺服器、客戶端
基於 OpenAI 工具 的 Agent，包含對話歷史記錄伺服器、客戶端
RunnableWithMessageHistory 實作在後端持續保存的聊天記錄，並以客戶端提供的 session_id 作為鍵值。伺服器、客戶端
RunnableWithMessageHistory 實作在後端持續保存的聊天記錄，並以客戶端提供的 conversation_id 和 user_id 作為鍵值（請參閱 Auth 章節以正確實作 user_id）。伺服器、客戶端
Configurable Runnable 建立一個 retriever，支援在執行時配置索引名稱。伺服器、客戶端
Configurable Runnable 顯示可配置的欄位和可配置的替代方案。伺服器、客戶端
APIHandler 示範如何使用 APIHandler 而非 add_routes。這為開發人員提供了更大的彈性來定義端點。與所有 FastAPI 模式搭配良好，但需要花費更多力氣。伺服器
LCEL 範例 使用 LCEL 操作字典輸入的範例。伺服器、客戶端
使用 add_routes 的 Auth：簡單的身份驗證，可以應用於與應用程式相關聯的所有端點。（本身對於實作每個使用者的邏輯沒有用處。）伺服器
使用 add_routes 的 Auth：基於路徑依賴性的簡單身份驗證機制。（本身對於實作每個使用者的邏輯沒有用處。）伺服器
使用 add_routes 的 Auth：為使用每個請求配置修飾符的端點實作每個使用者的邏輯和身份驗證。（注意：目前，不與 OpenAPI 文件整合。）伺服器、客戶端
使用 APIHandler 的 Auth：實作每個使用者的邏輯和身份驗證，示範如何僅在使用者擁有的文件中搜尋。伺服器、客戶端
Widgets 可與 playground 搭配使用的不同 widget（檔案上傳和聊天）伺服器
Widgets 用於 LangServe playground 的檔案上傳 widget。伺服器、客戶端

範例應用程式​

伺服器​

這是一個伺服器，部署了 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://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，選擇加入並在 /my_runnable/playground/ 下使用。與一般的 playground 不同，僅支援某些類型的 runnables - runnable 的輸入 schema 必須是 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"

以下是一個開啟上述兩個選項的範例

注意：如果您啟用公開追蹤連結，您的鏈的內部結構將會公開。我們建議僅在示範或測試中使用此設定。

傳統鏈​

LangServe 適用於 Runnables（透過 LangChain Expression Language 構建）和傳統鏈（繼承自 Chain）。但是，傳統鏈的一些輸入 schema 可能不完整/不正確，導致錯誤。這可以透過更新 LangChain 中這些鏈的 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 Container Apps (Serverless) 部署到 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 primitives 將身份驗證邏輯連接到 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 定義自訂 widget。

以下是一些範例

描述連結
Widgets 可與 playground 搭配使用的不同 widget（檔案上傳和聊天）伺服器、客戶端
Widgets 用於 LangServe playground 的檔案上傳 widget。伺服器、客戶端

Schema​

  • widget 在欄位層級指定，並作為輸入類型的 JSON schema 的一部分發布
  • widget 必須包含一個名為 type 的鍵，其值為一組眾所周知的 widget 之一
  • 其他 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​

目前使用者只能手動指定兩個 widget

  1. 檔案上傳 Widget
  2. 聊天歷史記錄 Widget

請參閱以下關於這些 widget 的更多資訊。

playground UI 上的所有其他 widget 都是由 UI 基於 Runnable 的 config schema 自動建立和管理的。當您建立 Configurable Runnables 時，playground 應為您建立適當的 widget 來控制行為。

檔案上傳 Widget​

允許在 UI playground 中建立檔案上傳輸入，用於以 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

Widget 範例

聊天 Widget​

請查看 widget 範例。

若要定義聊天 widget，請確保您傳遞 "type": "chat"。

  • "input" 是 JSONPath，指向 Request 中包含新輸入訊息的欄位。
  • "output" 是 JSONPath，指向 Response 中包含新輸出訊息的欄位。
  • 如果應按原樣使用整個輸入或輸出（例如，如果輸出是聊天訊息列表），則不要指定這些欄位。

以下是一個程式碼片段

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

Widget 範例

您也可以直接將訊息列表指定為參數，如此程式碼片段所示

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

停用：以下程式碼將停用鏈的 playground

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

此頁面是否對您有幫助？