Langflow：含演示项目的指南

如何安装 Langflow？

我们可以免费使用 Langflow，既可以通过其官方网站的 Web 界面使用，也可以本地安装。

本教程将使用基于 Docker 的本地环境。您不需要懂 Docker 也能跟上。如果想进一步了解，请查看 这份 Docker 初学者指南和这门Docker 课程。

Docker 是一种工具，让我们在称为容器的隔离环境中运行应用，便于一致地开发、测试与部署软件。就我们的场景而言，使用 Docker 能为用 Python 构建自定义组件提供更大灵活性，例如可使用本地数据库，而无需连接外部服务。

Langflow 提供了可直接运行的预构建示例，因此我们无需自行搭建。

使用 Docker 在本地运行 Langflow

要在本地通过 Docker 运行 Langflow，请按照以下步骤：

1. 按照 此处的步骤安装 Docker。
2. 克隆或下载 Langflow 官方仓库。
3. 在 docker_example 文件夹中打开终端。
4. 运行命令 docker compose up。

现在，Langflow 应该已在本地运行。您应能在终端中看到如下信息：

我们可以访问终端显示的 URL 来打开应用，上例为 http://0.0.0.0:7860。

用 Langflow 探索 Simple Agent

本节我们将通过查看 Langflow 的简易 AI 代理模板来了解其工作方式。为此，请在首屏选择 "Simple Agent"。

这将创建如下所示的代理工作流：

该工作流的核心组件是中间的 "Agent" 组件：

该代理组件使用 OpenAI 的 gpt-4.1 模型。要使用它，我们需要粘贴 OpenAI API 密钥。如果您还没有，可以在这里创建。

在代理组件左侧，我们可以看到可用的输入和工具（见下图）。此处包含两个工具和一个输入：

• 一个聊天输入，用于将聊天界面中用户的消息作为输入。
• 一个计算器工具，用于执行计算。
• 一个 URL 工具，用于从指定 URL 抓取信息。

聊天输入意味着在执行工作流时，我们可以使用 Langflow 内置的聊天界面向模型发送消息。每个工具顶部都有一段描述。模型会据此决定是否调用该工具。

右侧为输出。在本例中只有一个输出组件，即聊天输出组件，表明代理的回答应发送到聊天中。

理解组件

在运行 Simple Agent 前，我们先用 URL 组件来理解组件的通用工作方式。复制一份该组件进行试验。

1. 选中 URL 组件。
2. 复制并粘贴。
3. 不要连接到任何其他组件。

在新组件顶部有个 "Tool Mode" 开关；将其关闭。

在 URLs 字段中填入例如 https://en.wikipedia.org/wiki/Data_science（数据科学的维基百科页面）。点击右上角的运行箭头执行该组件。

组件会执行并加载我们提供的 URL 内容。我们可以点击输出连接旁的检查器查看。每个图标代表不同的输出格式。

在幕后，每个节点都是一个 Python 脚本。稍后在构建自定义节点时我们会进一步了解。可点击顶部的 "Code" 按钮查看代码：

在原始的 Simple Agent 工作流中，URL 节点被配置为工具。这意味着它将由代理调用。在这种情况下，URL 并非显式给出，而是由代理自行提供。

执行 Simple Agent

回到 Simple Agent，我们可以点击右上角的 "Playground" 按钮来执行它。此操作会打开聊天界面，让我们与代理对话。

让我们请它总结数据科学的维基百科页面内容：

如果展开回答顶部的框，您将看到工作流中执行的各个步骤。

例如，我们可以看到它使用了 fetch_content_text() 函数，该函数定义在我们提供给代理的 URL 工具中。

我们可以通过让它执行一次计算来测试计算工具的使用。表达式会被解析并发送到该工具进行求值。相比只依赖 LLM，这能提供更可靠的答案。

在 Langflow 中用 Python 创建自定义组件

本节我们学习如何创建自定义组件。首先，点击左下角的 "New Custom Component" 按钮：

默认情况下，这会创建一个以文本值为输入并原样输出该文本的组件。默认输入是 "Hello, World!"，如果我们执行并检查输出，正是如此。

查看代码，我们会看到如下内容：

1. 这些字段是组件的配置值。描述非常重要，因为当组件作为工具提供给 AI 代理时，代理会据此判断何时使用该组件。documentation 字段是可选的组件文档 URL，本例无需关心。
2. inputs 列表指定组件可用的输入类型。本例包含一个名为 input_value 的 MessageTextInput 类输入，对应文本字段。名称必须是有效的 Python 变量名，因为它会成为类的属性。当组件被设置为工具时，这些输入会由 AI 代理自动填充。代理会使用 info 字段来确定输入应是什么。Langflow 支持多种输入类型。 这里有完整的支持输入列表。
3. 此部分定义输出。method 字段指定用于生成输出的函数。该函数可通过 self 访问输入。例如本例的输入名为 input_value，因此可通过 self.input_value 访问。返回值应为受支持的输出类中的某个类。

关于创建自定义组件的更多信息，请查阅 Langflow 组件官方文档。

下面是一个将两个数字相加的组件示例：

from langflow.custom import Component
from langflow.io import MessageTextInput, Output
from langflow.schema import Data

class AddNumbers(Component):
    display_name = "Add Numbers"
    description = "This component adds two numbers together"
    icon = "code"
    name = "AddNumbers"

    inputs = [
        IntInput(
            name="number1",
            display_name="First Number",
            info="The first number to add",
        ),
        IntInput(
            name="number2",
            display_name="Second Number",
            info="The second number to add",
        ),
    ]

    outputs = [
        Output(display_name="Output", name="output", method="add_numbers"),
    ]

    def add_numbers(self) -> Data:
        result = self.number1 + self.number2
        data = Data(value=result)
        return data技术支持

用 Langflow 构建语言导师

让我们构建一个 AI 代理，帮助我们在学习新语言时练习阅读能力。

以我的经验，学习一门新语言的方式之一是阅读。当然，起步阶段我们只认识少量单词，因此读书几乎不可能。因此，我们希望拥有能围绕当前已掌握词汇展开的阅读材料。

我们将使用 LLM 为我们生成简短故事。但为了实现个性化学习体验，我们希望生成的故事聚焦于我们已经掌握的词汇。

以下是我们所需内容的拆解：

1. 一个已知词汇的数据库。
2. AI 代理需要访问该数据库，检索已知词汇并生成故事。
3. AI 代理需要访问该数据库以添加新单词。

数据库初始化组件

当我们用 Docker 运行 Langflow 时，会启动一个 Postgres 数据库。我们可以使用 psycopg2 包进行访问。想了解更多 Postgres，请查看这篇PostgreSQL 教程。

我们将创建一个节点，用于上传包含已知单词的 CSV 文件来初始化数据库。我的 CSV 文件如下：

我创建了一个具有两个输入的组件。

1. 一个列名，用于指定包含待纳入单词的列。
2. 一个文件输入，用于提供包含单词的 CSV 文件。

以下是组件代码：

from langflow.custom import Component
from langflow.io import MessageTextInput, Output
from langflow.schema import Message
import psycopg2
import csv

def connect_to_database():
    # Connect to the Postgres database provided by the Docker setup
    conn = psycopg2.connect(
        dbname="langflow",
        user="langflow",
        password="langflow",
        host="postgres",
        port="5432"
    )
    conn.autocommit = True    
    return conn.cursor()

def add_word(cursor, word):
    # Add a word to the database
    cursor.execute(
        "INSERT INTO words (word) VALUES (%s) ON CONFLICT (word) DO NOTHING;",
        (word,)
    )

def initialize_database(cursor):
    # Initialize the database by creating the word table if it doesn't yet exist
    create_table_query = """
        CREATE TABLE IF NOT EXISTS words (
            word TEXT PRIMARY KEY
        );
    """
    cursor.execute(create_table_query)    

class UploadWordFile(Component):
    display_name = "Upload Word File"
    description = "Upload a CSV file of words to the database."
    icon = "code"
    name = "UploadWordFile"

    inputs = [
        StrInput(
            name="column_name",
            display_name="Column Name",
            info="The name of the column containing the words",
        ),
        FileInput(
            name="csv_file",
            display_name="CSV file",
            info="CSV input file",
            file_types=["csv"]
        ),
    ]

    outputs = [
        Output(display_name="Output", name="output", method="load_words_into_database"),
    ]

    def load_words_into_database(self) -> Message:
        try:
            cursor = connect_to_database()
            initialize_database(cursor)
            with open(self.csv_file, "rt") as f:
                rows = list(csv.reader(f))
            headers = list(map(lambda header: header.lower(), rows[0]))
            column_index = headers.index(self.column_name)
            for row in rows[1:]:
                add_word(cursor, row[column_index])
            return "Success"
        except Exception as e:
            return f"Error: {str(e)}"技术支持

组件如下所示：

该组件旨在单独使用，不连接其他组件。我们可以用它手动将单词上传到数据库。

添加单词工具

这里我们创建一个工具，供 AI 代理用来向数据库添加新单词。这样，用户只需向代理发送消息即可轻松添加新词。

此代码复用了前一节点中的 connect_to_database() 和 add_word() 函数。如果有更好的 Docker 设置，我们可以避免重复代码，但在本教程中我们保持简单。

以下是该工具的代码：

# from langflow.field_typing import Data
from langflow.custom import Component
from langflow.io import MessageTextInput, Output
from langflow.schema import Message

def connect_to_database():
    # Connect to the Postgres database provided by the Docker setup
    conn = psycopg2.connect(
        dbname="langflow",
        user="langflow",
        password="langflow",
        host="postgres",
        port="5432"
    )
    conn.autocommit = True    
    return conn.cursor()

def add_word(cursor, word):
    # Add a word to the database
    cursor.execute(
        "INSERT INTO words (word) VALUES (%s) ON CONFLICT (word) DO NOTHING;",
        (word,)
    )

class AddWordTool(Component):
    display_name = "Add word tool"
    description = "Use this tool to add a new word"
    icon = "code"
    name = "AddWordTool"

    inputs = [
        MessageTextInput(
            name="word",
            display_name="Word",
            info="The word to add",
            tool_mode=True,
        ),
    ]

    outputs = [
        Output(display_name="Output", name="output", method="add_new_word"),
    ]

    def add_new_word(self) -> Message:
        cursor = connect_to_database()
        add_word(cursor, self.word)
        return f"Added word: {self.word}"技术支持

请注意，该工具假定数据库已事先创建。若未至少运行过前一个组件一次，将无法正常工作。

记得将其设置为工具模式：

故事生成工具

要创建故事，我们使用一个 AI 代理组件。结构如下：

在代理说明中，我们没有直接在组件里添加提示词，而是使用了一个Prompt 组件，因为这样可以为提示词添加参数。我们使用的提示词是：

Create a story in {language} using only words from the following list:

{words}技术支持

该提示词包含两个参数，{language} 和 {words}。像这样添加参数时，会在组件中为每个参数新增一个字段。

language 的值我们直接在文本框中设置。words 需要从数据库加载，因此我们创建了一个自定义组件从数据库加载所有单词，并将该节点的输出连接到 words 字段。

以下是词汇加载器组件的代码：

from langflow.custom import Component
from langflow.io import MessageTextInput, Output
from langflow.schema import Message
import psycopg2

def load_words():
    conn = psycopg2.connect(
        dbname="langflow",
        user="langflow",
        password="langflow",
        host="postgres",
        port="5432"
    )    
    cursor = conn.cursor()
    cursor.execute("""
        SELECT * FROM words;
    """)
    rows = cursor.fetchall()
    return map(lambda row: row[0], rows)

class WordLoader(Component):
    display_name = "Word Loader"
    description = "Load words from the database"
    icon = "code"
    name = "WordLoader"

    outputs = [
        Output(display_name="Output", name="output", method="build_output"),
    ]

    def build_output(self) -> Message:
        return str(", ".join(load_words()))技术支持

接下来，我们将该代理设置为工具。这样，主代理（我们随后创建）就能在需要时调用它来生成故事。

最后，我们重命名该代理并修改工具描述，以便让 AI 知道何时使用该工具：

语言代理

对于主代理，我们使用一个常规的 AI 代理节点组件，并连接到我们创建的两个工具。该代理使用聊天输入和聊天输出组件，以便我们通过聊天界面与其交互。

最终架构如下：

我们使用的提示词为：

You will help the user practice their language skills. You will either be asked to create a story or to add a new word to the vocabulary.

- If the user asks you to create a story, use the story generation tool.  
- If the user asks you to add a word, use the word add tool.  

When using a tool, your answer should just be the result from the tool and nothing else.技术支持

我们可以点击 “Playground” 按钮来执行完整工作流。以下是一段示例对话：

总结

我们探索了 Langflow 的基本要点，以及如何使用它构建帮助用户练习阅读能力的 AI 代理工作流。通过使用 Python 创建自定义组件，我们解锁了更多可能性。

这种混合方式融合了两者的优势：Python 脚本的灵活性（能够执行任何所需任务）与 Langflow 直观的图形界面（无需深入复杂代码即可轻松搭建工作流）。

想进一步了解 AI 代理，推荐阅读以下博客：

• OpenAI 的 Codex：含 3 个实用示例的指南
• n8n：含实用示例的指南
• Haystack AI 教程：构建 Agentic 工作流
• Google 的 Agent Development Kit（ADK）：含演示项目的指南
• Agent2Agent（A2A）：定义、示例、与 MCP 的比较