Langflow 拖拽式 Agent¶

为什么要学 Langflow¶

Langflow 是由 LangChain 团队打造的可视化 AI 应用构建平台，让你通过拖拽组件来设计和部署 LLM 工作流。相比 Flowise，Langflow 与 LangChain 生态深度集成，组件更丰富，同时提供了 Python 代码导出能力，便于从原型过渡到生产。对于同时需要快速原型验证和后续代码化的团队，Langflow 是理想选择。

核心概念¶

概念	白话解释	用途
Flow	工作流	一组连接的组件构成完整应用
Component	组件	可复用的功能单元（LLM、工具等）
Playground	测试区	实时测试和调试 Flow
API	接口	自动为 Flow 生成调用接口
Starter Projects	模板项目	预构建的常见场景模板
Custom Component	自定义组件	用 Python 扩展新功能

安装配置¶

pip 安装¶

pip install langflow

# 启动
langflow run --port 7860

# 带认证启动
langflow run --port 7860 \
  --superuser admin \
  --superuser-password admin123

Docker 部署¶

docker run -d \
  --name langflow \
  -p 7860:7860 \
  -v langflow-data:/app/langflow \
  langflowai/langflow:latest

Docker Compose 生产环境¶

version: '3.8'
services:
  langflow:
    image: langflowai/langflow:latest
    ports:
      - "7860:7860"
    environment:
      - LANGFLOW_DATABASE_URL=postgresql://langflow:pass@postgres:5432/langflow
      - LANGFLOW_SUPERUSER=admin
      - LANGFLOW_SUPERUSER_PASSWORD=${ADMIN_PASSWORD}
      - LANGFLOW_SECRET_KEY=${SECRET_KEY}
    depends_on:
      - postgres
    restart: unless-stopped

  postgres:
    image: postgres:16-alpine
    environment:
      - POSTGRES_USER=langflow
      - POSTGRES_PASSWORD=pass
      - POSTGRES_DB=langflow
    volumes:
      - pg-data:/var/lib/postgresql/data

volumes:
  pg-data:

快速上手¶

使用模板快速开始¶

打开 http://localhost:7860
点击 "New Flow"
选择 Starter Project：
Basic Prompting: 简单的提示词对话
Vector Store RAG: 文档问答
Simple Agent: 带工具的智能体
修改配置（如 API Key、模型选择）
在 Playground 中测试

构建 RAG Flow¶

1. 添加组件：
   - File Loader → Text Splitter → Embeddings → Vector Store
   - Chat Input → Retriever → Prompt → LLM → Chat Output

2. 连接逻辑：
   文档侧：File → Splitter → (Embeddings) → Vector Store
   查询侧：Input → Vector Store.search → Prompt → LLM → Output

Playground 测试¶

Flow 构建完成后，点击 Playground 标签页： - 直接输入问题测试 - 查看中间步骤的输出 - 调整参数后实时验证效果

API 调用¶

import requests

FLOW_ID = "your-flow-id"
BASE_URL = "http://localhost:7860"

# 简单调用
response = requests.post(
    f"{BASE_URL}/api/v1/run/{FLOW_ID}",
    json={
        "input_value": "什么是微服务架构？",
        "output_type": "chat",
        "input_type": "chat"
    }
)
result = response.json()
print(result["outputs"][0]["outputs"][0]["results"]["message"]["text"])

进阶用法¶

自定义 Python 组件¶

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

class DataFilterComponent(Component):
    display_name = "数据过滤器"
    description = "根据条件过滤输入数据"

    inputs = [
        MessageTextInput(name="input_data", display_name="输入数据"),
        MessageTextInput(name="filter_key", display_name="过滤关键词"),
    ]

    outputs = [
        Output(display_name="过滤结果", name="filtered", method="filter_data"),
    ]

    def filter_data(self) -> Data:
        data = self.input_data
        key = self.filter_key
        # 自定义过滤逻辑
        filtered = [item for item in data if key in str(item)]
        return Data(value=filtered)

Multi-Agent 协作¶

构建多 Agent 协作流：

1. Researcher Agent（配工具：Web Search、Wikipedia）
2. Writer Agent（配工具：Text Generator）
3. Reviewer Agent（配工具：Grammar Check）

连接方式：
User Input → Router → [Researcher → Writer → Reviewer] → Output

条件路由¶

# 使用 Conditional Router 组件
# 根据输入内容路由到不同处理分支

# 条件示例：
# if "code" in input → Code Assistant Flow
# elif "data" in input → Data Analysis Flow
# else → General Chat Flow

批量处理¶

import requests
from concurrent.futures import ThreadPoolExecutor

FLOW_ID = "your-flow-id"
BASE_URL = "http://localhost:7860/api/v1/run"

questions = [
    "问题1", "问题2", "问题3", ...
]

def ask(question):
    resp = requests.post(f"{BASE_URL}/{FLOW_ID}", json={
        "input_value": question,
        "output_type": "chat",
        "input_type": "chat"
    })
    return resp.json()

with ThreadPoolExecutor(max_workers=5) as executor:
    results = list(executor.map(ask, questions))

与 LangSmith 集成¶

# 环境变量配置
export LANGCHAIN_TRACING_V2=true
export LANGCHAIN_API_KEY=your-langsmith-key
export LANGCHAIN_PROJECT=langflow-production

Flow 导出为 Python 代码¶

Langflow 支持将可视化 Flow 导出为等效的 Python 代码： 1. 在 Flow 编辑页面点击 "Code" 按钮 2. 复制生成的 LangChain 代码 3. 可以在生产环境中直接运行，无需 Langflow UI

常见问题¶

Q: Langflow 和 Flowise 选哪个？¶

Langflow：LangChain 深度集成，组件更多，支持代码导出，适合技术团队
Flowise：更轻量，LangChain/LlamaIndex 都支持，适合快速部署

Q: 组件连接失败怎么排查？¶

检查组件版本兼容性
确认连线的输入/输出类型匹配
查看底部日志面板的错误信息
确保 API Key 等凭据已正确配置

Q: 如何处理大文件？¶

在 Text Splitter 中调整 chunk_size 和 overlap
使用 Astra DB 或 Pinecone 等云向量库代替内存存储
文件预处理后再上传

Q: 支持自托管 LLM 吗？¶

支持。通过 Ollama、LocalAI、vLLM 等节点连接本地模型。配置 base_url 指向本地服务即可。

参考资源¶

GitHub：https://github.com/langflow-ai/langflow
文档：https://docs.langflow.org/
组件参考：https://docs.langflow.org/components
Starter Projects：https://docs.langflow.org/starter-projects
Discord：https://discord.gg/EqksyE2EX9