Skip to main content

Flyte (飞行)

Flyte 是一个开源的编排器,用于构建生产级的数据和机器学习流水线。 它以可扩展性和可重现性为基础,利用 Kubernetes 作为其底层平台。

本笔记本的目的是演示将 FlyteCallback 集成到 Flyte 任务中,使您能够有效地监视和跟踪您的 LangChain 实验。

安装和设置

  • 运行命令 pip install flytekit 安装 Flytekit 库。
  • 运行命令 pip install flytekitplugins-envd 安装 Flytekit-Envd 插件。
  • 运行命令 pip install langchain 安装 LangChain。
  • 在您的系统上安装 Docker

Flyte 任务

Flyte 任务 是 Flyte 的基础构建块。 要执行 LangChain 实验,您需要编写定义特定步骤和操作的 Flyte 任务。

注意:入门指南 提供了关于在本地安装 Flyte 并运行初始 Flyte 流水线的详细逐步说明。

首先,导入必要的依赖项以支持您的 LangChain 实验。

import os

from flytekit import ImageSpec, task
from langchain.agents import AgentType, initialize_agent, load_tools
from langchain.callbacks import FlyteCallbackHandler
from langchain.chains import LLMChain
from langchain.chat_models import ChatOpenAI
from langchain.prompts import PromptTemplate
from langchain.schema import HumanMessage

设置必要的环境变量以利用 OpenAI API 和 Serp API:

# 设置 OpenAI API 密钥
os.environ["OPENAI_API_KEY"] = "<your_openai_api_key>"

# 设置 Serp API 密钥
os.environ["SERPAPI_API_KEY"] = "<your_serp_api_key>"

<your_openai_api_key><your_serp_api_key> 替换为您从 OpenAI 和 Serp API 获取的相应 API 密钥。

为了确保流水线的可重现性,Flyte 任务被容器化。 每个 Flyte 任务必须与一个镜像关联,该镜像可以在整个 Flyte 工作流中共享,也可以为每个任务单独提供。

为了简化为每个 Flyte 任务提供所需依赖项的过程,您可以初始化一个 ImageSpec 对象。 这种方法会自动触发 Docker 构建,减轻用户手动创建 Docker 镜像的需求。

custom_image = ImageSpec(
name="langchain-flyte",
packages=[
"langchain",
"openai",
"spacy",
"https://github.com/explosion/spacy-models/releases/download/en_core_web_sm-3.5.0/en_core_web_sm-3.5.0.tar.gz",
"textstat",
"google-search-results",
],
registry="<your-registry>",
)

您可以灵活选择将 Docker 镜像推送到您喜欢的注册表。 Docker HubGitHub Container Registry (GHCR) 是一个方便的开始选项。

一旦您选择了一个注册表,您可以继续创建将 LangChain 指标记录到 Flyte Deck 的 Flyte 任务。

以下示例演示了与 OpenAI LLM、链和带有工具的代理相关的任务:

LLM

@task(disable_deck=False, container_image=custom_image)
def langchain_llm() -> str:
llm = ChatOpenAI(
model_name="gpt-3.5-turbo",
temperature=0.2,
callbacks=[FlyteCallbackHandler()],
)
return llm([HumanMessage(content="Tell me a joke")]).content

Chain

@task(disable_deck=False, container_image=custom_image)
def langchain_chain() -> list[dict[str, str]]:
template = """You are a playwright. Given the title of play, it is your job to write a synopsis for that title.
Title: {title}
Playwright: This is a synopsis for the above play:"""
llm = ChatOpenAI(
model_name="gpt-3.5-turbo",
temperature=0,
callbacks=[FlyteCallbackHandler()],
)
prompt_template = PromptTemplate(input_variables=["title"], template=template)
synopsis_chain = LLMChain(
llm=llm, prompt=prompt_template, callbacks=[FlyteCallbackHandler()]
)
test_prompts = [
{
"title": "documentary about good video games that push the boundary of game design"
},
]
return synopsis_chain.apply(test_prompts)

Agent

@task(disable_deck=False, container_image=custom_image)
def langchain_agent() -> str:
llm = OpenAI(
model_name="gpt-3.5-turbo",
temperature=0,
callbacks=[FlyteCallbackHandler()],
)
tools = load_tools(
["serpapi", "llm-math"], llm=llm, callbacks=[FlyteCallbackHandler()]
)
agent = initialize_agent(
tools,
llm,
agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
callbacks=[FlyteCallbackHandler()],
verbose=True,
)
return agent.run(
"Who is Leonardo DiCaprio's girlfriend? Could you calculate her current age and raise it to the power of 0.43?"
)

这些任务是在 Flyte 中运行您的 LangChain 实验的起点。

在 Kubernetes 上执行 Flyte 任务

要在配置的 Flyte 后端上执行 Flyte 任务,请使用以下命令:

pyflyte run --image <your-image> langchain_flyte.py langchain_llm

此命令将在 Flyte 后端上启动 langchain_llm 任务的执行。您可以以类似的方式触发其余两个任务。

指标将显示在 Flyte UI 上,如下所示:

LangChain LLM