用 Langflow 自定义组件规范化食堂报修单

Langflow 的强项是用 Python 写自定义组件。食堂报修单经常是口语化的「305 室水龙头一直漏水」，进流程前最好先结构化。 这篇实现 TicketNormalizer：可在无 Langflow 时单测的 normalize_ticket，外加符合官方自定义组件形状的类骨架；再用 check_component.py 用 AST 检查 display_name / 输出方法等。

先看清成品

• normalize_ticket("305室水龙头一直漏水") → room=305, category=漏水, urgent=true。
• 组件骨架检查通过；bad_component.py 报出缺属性/缺方法。
• 5 项 pytest 通过。未加载进真实 Langflow。

合规提醒：用自己的 Langflow 实例与模型密钥；自定义组件里的代码会在服务器上执行，不要加载不可信流程。密钥放环境变量，导出流程前确认不会带出 Key。费用以模型供应商为准，本文不写价格。

运行说明：我没有安装完整 Langflow 服务。真实运行过的是：组件骨架检查、normalize 逻辑、Run API 客户端对 mock 的请求、.env 检查，以及 pytest（2026-10-23，UTC+8）。服务端行为来自官方文档。

做完之后你会得到什么？

需要准备哪些环境？

关键事实

文件结构

92-langflow-A-src/
├── components/
│   ├── __init__.py
│   └── ticket_normalizer.py
├── check_component.py
├── bad_component.py
└── test_92a.py

第 1 步：组件

"""Langflow-style custom component: normalize cafeteria repair tickets.

This file follows the official custom-component shape (display_name, inputs,
outputs, and a method matching the output). It also runs standalone so we can
test the logic without installing the full Langflow server.
"""
from __future__ import annotations

import re
from dataclasses import dataclass
from typing import Any

# --- Standalone core (also used by tests) ---------------------------------

ROOM_RE = re.compile(r"(?P<room>\d{2,4})\s*(室|房间|房)?")
CATS = [
    ("漏水", ["漏水", "滴水", "水管", "水龙头"]),
    ("电路", ["停电", "跳闸", "插座", "灯不亮", "电线"]),
    ("门窗", ["门锁", "钥匙", "窗户", "玻璃"]),
    ("空调", ["空调", "不制冷", "外机"]),
]


def normalize_ticket(text: str) -> dict[str, Any]:
    text = (text or "").strip()
    if not text:
        raise ValueError("empty ticket")
    room = None
    m = ROOM_RE.search(text)
    if m:
        room = m.group("room")
    cat = "其他"
    for name, kws in CATS:
        if any(k in text for k in kws):
            cat = name
            break
    urgent = any(k in text for k in ("一直", "紧急", "整层", "冒烟", "漏电"))
    return {
        "room": room,
        "category": cat,
        "urgent": urgent,
        "summary": text[:80],
    }


# --- Langflow Component skeleton (docs.langflow.org/components-custom-components)

try:
    from lfx.custom.custom_component.component import Component  # type: ignore
    from lfx.io import MessageTextInput, Output  # type: ignore
    from lfx.schema import Data  # type: ignore

    class TicketNormalizer(Component):
        display_name = "Ticket Normalizer"
        description = "Normalize cafeteria repair tickets into structured fields."
        icon = "wrench"
        name = "ticket_normalizer"

        inputs = [
            MessageTextInput(name="text", display_name="Ticket text", required=True),
        ]
        outputs = [
            Output(display_name="Ticket", name="ticket", method="build_ticket"),
        ]

        def build_ticket(self) -> Data:
            data = normalize_ticket(self.text)
            return Data(data=data)

except ImportError:
    # Langflow / lfx not installed — skeleton still documents the interface.
    class TicketNormalizer:  # type: ignore
        display_name = "Ticket Normalizer"
        description = "Normalize cafeteria repair tickets into structured fields."
        icon = "wrench"
        name = "ticket_normalizer"

        def build_ticket(self, text: str) -> dict:
            return normalize_ticket(text)

第 2 步：检查器

#!/usr/bin/env python3
"""Validate that a Python file looks like a Langflow custom component."""
from __future__ import annotations

import argparse
import ast
import sys
from pathlib import Path

REQUIRED_ATTRS = ("display_name", "description", "name")


def check(path: Path) -> list[str]:
    src = path.read_text(encoding="utf-8")
    tree = ast.parse(src, filename=str(path))
    errs: list[str] = []
    classes = [n for n in ast.walk(tree) if isinstance(n, ast.ClassDef)]
    # Prefer a class that defines display_name (may live inside try/except ImportError)
    target = None
    for c in classes:
        names = {n.targets[0].id for n in c.body if isinstance(n, ast.Assign) for t in n.targets if isinstance(t, ast.Name)}
        # also AnnAssign
        for n in c.body:
            if isinstance(n, ast.AnnAssign) and isinstance(n.target, ast.Name):
                names.add(n.target.id)
        if "display_name" in names or any(isinstance(n, ast.Assign) and any(getattr(t, "id", None) == "display_name" for t in getattr(n, "targets", [])) for n in c.body):
            target = c
            break
    if target is None and classes:
        target = classes[-1]
    if target is None:
        return ["没有 class 定义"]

    attrs: set[str] = set()
    methods: set[str] = set()
    for n in target.body:
        if isinstance(n, ast.Assign):
            for t in n.targets:
                if isinstance(t, ast.Name):
                    attrs.add(t.id)
        elif isinstance(n, ast.AnnAssign) and isinstance(n.target, ast.Name):
            attrs.add(n.target.id)
        elif isinstance(n, ast.FunctionDef):
            methods.add(n.name)

    for a in REQUIRED_ATTRS:
        if a not in attrs:
            errs.append(f"缺少类属性 {a}")
    if "build_" not in " ".join(methods) and "process" not in methods and not any(m.startswith("build") for m in methods):
        # Official pattern: output method name matches outputs[].method
        if not methods - {"__init__", "__doc__"}:
            errs.append("缺少处理用的方法（例如 build_ticket）")
    # Soft check: mentions Component inheritance in source
    if "Component" not in src and "lfx" not in src and "langflow" not in src:
        errs.append("未看到 Component / lfx / langflow 引用（可能不是可加载的组件）")
    return errs


def main(argv=None) -> int:
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument("path", type=Path)
    args = p.parse_args(argv)
    errs = check(args.path)
    if errs:
        print(f"✗ {args.path.name}")
        for e in errs:
            print(f"  - {e}")
        return 1
    print(f"✓ {args.path.name}: 组件骨架检查通过")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

第 3 步：测试

from pathlib import Path
from components.ticket_normalizer import normalize_ticket
from check_component import check, main

ROOT = Path(__file__).parent

def test_normalize():
    r = normalize_ticket("302 卫生间水龙头一直滴水")
    assert r["room"] == "302"
    assert r["category"] == "漏水"
    assert r["urgent"] is True

def test_empty():
    try:
        normalize_ticket("  ")
        assert False
    except ValueError:
        pass

def test_component_ok():
    assert check(ROOT / "components/ticket_normalizer.py") == []

def test_component_bad():
    errs = check(ROOT / "bad_component.py")
    assert errs

def test_cli():
    assert main([str(ROOT / "components/ticket_normalizer.py")]) == 0
    assert main([str(ROOT / "bad_component.py")]) == 1

第 4 步：运行

# 2026-10-23 12:37 UTC+8
$ python check_component.py components/ticket_normalizer.py
✓ ticket_normalizer.py: 组件骨架检查通过
$ python check_component.py bad_component.py
✗ bad_component.py
  - 缺少类属性 display_name
  - 缺少类属性 description
  - 缺少类属性 name
  - 缺少处理用的方法（例如 build_ticket）
  - 未看到 Component / lfx / langflow 引用（可能不是可加载的组件）
$ python -c "from components.ticket_normalizer import normalize_ticket; import json; print(json.dumps(normalize_ticket('305室水龙头一直漏水'), ensure_ascii=False, indent=2))"
{
  "room": "305",
  "category": "漏水",
  "urgent": true,
  "summary": "305室水龙头一直漏水"
}
$ python -m pytest -q
5 passed

怎么检查结果？

• 骨架检查必须通过；bad 必须失败。
• normalize 对漏水/紧急关键词分类正确。
• 装好 Langflow 后，把 components 目录按官方方式挂载再在画布里选组件。

调试示例

AST 看不到 try 里的 class。 第一版 check_component 只扫 tree.body 顶层 ClassDef。组件文件把 TicketNormalizer 放在 try: import lfx ... except ImportError: 里，顶层没有 ClassDef，于是报「没有 class 定义」。改成 ast.walk 后才能找到。自定义组件经常要兼容「有/无 lfx」，检查器必须能进 try/except。

常见问题

模板规则

• 核心逻辑与框架 import 解耦，便于 pytest。
• 提交前跑 check_component。
• 不加载不可信自定义组件。

项目资源

先把结构化逻辑测绿，再拖进画布。