用Langflow和Streamlit打造无代码聊天机器人-洪萨配资

用 Langflow 和 Streamlit 打造无代码聊天机器人

在生成式 AI 技术席卷各行各业的今天，越来越多的产品经理、业务分析师甚至非技术人员都希望能快速构建一个能“说话”的智能助手。但传统开发模式下，从设计提示词、搭建链路、集成模型到部署 Web 界面，整个流程动辄需要数天甚至数周。

有没有可能像搭积木一样，把大语言模型（LLM）的能力拼接起来，并且一键变成可交互的应用？答案是肯定的——Langflow + Streamlit正好提供了这样一条“零代码原型 → 快速上线”的黄金路径。

想象这样一个场景：你手头有一份企业内部的旅游手册 PDF，老板希望你能做一个聊天机器人，让员工随时查询某个城市的签证政策或推荐景点。过去这可能需要调用多个 API、写一堆 Python 脚本、再找前端同事做个页面；而现在，只需三步：

在 Langflow 中拖拽几个组件，完成数据加载、向量化和问答链的配置；
启动服务并获取流程 ID；
用不到百行的 Streamlit 代码封装成网页应用。

整个过程不需要写后端接口，也不用懂 React，甚至连 LangChain 的底层细节都可以先放一放。这就是现代 AI 工具链带来的变革：让创意更快落地，让技术更贴近普通人。

Langflow 的核心魅力在于它把 LangChain 那些复杂的抽象——比如 Chains、Prompts、Memory、Vector Stores——统统变成了可视化的“节点”。你可以把这些节点理解为乐高积木块，每个都有明确的功能标签：

LLM节点负责调用模型，支持 OpenAI、HuggingFace、Ollama 等主流提供商；
Prompt Template定义输入格式，比如“请以导游口吻回答以下问题”；
Document Loader可以读取 PDF、TXT、Markdown 文件；
Text Splitter将长文档切分成适合嵌入的小段落；
Embeddings调用 OpenAI 或 Sentence Transformers 实现语义编码；
Vector Store如 Chroma、FAISS 存储向量，支持高效检索；
Retriever根据用户提问找出最相关的知识片段；
ConversationalRetrievalQAChain则是关键角色，它整合了检索结果与对话历史，生成自然语言回答。

把这些节点连起来，就构成了一个完整的 RAG（检索增强生成）系统。更重要的是，你在画布上点击任何一个节点，都能实时看到它的输出结果。这种即时反馈机制极大提升了调试效率，尤其适合探索性实验。

比如你想测试不同文本分割策略对最终回答质量的影响，只需修改RecursiveCharacterTextSplitter的参数，运行一下就能对比效果，完全不用重启服务或重新训练模型。

不过要注意的是，Langflow 本质上是一个原型设计工具。它的内置界面虽然功能完整，但样式固定、交互有限，不适合直接对外发布。真正让它“走出实验室”的方式，是将其作为后台推理引擎，通过 API 对外提供服务。

幸运的是，Langflow 内建了 RESTful 接口。每个保存的工作流都有一个唯一的FLOW_ID，可以通过/api/v1/process/{flow_id}这个端点触发执行。这意味着我们完全可以将它当作一个黑盒 AI 服务器来使用。

于是自然想到：能不能用一个极简的前端框架来消费这个 API？Streamlit 就是最理想的选择之一。

它只有两个核心优势：
1.极低门槛：几行st.write()就能出页面，Python 开发者几乎零学习成本；
2.状态管理友好：st.session_state让多轮对话的记忆变得轻而易举。

两者结合，形成了一种新的开发范式：Langflow 做大脑，Streamlit 做嘴巴和脸。

下面我们以“旅游指南聊天机器人”为例，走一遍完整流程。

首先准备环境：

pip install langflow streamlit requests streamlit-chat -Uq

启动 Langflow：

langflow run

访问http://localhost:7860，开始构建两个独立流程：

数据预处理流

使用PyPDFLoader加载《欧洲旅行指南》PDF；
通过RecursiveCharacterTextSplitter按段落拆分文本；
调用OpenAIEmbeddings生成向量；
存入Chroma向量数据库。

运行一次即可完成知识库初始化。

对话查询流

接收用户输入的question；
使用ConversationalRetrievalQAChain连接向量库和 LLM；
配合ConversationBufferMemory记住上下文；
输出结构化回答。

保存该流程后，从浏览器 URL 中提取FLOW_ID，例如：

http://localhost:7860/flow/f3a8b7e2-1c5d-4f3a-b9c1-d3e5f7a8b9c1

其中的 UUID 就是我们后续调用的关键凭证。

接下来进入 Streamlit 编程环节。新建app.py，代码如下：

import logging import sys import requests import streamlit as st from streamlit_chat import message # 日志配置 logging.basicConfig( format="%(asctime)s - %(levelname)s - %(message)s", level=logging.INFO, stream=sys.stdout ) # Langflow API 配置 BASE_API_URL = "http://localhost:7860/api/v1/process" FLOW_ID = "f3a8b7e2-1c5d-4f3a-b9c1-d3e5f7a8b9c1" # 替换为你的实际 Flow ID # 可选：动态调整模型参数 TWEAKS = { "ChatOpenAI-X1Y2Z": { "model_name": "gpt-3.5-turbo", "temperature": 0.7 }, "ConversationalRetrievalChain-ABC123": {} } AVATAR_USER = "https://raw.githubusercontent.com/garystafford-aws/static-assets/main/static/people-64px.png" AVATAR_BOT = "https://raw.githubusercontent.com/garystafford-aws/static-assets/main/static/traveler-64px.png" def run_langflow(prompt: str, flow_id: str, tweaks: dict = None) -> dict: """ 调用 Langflow API 执行指定流程 """ url = f"{BASE_API_URL}/{flow_id}" payload = { "inputs": {"question": prompt} } if tweaks: payload["tweaks"] = tweaks try: response = requests.post(url, json=payload, timeout=30) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: logging.error(f"API request failed: {e}") return {"error": str(e)} def generate_response(user_input: str): """ 生成机器人回复 """ logging.info(f"User question: {user_input}") result = run_langflow(user_input, FLOW_ID, TWEAKS) if "error" in result: return "抱歉，我在查找答案时遇到了问题，请稍后再试。" try: answer = result["result"]["answer"] logging.info(f"Assistant answer: {answer}") return answer except KeyError: logging.error(f"Unexpected response format: {result}") return "无法解析回答内容，请检查 Langflow 流程配置。" # Streamlit 页面主体 def main(): st.set_page_config(page_title="🌍 旅行助手", layout="centered") st.title("✈️ 欢迎使用智能旅行指南") st.markdown("我可以帮你解答关于城市、景点、签证、交通等问题！") # 初始化聊天记录 if "messages" not in st.session_state: st.session_state.messages = [] # 显示历史消息 for msg in st.session_state.messages: message( msg["content"], is_user=msg["role"] == "user", avatar_style="initials", seed=msg["avatar_seed"], key=msg["key"] ) # 用户输入 if prompt := st.chat_input("请输入你的旅行问题..."): # 添加用户消息到历史 user_msg = { "role": "user", "content": prompt, "avatar_seed": "U", "key": f"user_{len(st.session_state.messages)}" } st.session_state.messages.append(user_msg) # 显示用户消息 message( prompt, is_user=True, avatar_style="initials", seed="U", key=user_msg["key"] ) # 显示机器人思考状态 with st.spinner("正在为你查询最佳旅行建议..."): assistant_response = generate_response(prompt) # 添加并显示机器人回复 bot_msg = { "role": "assistant", "content": assistant_response, "avatar_seed": "T", "key": f"bot_{len(st.session_state.messages)}" } st.session_state.messages.append(bot_msg) message( assistant_response, is_user=False, avatar_style="initials", seed="T", key=bot_msg["key"] ) if __name__ == "__main__": main()

这段代码有几个值得强调的设计点：

会话状态持久化：利用st.session_state.messages自动维护聊天记录，刷新页面也不会丢失；
错误隔离机制：网络请求失败时返回友好提示，避免前端崩溃；
日志透明化：所有关键步骤都会打印日志，便于排查问题；
扩展性强：TWEAKS字段预留了运行时修改模型参数的空间。

运行命令：

streamlit run app.py

打开http://localhost:8501，你就拥有一个具备完整上下文记忆能力的旅游聊天机器人了。

当然，真实项目中还可以加入更多工程实践来提升健壮性和用户体验。

动态切换回答风格

你可以在侧边栏添加一个选择器，让用户决定回答是“严谨”还是“活泼”：

mode = st.sidebar.selectbox("回答风格", ["严谨", "创意"]) tweaks_map = { "严谨": { "ChatOpenAI-X1Y2Z": {"temperature": 0.3, "model_name": "gpt-3.5-turbo"} }, "创意": { "ChatOpenAI-X1Y2Z": {"temperature": 1.0, "model_name": "gpt-4"} } }

然后在调用 API 时传入对应的tweaks，实现同一流程、多种行为。

加入重试机制

网络不稳定时，简单的重试策略可以显著提高成功率：

from tenacity import retry, stop_after_attempt, wait_fixed @retry(stop=stop_after_attempt(3), wait=wait_fixed(2)) def run_langflow_with_retry(prompt, flow_id, tweaks=None): return run_langflow(prompt, flow_id, tweaks)