LangChain 的 Steamship Python 客户端库 (🦜️🔗)

Steamship 是构建、部署和使用全生命周期语言 AI 的最快方式。

本仓库包含用于 Steamship 的 LangChain 适配器，使 LangChain 开发者能够快速在 Steamship 上部署其应用，从而自动获得：

• 生产就绪的 API 端点
• 跨依赖项/后端的水平扩展
• 应用状态的持久化存储（包括缓存）
• 内置的身份验证与授权支持
• 多租户支持
• 与其他 Steamship 技能的无缝集成（例如音频转录）
• 使用情况指标与日志记录
• 以及更多……

有关 Steamship 和 LangChain 的更多信息，请访问我们的 网站。

安装

通过 pip 安装：

pip install steamship-langchain

适配器

目前提供以下初步支持（更多功能即将推出）：

• LLM
  • 提供了针对 Steamship OpenAI 集成的适配器 (steamship_langchain.llms.OpenAI)
  • 提供了通过 Steamship 键值存储实现 LLM 调用缓存的适配器 (SteamshipCache)
• 回调
  • 提供了一个使用 Python logging 模块记录事件的回调 (steamship_langchain.callbacks.LoggingCallbackHandler)。此回调可与 ship logs 配合使用，在部署时访问详细日志。
• 文档加载器
  • 提供了一个将 Steamship 文件导出为 LangChain 文档的适配器 (steamship_langchain.document_loaders.SteamshipLoader)
• 工具
  • 搜索：
    • 提供了针对 Steamship SERPAPI 集成的适配器 (SteamshipSERP)
• 记忆
  • 对话历史 (steamship_langchain.memory.ChatMessageHistory)
• 向量存储
  • 提供了一个持久化向量存储的适配器 (steamship_langchain.vectorstores.SteamshipVectorStore)
• 文本分割器
  • 提供了一个基于 AST 的 Python 代码分割器 (steamship_langchain.python_splitter.PythonCodeSplitter)。该分割器在以函数定义为边界拆分代码的同时，还能为代码片段提供额外上下文信息（父类）。
• 其他实用工具
  • 将数据导入 Steamship
    • 为了充分利用 Steamship 的持久化存储能力，我们提供了一系列初始加载工具，适用于多种数据源，包括：
      • 文本文件：steamship_langchain.file_loaders.TextFileLoader
      • 目录：steamship_langchain.file_loaders.DirectoryLoader
      • GitHub 仓库：steamship_langchain.file_loaders.GitHubRepositoryLoader
      • Sphinx 文档站点：steamship_langchain.file_loaders.SphinxSiteLoader（以及其他）
      • YouTube 视频：steamship_langchain.file_loaders.YouTubeFileLoader
      • 各种文本和图像格式：steamship_langchain.file_loaders.UnstructuredFileLoader

📖 文档

请参阅我们的 文档，了解关于以下内容的完整说明：

• 入门指南（安装、环境设置、简单示例）
• 操作示例（演示、集成、辅助函数）

示例用法

以下是几个在 Steamship 上使用 LangChain 的示例：

• 基础提示
• 自问自答与搜索
• 聊天机器人
• 音频摘要
• 带来源的问题回答

这些示例使用临时工作空间，以便在实验过程中进行完全清理。 工作空间 在 Steamship 中提供了租户隔离单元。 对于生产用途，可以通过 Steamship(workspace_handle="my_workspace") 创建并获取持久化工作空间。

注意 这些示例省略了 import 语句。完整的源代码请参考 examples/ 目录。

注意 客户端示例假定用户已拥有 Steamship API 密钥，并且该密钥已暴露于环境变量中（详见：API 密钥）。

基础提示

使用 Steamship LLM 集成的基本提示示例（完整源码：examples/greeting）

服务器片段

from steamship_langchain.llms import OpenAI

@post("greet")
def greet(self, user: str) -> str:
    prompt = PromptTemplate(
      input_variables=["user"],
      template=
      "为用户 {user} 创作一条欢迎消息。感谢他们使用 LangChain 在 Steamship 上运行应用。鼓励他们在准备就绪时通过 `ship it` 部署应用。",
    )
    llm = OpenAI(client=self.client, temperature=0.8)
    return llm(prompt.format(user=user))

客户端片段

with Steamship.temporary_workspace() as client:
    api = client.use("my-langchain-app")
    while True:
        name = input("姓名：")
        print(f'{api.invoke("/greet", user=name).strip()}\n')

自我提问与搜索

使用 Steamship GPT 和 SERP Tool 插件执行 LangChain 的 self-ask-with-search 代理（完整源码：examples/self-ask-with-search）

服务器片段

from steamship_langchain.llms import OpenAI

@post("/self_ask_with_search")
def self_ask_with_search(self, query: str) -> str:
    llm = OpenAI(client=self.client, temperature=0.0, cache=True)
    serp_tool = SteamshipSERP(client=self.client, cache=True)
    tools = [Tool(name="Intermediate Answer", func=serp_tool.search)]
    self_ask_with_search = initialize_agent(tools, llm, agent="self-ask-with-search", verbose=False)
    return self_ask_with_search.run(query)

客户端片段

with Steamship.temporary_workspace() as client:
    api = client.use("my-langchain-app")
    query = "双城队上一次赢得世界大赛时，谁是总统？"
    print(f"查询：{query}")
    print(f"答案：{api.invoke('/self_ask_with_search', query=query)}")

聊天机器人

在 Steamship 中使用 LangChain 实现一个基础的聊天机器人（类似于 ChatGPT）（完整源码：examples/chatbot）。

注意 整个聊天机器人的对话记录将在 Steamship 工作空间的生命周期内持续保存。

服务器片段

from langchain.memory import ConversationBufferWindowMemory

from steamship_langchain.llms import OpenAIChat
from steamship_langchain.memory import ChatMessageHistory

@post("/send_message")
def send_message(self, message: str, chat_history_handle: str) -> str:
  chat_memory = ChatMessageHistory(client=self.client, key=chat_history_handle)
  mem = ConversationBufferWindowMemory(chat_memory=chat_memory, k=2)
  chatgpt = LLMChain(
    llm=OpenAIChat(client=self.client, temperature=0),
    prompt=CHATBOT_PROMPT,
    memory=mem,
  )

  return chatgpt.predict(human_input=message)

客户端片段

with Steamship.temporary_workspace() as client:
    api = client.use("my-langchain-app")
    session_handle = "foo-user-session-1234"
    while True:
        msg = input("你：")
        print(f"AI：{api.invoke('/send_message', message=msg，chat_history_handle=session_handle)}")

异步链式处理音频摘要

Steamship 目前尚未将音频转录支持视为完全成熟的生产环境。我们正在努力实现大规模音频转录的支持，但在尝试此功能时，您可能会遇到一些现有问题。

本示例展示了如何使用 LangChain 处理通过 Steamship 的语音转文本插件获取的音频转录内容（完整源码：examples/summarize-audio）。

该示例简要介绍了任务系统（以及用于链式调用的任务依赖关系）。这里我们使用了 task.wait() 风格的轮询方式，但也可以使用基于时间的 task.refresh() 等其他轮询方式。

服务器片段

from steamship_langchain.llms import OpenAI

@post("summarize_file")
def summarize_file(self, file_handle: str) -> str:
    file = File.get(self.client, handle=file_handle)
    text_splitter = CharacterTextSplitter()
    texts = []
    for block in file.blocks:
        texts.extend(text_splitter.split_text(block.text))
    docs = [Document(page_content=t) for t in texts]
    llm = OpenAI(client=self.client, cache=True)
    chain = load_summarize_chain(llm, chain_type="map_reduce")
    return chain.run(docs)

@post("summarize_audio_file")
def summarize_audio_file(self, audio_file_handle: str) -> Task[str]:
    transcriber = self.client.use_plugin("whisper-s2t-blockifier")
    audio_file = File.get(self.client, handle=audio_file_handle)
    transcribe_task = audio_file.blockify(plugin_instance=transcriber.handle)
    return self.invoke_later("summarize_file", wait_on_tasks=[transcribe_task], arguments={"file_handle": audio_file.handle})

客户端片段

churchill_yt_url = "https://www.youtube.com/watch?v=MkTw3_PmKtc"

with Steamship.temporary_workspace() as client:
    api = client.use("my-langchain-app")
    yt_importer = client.use_plugin("youtube-file-importer")
    import_task = File.create_with_plugin(client=client,
                                         plugin_instance=yt_importer.handle, 
                                         url=churchill_yt_url)
    import_task.wait()
    audio_file = import_task.output
    
    summarize_task_response = api.invoke("/summarize_audio_file", audio_file_handle=audio_file.handle)
    summarize_task = Task(client=client, **summarize_task_response)
    summarize_task.wait()
    
    if summarize_task.state == TaskState.succeeded:
      summary = base64.b64decode(summarize_task.output).decode("utf-8")
      print(f"摘要：{summary.strip()}")

带来源的答案问答（嵌入）

提供了一个使用 Steamship 管理嵌入并为 LangChain 代理提供支持的基本示例，用于实现带来源的答案问答功能（完整源码：examples/qa_with_sources）

注意 嵌入将在工作区的生命周期内持续保存。

服务器代码片段

from steamship_langchain.llms import OpenAI

def __init__(self, **kwargs):
    super().__init__(**kwargs)
    langchain.llm_cache = SteamshipCache(self.client)
    self.llm = OpenAI(client=self.client, temperature=0, cache=True, max_words=250)
    # 创建一个持久化的嵌入存储
    self.index = SteamshipVectorStore(
        client=self.client, index_name="qa-demo", embedding="text-embedding-ada-002"
    )

@post("index_file")
def index_file(self, file_handle: str) -> bool:
    text_splitter = CharacterTextSplitter(chunk_size=250, chunk_overlap=0)
    file = File.get(self.client, handle=file_handle)
    texts = [text for block in file.blocks for text in text_splitter.split_text(block.text)]
    metadatas = [{"source": f"{file.handle}-offset-{i * 250}"} for i, text in enumerate(texts)]

    self.index.add_texts(texts=texts, metadatas=metadatas)
    return True

@post("search_embeddings")
def search_embeddings(self, query: str, k: int) -> List[SearchResult]:
    """返回嵌入索引中最接近的 `k` 个条目。"""
    search_results = self.index.search(query, k=k)
    search_results.wait()
    items = search_results.output.items
    return items


@post("/qa_with_sources")
def qa_with_sources(self, query: str) -> Dict[str, Any]:
    chain = VectorDBQAWithSourcesChain.from_chain_type(
        OpenAI(client=self.client, temperature=0),
        chain_type="map_reduce",
        vectorstore=self.index,
    )

    return chain({"question": query}, return_only_outputs=False)

客户端代码片段

with Steamship.temporary_workspace() as client:
    api = client.use("my-langchain-app")
    
    # 上传国情咨文演讲稿
    with open("state_of_the_union.txt") as f:
        sotu_file = File.create(client, blocks=[Block(text=f.read())])

    # 进行嵌入
    api.invoke("/index_file", file_handle=sotu_file.handle)

    # 发出查询
    query = "总统对布雷耶大法官说了什么？"
    response = api.invoke("/qa_with_sources", query=query)
    print(f"答案：{response['result'].strip()}")

API 密钥

Steamship API 密钥可让您访问我们的 SDK，用于 AI 模型，包括 OpenAI、GPT、Cohere、Whisper 等。

在此处获取您的免费 API 密钥：https://steamship.com/account/api。

获得 API 密钥后，您可以：

• 设置客户端的环境变量 STEAMSHIP_API_KEY
• 直接通过 Steamship(api_key=) 或 Steamship.tempory_workspace(api_key=) 传递密钥。

或者，您也可以运行 ship login，它将指导您完成环境设置。

在 Steamship 上部署

在 Steamship 上部署 LangChain 应用非常简单：只需执行 ship it 命令即可。

从您的包目录（即 api.py 所在的目录）中，您可以运行 ship it 命令来生成清单文件，并将您的包推送到 Steamship。随后，您可以使用 Steamship SDK 根据需要在工作区中创建您的包实例。

有关部署和工作区的更多信息，请参阅我们的文档。

反馈与支持

您对本包或对 Steamship 整体有任何反馈吗？

我们非常期待您的意见。请联系我们：hello@steamship.com。

Steamship-LangChain 快速上手指南

本指南帮助中国开发者快速将 LangChain 应用部署到 Steamship 平台，以获得生产级 API、自动扩缩容、持久化存储及多租户支持。

环境准备

在开始之前，请确保满足以下前置条件：

1. Python 环境：建议使用 Python 3.8 或更高版本。
2. Steamship 账号与 API Key：
   • 访问 Steamship 官网 注册账号。
   • 获取 API Key 并配置到环境变量中：

     export STEAMSHIP_API_KEY=your_api_key_here
     

3. 网络环境：由于 Steamship 服务位于海外，建议确保网络通畅以便连接其云端服务。

安装步骤

通过 pip 安装官方客户端库：

pip install steamship-langchain

提示：如果下载速度较慢，可尝试使用国内镜像源加速安装：

pip install steamship-langchain -i https://pypi.tuna.tsinghua.edu.cn/simple

基本使用

以下示例展示如何创建一个最简单的 LangChain 应用，该应用接收用户名称并生成欢迎语。此示例利用了 Steamship 的临时工作区（Temporary Workspace）进行快速测试。

1. 服务端代码 (Server Snippet)

创建一个继承自 PackageService 的类，定义一个 POST 接口 /greet，调用 Steamship 集成的 OpenAI 模型。

from steamship import PackageService, post
from langchain.prompts import PromptTemplate
from steamship_langchain.llms import OpenAI

class MyLangChainApp(PackageService):
    @post("greet")
    def greet(self, user: str) -> str:
        # 定义提示模板
        prompt = PromptTemplate(
          input_variables=["user"],
          template=
          "Create a welcome message for user {user}. Thank them for running their LangChain app on Steamship. "
          "Encourage them to deploy their app via `ship it` when ready.",
        )
        
        # 初始化 LLM (使用 Steamship 客户端)
        llm = OpenAI(client=self.client, temperature=0.8)
        
        # 执行生成
        return llm(prompt.format(user=user))

2. 客户端调用 (Client Snippet)

在本地脚本中启动临时工作区，部署上述应用并进行交互调用。

from steamship import Steamship

# 启动临时工作区（实验结束后自动清理）
with Steamship.temporary_workspace() as client:
    # 部署并使用应用 (假设上述代码已打包为 my-langchain-app)
    # 在实际开发中，通常先运行 'ship deploy' 命令
    api = client.use("my-langchain-app")
    
    while True:
        name = input("Name: ")
        if not name:
            break
        # 调用远程接口
        response = api.invoke("/greet", user=name)
        print(f'{response.strip()}\n')

核心特性说明

• 持久化与扩展：上述代码中的 OpenAI 适配器自动支持缓存和后端扩展。
• 生产部署：实验完成后，可使用 ship deploy 命令将应用部署为永久运行的生产级 API，无需修改代码即可获得身份验证、日志记录和监控功能。
• 更多组件：除了 LLM，该库还支持 SteamshipVectorStore (向量存储)、SteamshipSERP (搜索工具) 以及多种文件加载器（如 GitHub 仓库、YouTube 视频等）。