当前位置：首页 > news >正文

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

news 2025/9/1 6:05:47

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: strdisplay_name: str | None = Nonefield_type: str = "str"advanced: bool = Falsemultiline: bool = Trueinput_types: list[str] = DEFAULT_PROMPT_INTUT_TYPESvalue: 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 = Nonesender_name: str | None = Nonefiles: 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-optimizationvariables_append = variables.appendseen_add = seen.addseen_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_variablesinput_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 excreturn 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_variablesnew_var = var# Handle variables starting with a numberif 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 invalidnew_var = new_var.replace("{{", "ᴛᴇᴍᴘᴏᴘᴇɴ").replace("}}", "ᴛᴇᴍᴘᴄʟᴏsᴇ")# Remove invalid charactersfor 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 variablesinput_variables = validate_prompt(template)# Initialize custom_fields if Noneif custom_fields is None:custom_fields = defaultdict(list)# Retrieve old custom fieldsold_custom_fields = get_old_custom_fields(custom_fields, name)# Add new variables to the templateadd_new_variables_to_template(input_variables, custom_fields, frontend_node_template, name)# Remove old variables from the templateremove_old_variables_from_template(old_custom_fields, input_variables, custom_fields, frontend_node_template, name)# Update the input variables field in the templateupdate_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 valuetemplate_field.value = template[variable]["value"]template[variable] = template_field.to_dict()# Check if variable is not already in the list before appendingif 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 Messaged_copy = deepcopy(d)for key, value in d_copy.items():# it could be a list of data or documents or stringsif isinstance(value, list):for i, item in enumerate(value):if isinstance(item, Message):d_copy[key][i] = item.textelif 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.textelif 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_promptreturn 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 = resultreturn 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
)