Langflow 框架中 Prompt 技术底层实现分析

类型安全：通过 Pydantic 模型确保数据类型安全可扩展性：组件化设计支持自定义 Prompt 组件兼容性：与 LangChain 生态系统完全兼容验证机制：完善的变量验证和错误处理多模态支持：支持文本、图片等多种内容类型。

nlog3n

703人浏览 · 2025-08-31 14:44:49

nlog3n · 2025-08-31 14:44:49 发布

Langflow 框架中 Prompt 技术底层实现分析

1. Prompt 技术概述

Langflow 是一个基于 LangChain 的可视化 AI 工作流构建框架，其 Prompt 技术是整个系统的核心组件之一。Prompt 技术主要负责：

模板化处理：支持动态变量替换的提示词模板
变量验证：确保模板变量的合法性和安全性
类型转换：将各种数据类型转换为字符串格式
与 LLM 集成：将处理后的 Prompt 传递给语言模型

2. 核心类和接口分析

2.1 PromptComponent 类

位置：base/langflow/components/processing/prompt.py

class PromptComponent(Component):
    display_name: str = "提示词模板"
    description: str = "创建带有动态变量的提示词模板。"
    documentation: str = "https://docs.langflow.org/components-prompts"
    icon = "braces"
    trace_type = "prompt"
    name = "Prompt Template"
    priority = 0  # 设置优先级为0，使其首先出现

    inputs = [
        PromptInput(
            name="template", 
            display_name="模板",
            info="提示词模板，支持使用 {变量名} 格式的动态变量。"
        ),
        MessageTextInput(
            name="tool_placeholder",
            display_name="工具占位符",
            tool_mode=True,
            advanced=True,
            info="工具模式的占位符输入。",
        ),
    ]

    outputs = [
        Output(display_name="提示词", name="prompt", method="build_prompt"),
    ]

核心功能：

继承自 Component 基类，提供组件化的 Prompt 处理能力
支持动态模板更新和前端节点同步
实现了 build_prompt() 方法生成最终的 Message 对象

2.2 DefaultPromptField 类

位置：base/langflow/inputs/inputs.py

class DefaultPromptField(Input):
    name: str
    display_name: str | None = None
    field_type: str = "str"
    advanced: bool = False
    multiline: bool = True
    input_types: list[str] = DEFAULT_PROMPT_INTUT_TYPES
    value: Any = ""  # Set the value to empty string

核心功能：

定义了 Prompt 字段的基本属性
支持多行文本输入
默认接受 Message 类型的输入

2.3 Message 类

位置：base/langflow/schema/message.py

class Message(Data):
    text_key: str = "text"
    text: str | AsyncIterator | Iterator | None = Field(default="")
    sender: str | None = None
    sender_name: str | None = None
    files: list[str | Image] | None = Field(default=[])
    session_id: str | UUID | None = Field(default="")
    timestamp: Annotated[str, timestamp_to_str_validator] = Field(
        default_factory=lambda: datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S %Z")
    )
    # ... 其他字段

3. 关键实现细节

3.1 变量提取机制

位置：base/langflow/interface/utils.py

def extract_input_variables_from_prompt(prompt: str) -> list[str]:
    """Extract variable names from a prompt string using Python's built-in string formatter.

    Uses the same convention as Python's .format() method:
    - Single braces {name} are variable placeholders
    - Double braces {{name}} are escape sequences that render as literal {name}
    """
    formatter = Formatter()
    variables: list[str] = []
    seen: set[str] = set()

    # Use local bindings for micro-optimization
    variables_append = variables.append
    seen_add = seen.add
    seen_contains = seen.__contains__

    for _, field_name, _, _ in formatter.parse(prompt):
        if field_name and not seen_contains(field_name):
            variables_append(field_name)
            seen_add(field_name)

    return variables

实现原理：

使用 Python 内置的 string.Formatter 类解析模板
遵循 Python .format() 方法的约定
单花括号 {name} 表示变量占位符
双花括号 {{name}} 表示转义序列，渲染为字面量 {name}
通过集合去重确保变量唯一性

3.2 变量验证机制

位置：base/langflow/base/prompts/api_utils.py

def validate_prompt(prompt_template: str, *, silent_errors: bool = False) -> list[str]:
    input_variables = extract_input_variables_from_prompt(prompt_template)

    # Check if there are invalid characters in the input_variables
    input_variables = _check_input_variables(input_variables)
    if any(var in _INVALID_NAMES for var in input_variables):
        msg = f"Invalid input variables. None of the variables can be named {', '.join(input_variables)}. "
        raise ValueError(msg)

    try:
        PromptTemplate(template=prompt_template, input_variables=input_variables)
    except Exception as exc:
        msg = f"Invalid prompt: {exc}"
        logger.exception(msg)
        if not silent_errors:
            raise ValueError(msg) from exc

    return input_variables

验证规则：

字符验证：禁止使用特殊字符（空格、逗号、括号等）
保留字检查：禁止使用系统保留的变量名
LangChain 兼容性：通过 LangChain 的 PromptTemplate 验证模板格式

无效字符集合：

_INVALID_CHARACTERS = {
    " ", ",", ".", ":", ";", "!", "?", "/", "\\", "(", ")", "[", "]",
}

_INVALID_NAMES = {
    "code", "input_variables", "output_parser", "partial_variables", 
    "template", "template_format", "validate_template",
}

3.3 变量修复机制

def _fix_variable(var, invalid_chars, wrong_variables):
    if not var:
        return var, invalid_chars, wrong_variables
    new_var = var

    # Handle variables starting with a number
    if var[0].isdigit():
        invalid_chars.append(var[0])
        new_var, invalid_chars, wrong_variables = _fix_variable(var[1:], invalid_chars, wrong_variables)

    # Temporarily replace {{ and }} to avoid treating them as invalid
    new_var = new_var.replace("{{", "ᴛᴇᴍᴘᴏᴘᴇɴ").replace("}}", "ᴛᴇᴍᴘᴄʟᴏsᴇ")

    # Remove invalid characters
    for char in new_var:
        if char in _INVALID_CHARACTERS:
            invalid_chars.append(char)
            new_var = new_var.replace(char, "")
            if var not in wrong_variables:
                wrong_variables.append(var)

    # Restore {{ and }}
    new_var = new_var.replace("ᴛᴇᴍᴘᴏᴘᴇɴ", "{{").replace("ᴛᴇᴍᴘᴄʟᴏsᴇ", "}}")

    return new_var, invalid_chars, wrong_variables

修复策略：

移除数字开头的字符
临时替换双花括号避免误处理
移除所有无效字符
恢复双花括号的原始格式

4. Prompt 模板系统实现

4.1 模板处理流程

位置：base/langflow/base/prompts/api_utils.py

def process_prompt_template(
    template: str, name: str, custom_fields: dict[str, list[str]] | None, frontend_node_template: dict[str, Any]
):
    """Process and validate prompt template, update template and custom fields."""
    # Validate the prompt template and extract input variables
    input_variables = validate_prompt(template)

    # Initialize custom_fields if None
    if custom_fields is None:
        custom_fields = defaultdict(list)

    # Retrieve old custom fields
    old_custom_fields = get_old_custom_fields(custom_fields, name)

    # Add new variables to the template
    add_new_variables_to_template(input_variables, custom_fields, frontend_node_template, name)

    # Remove old variables from the template
    remove_old_variables_from_template(old_custom_fields, input_variables, custom_fields, frontend_node_template, name)

    # Update the input variables field in the template
    update_input_variables_field(input_variables, frontend_node_template)

    return input_variables

处理步骤：

模板验证：验证模板格式和变量合法性
字段初始化：初始化自定义字段字典
旧字段处理：获取并处理旧的自定义字段
新变量添加：将新发现的变量添加到模板
旧变量清理：移除不再使用的变量
字段更新：更新输入变量字段

4.2 动态字段管理

def add_new_variables_to_template(input_variables, custom_fields, template, name) -> None:
    for variable in input_variables:
        try:
            template_field = DefaultPromptField(name=variable, display_name=variable)
            if variable in template:
                # Set the new field with the old value
                template_field.value = template[variable]["value"]

            template[variable] = template_field.to_dict()

            # Check if variable is not already in the list before appending
            if variable not in custom_fields[name]:
                custom_fields[name].append(variable)

        except Exception as exc:
            raise HTTPException(status_code=500, detail=str(exc)) from exc

动态管理特性：

自动为新变量创建输入字段
保留现有变量的值
避免重复添加相同变量
异常处理确保系统稳定性

5. 变量处理机制

5.1 数据类型转换

位置：base/langflow/base/prompts/utils.py

def dict_values_to_string(d: dict) -> dict:
    """Converts the values of a dictionary to strings."""
    from langflow.schema.message import Message

    d_copy = deepcopy(d)
    for key, value in d_copy.items():
        # it could be a list of data or documents or strings
        if isinstance(value, list):
            for i, item in enumerate(value):
                if isinstance(item, Message):
                    d_copy[key][i] = item.text
                elif isinstance(item, Data):
                    d_copy[key][i] = data_to_string(item)
                elif isinstance(item, Document):
                    d_copy[key][i] = document_to_string(item)
        elif isinstance(value, Message):
            d_copy[key] = value.text
        elif isinstance(value, Data):
            d_copy[key] = data_to_string(value)
        elif isinstance(value, Document):
            d_copy[key] = document_to_string(value)
    return d_copy

转换规则：

Message 对象：提取 text 属性
Data 对象：调用 get_text() 方法
Document 对象：提取 page_content 属性
列表类型：递归处理每个元素
深拷贝：避免修改原始数据

5.2 模板格式化

位置：base/langflow/schema/message.py

def format_text(self):
    prompt_template = PromptTemplate.from_template(self.template)
    variables_with_str_values = dict_values_to_string(self.variables)
    formatted_prompt = prompt_template.format(**variables_with_str_values)
    self.text = formatted_prompt
    return formatted_prompt

格式化流程：

创建 LangChain PromptTemplate 对象
将所有变量值转换为字符串
使用 LangChain 的格式化方法
更新 Message 对象的 text 属性

6. 与 LLM 的集成方式

6.1 Message 创建流程

@classmethod
def from_template(cls, template: str, **variables):
    instance = cls(template=template, variables=variables)
    text = instance.format_text()
    message = HumanMessage(content=text)
    contents = []
    for value in variables.values():
        if isinstance(value, cls) and value.files:
            content_dicts = value.get_file_content_dicts()
            contents.extend(content_dicts)
    if contents:
        message = HumanMessage(content=[{"type": "text", "text": text}, *contents])

    prompt_template = ChatPromptTemplate.from_messages([message])

    instance.prompt = jsonable_encoder(prompt_template.to_json())
    instance.messages = instance.prompt.get("kwargs", {}).get("messages", [])
    return instance

集成特性：

创建 LangChain 兼容的 HumanMessage
支持多模态内容（文本 + 文件）
生成 ChatPromptTemplate 对象
JSON 序列化支持

6.2 LLM 模型集成

位置：base/langflow/base/models/model.py

class LCModelComponent(Component):
    _base_inputs: list[InputTypes] = [
        MessageInput(name="input_value", display_name="输入", info="发送给模型的输入内容。"),
        MultilineInput(
            name="system_message",
            display_name="系统提示",
            info="大模型的系统消息.",
            advanced=False,
        ),
        BoolInput(name="stream", display_name="Stream模式", info=STREAM_INFO_TEXT, advanced=True),
    ]

    async def text_response(self) -> Message:
        output = self.build_model()
        result = await self.get_chat_result(
            runnable=output, stream=self.stream, input_value=self.input_value, system_message=self.system_message
        )
        self.status = result
        return result

集成机制：

接受 Message 类型的输入
支持系统消息设置
异步处理支持流式输出
统一的模型接口抽象

7. 应用示例

7.1 基础 Prompt 模板示例

# 创建一个简单的问答模板
class SimpleQAPrompt:
    def __init__(self):
        self.template = "问题：{question}\n请提供详细的答案："
        
    def create_prompt(self, question: str) -> Message:
        # 使用 langflow 的 Message.from_template 方法
        return Message.from_template(
            template=self.template,
            question=question
        )

# 使用示例
qa_prompt = SimpleQAPrompt()
message = qa_prompt.create_prompt("什么是人工智能？")
print(message.text)
# 输出：问题：什么是人工智能？
#      请提供详细的答案：

7.2 多变量复杂模板示例

# 创建一个代码生成模板
class CodeGenerationPrompt:
    def __init__(self):
        self.template = """
你是一个专业的 {language} 程序员。

任务：{task}
要求：
- 编程语言：{language}
- 代码风格：{style}
- 包含注释：{include_comments}

请生成高质量的代码：
"""
    
    def create_prompt(self, language: str, task: str, style: str = "简洁", include_comments: bool = True) -> Message:
        return Message.from_template(
            template=self.template,
            language=language,
            task=task,
            style=style,
            include_comments="是" if include_comments else "否"
        )

# 使用示例
code_prompt = CodeGenerationPrompt()
message = code_prompt.create_prompt(
    language="Python",
    task="实现一个二分查找算法",
    style="Pythonic",
    include_comments=True
)