10分钟内构建一个Streamlit聊天机器人

据我所知，Streamlit是最快速实现可定制Web应用的方式。如果你想构建AI聊天机器人并将其部署到自己的前端，我想不到比这更好的选择了。

唯一的限制在于聊天元素库。它们基本上是专门为OpenAI的API和Python客户端设计的。

这当然很好——只需几行代码就能与顶尖技术交互，这本身就很棒。

但这还不是全部。

如果你希望对你的机器人有更多的控制权怎么办？比如，你可能需要一个多步骤的工作流程，或者检索增强生成（RAG）。这些额外的功能通常意味着需要整合各种依赖库。

真的是这样吗？

在本教程中，我将构建一个基于Streamlit的聊天机器人客户端。我会展示一个便于快速迭代和高度自定义的聊天界面。接着，你将学习如何通过自定义的OpenAI风格Python客户端集成聊天机器人。

如果你正在做原型开发，不必被依赖和技术细节所束缚。

而且，为了快速原型开发，如果你想跳过教程直接上手，代码已在GitHub。

准备好了 💣

第一步：构建聊天机器人逻辑

无论是用于工作流自动化还是预约预订聊天机器人，这里的可能性几乎无限。

如果你需要灵感，建议你多看看生成式AI聊天机器人的应用场景。为了简单起见，我会继续用我那希望已经小有名气的侍酒师Winona举例。

我们这个聪明又贴心的小机器人只需几个步骤就能实现。我会简要说明，你也可以参考很多详细且极具帮助的教程。

1. 添加指令

在Studio中，我们在左侧边栏点击首页。

你应该能在正中央看到指令部分。点击它即可用纯文本添加或修改指令。

这为我们的机器人提供了指令、个性和行为边界。通过自然语言，你可以有效引导机器人实现预期行为，让它更像人类。

2. 构建流程

这里是机器人个性的核心：访问特定信息、严格的步骤流程、代码执行等。

不要低估简单的力量。单个自主节点可媲美推理代理的功能。我就有一个节点连接到了我的知识库（KB）。

3. 添加知识库

如果说指令决定氛围，知识库则关乎事实。以我为例，知识库内容是葡萄酒评论数据集中的葡萄酒、描述和价格。我会把它当作机器人侍酒师的葡萄酒库存。

我会点击左侧面板的表格，然后在页面左上角点击新建表格，并为其命名。

点击右上角的竖向省略号（⋮），选择导入。

将你的.csv文件拖入弹窗，按屏幕提示操作即可。

要让机器人访问该表格，前往左侧边栏的知识库。

点击绿色的表格图标，选择相应的数据源。点击添加表格。

确保你的流程已能访问知识库，就可以开始了。

第二步：添加Chat API集成

机器人与本地客户端的连接点就是Chat API。要将其添加到机器人中，向下滚动到通信渠道，点击… 更多。

你可以浏览各种集成。我们需要的是Chat。我找了一会儿才看到它。

点击该集成，在弹窗中选择安装集成。

安装后，你会在webhook URL末尾看到Chat API ID。后面会用到。

第三步：编写Python客户端

Chat API 提供多个端点，可对用户、会话和消息进行增删改查操作。如前所述，我会将这些封装进一个可替代OpenAI客户端的Python客户端。

1. 添加凭证

class BotpressClient:
   def __init__(self, api_id=None, user_key=None):
       self.api_id = api_id or os.getenv("CHAT_API_ID")
       self.user_key = user_key or os.getenv("USER_KEY")
       self.base_url = f"{BASE_URI}/{self.api_id}"
       self.headers = {
           **HEADERS,
           "x-user-key": self.user_key,
       }

你可以把Chat API ID写进.env文件——这样调试更方便，但不是必须的。我们会在构建Streamlit应用时处理API ID和用户密钥。

我把BASE_URI和HEADERS放在单独的constants.py文件里，避免混乱。

# constants.py

BASE_URI = "https://chat.botpress.cloud"
HEADERS = {
   "accept": "application/json",
   "Content-Type": "application/json",
}

2. 创建请求方法

   def _request(self, method, path, json=None):
       url = f"{self.base_url}{path}"
       try:
           response = requests.request(method, url, headers=self.headers, json=json)
           response.raise_for_status()
           return response.json()
       except requests.HTTPError:
           return response.status_code, response.text


   # --- Core API Methods ---


   def get_user(self):
       return self._request("GET", "/users/me")


   def create_user(self, name, id):
       user_data = {"name": name, "id": id}
       return self._request("POST", "/users", json=user_data)


   def set_user_key(self, key):
       self.user_key = key
       self.headers["x-user-key"] = key


   def create_and_set_user(self, name, id):
       new_user = self.create_user(name, id)
       self.set_user_key(new_user["key"])


   def create_conversation(self):
       return self._request("POST", "/conversations", json={"body": {}})


   def list_conversations(self):
       return self._request("GET", "/conversations")


   def get_conversation(self, conversation_id):
       return self._request("GET", f"/conversations/{conversation_id}")


   def create_message(self, message, conversation_id):
       payload = {
           "payload": {"type": "text", "text": message},
           "conversationId": conversation_id,
       }
       return self._request("POST", "/messages", json=payload)


   def list_messages(self, conversation_id):
       return self._request("GET", f"/conversations/{conversation_id}/messages")

如前所述，这些方法基本都对应API的端点。我只是把它们封装进一个类里。

3. 创建SSE监听器

这就是全部的“黑科技”了。为了监听会话更新并与Streamlit前端循环交互，客户端需要一个方法来监听并返回来自机器人服务器推送事件的数据。

 def listen_conversation(self, conversation_id):
       url = f"{self.base_url}/conversations/{conversation_id}/listen"
       for event in sseclient.SSEClient(url, headers=self.headers):
           print(event.data)
           if event.data == "ping":
               continue
           data = json.loads(event.data)["data"]
	    yield {"id": data["id"], "text": data["payload"]["text"]}

这个函数接收conversation_id（将在应用内以编程方式获取），并实时返回收到的数据。

第四步：创建Streamlit应用

一切准备就绪后，就可以开始构建聊天机器人了。需要注意的是，我参考了Streamlit关于构建LLM聊天应用的官方指南，并做了一些功能增强。

1. 调整模板代码

理论上，只需对Streamlit示例中的模板做极少修改，就能让应用运行起来。

# app.py


from client import BotpressClient
import streamlit as st
from constants import CONVERSATION_ID


st.title("Botpress Front-end for Streamlit")


client = BotpressClient(
   api_id=st.secrets["CHAT_API_ID"], user_key=st.secrets["USER_KEY"]
)


if "messages" not in st.session_state:
   messages = client.list_messages(CONVERSATION_ID)
   next_token = messages["meta"]["nextToken"]
   st.session_state.messages = messages["messages"][::-1]


for message in st.session_state.messages:
   with st.chat_message(message["userId"]):
       st.markdown(message["payload"]["text"])


if prompt := st.chat_input("*wine*-d it up"):
   st.session_state.messages.append({"role": "user", "content": prompt})


   with st.chat_message("user"):
       st.markdown(prompt)


   client.create_message(prompt, conversation_id=CONVERSATION_ID)
  
   with st.chat_message("assistant"):
response_box = st.empty()
       last_rendered = ""
	
       for message in client.listen_conversation(CONVERSATION_ID):


           message_id = message["id"]
           message_text = message["text"]


           if message_id != last_rendered:
               last_rendered = message_id
               response_box.markdown(message_text)
               st.session_state.messages.append(
                   {"role": "assistant", "content": message_text}
               )

这里我们读取了密钥变量。所以请创建一个.streamlit/secrets.toml文件，并把变量写进去：

CHAT_API_ID = ”YOUR_API_ID”
USER_KEY = ”YOUR_USER_KEY”

核心逻辑在这里：

with st.chat_message("assistant"):
response_box = st.empty()
       last_rendered = ""
	
       for message in client.listen_conversation(CONVERSATION_ID):


           message_id = message["id"]
           message_text = message["text"]


           if message_id != last_rendered:
               last_rendered = message_id
               response_box.markdown(message_text)
               st.session_state.messages.append(
                   {"role": "assistant", "content": message_text}
               )

客户端通过这里连接聊天元素，实现消息的发送和接收。

这样可以运行，但有几个不理想的地方：

• 你必须单独新建一个会话。
• 旧消息的格式和新消息不同，因为它们没有角色标记（用户或助手）。
• 无法切换会话。

2. 动态创建会话

从零开始，我会自动创建新会话或打开最近的会话：

# app.py


from client import BotpressClient
import streamlit as st


st.title("Botpress Front-end for Streamlit")


client = BotpressClient(
   api_id=st.secrets["CHAT_API_ID"], user_key=st.secrets["users"][0]["key"]
)


# user info
user = client.get_user()
user_id = user["user"]["id"]


conversations = client.list_conversations()["conversations"]
conversation_ids = [conv["id"] for conv in conversations]




# conversation
def create_conversation():
   res = client.create_conversation()
   print(f"Created new conversation: {res}")
   conversation_id = res["conversation"]["id"]
   st.session_state.active_conversation = conversation_id
   st.session_state.messages = []
   st.rerun()




if not conversations:
   create_conversation()


if "active_conversation" not in st.session_state:
   st.session_state["active_conversation"] = conversations[0]["id"]

注意我修改了密钥，可以存储多个用户。你需要相应修改.streamlit/secrets.toml文件：

[[users]]
key = "your_user_key"

你可以根据需要多次添加这段代码，把用户作为表格数组存储。

3. 允许用户创建和切换会话

正如标题所说，这会在顶部创建一个下拉菜单和按钮，让你选择会话。

col1, col2 = st.columns([5, 1])
with col1:
   conversation_id = st.selectbox(
       "Select Conversation",
       options=[conv["id"] for conv in conversations],
       index=conversation_ids.index(st.session_state.active_conversation),
   )


with col2:
   st.markdown("<div style='height: 1.9em'></div>", unsafe_allow_html=True)
   if st.button("➕"):
       create_conversation()


selected_conversation = client.get_conversation(conversation_id)

4. 为历史消息分配正确角色

我们将通过为每条历史消息分配用户或助手角色，解决上面提到的格式问题：

if (
   "messages" not in st.session_state
   or st.session_state.get("active_conversation") != conversation_id
):
   st.session_state.active_conversation = conversation_id
   st.session_state.messages = []


   messages = client.list_messages(conversation_id)
   next_token = messages["meta"].get("nextToken")


   for message in messages["messages"][::-1]:
       role = "user" if message["userId"] == user_id else "assistant"
       text = message["payload"]["text"]
       st.session_state.messages.append({"role": role, "content": text})

这样代码就符合Streamlit的结构要求了。

5. 添加消息逻辑

这部分和之前差不多，只是适配了新的结构。

# display chat history
for message in st.session_state.messages:
   with st.chat_message(message["role"]):
       st.markdown(message["content"])


if prompt := st.chat_input("*wine*-d it up"):


   st.session_state.messages.append({"role": "user", "content": prompt})
   client.create_message(prompt, conversation_id=conversation_id)


   with st.chat_message("user"):
       st.markdown(prompt)


   with st.chat_message("assistant"):
       stream = client.listen_conversation(conversation_id=conversation_id)
       response = st.write_stream(stream)


   st.session_state.messages.append({"role": "assistant", "content": response})

5. 创建用户

逻辑已经就绪，但你需要先创建一个用户才能运行应用。我选择单独添加这一步，以模拟注册服务的体验。幸运的是，我还写了一个脚本：

# create_user.py


import argparse
from pathlib import Path
from client import *
from constants import *

secrets_path = Path(".streamlit") / "secrets.toml"

template = """[[users]]
key="{}"

"""

client = BotpressClient()

def create_user(name, id, add_to_secrets=True):
   res = client.create_user(name, id)
   if not add_to_secrets:
       return res

   secrets_path.touch(exist_ok=True)

   with open(secrets_path, "a") as f:
       f.write(template.format(res["user"]["id"], res["key"]))

   return res


if __name__ == "__main__":
   parser = argparse.ArgumentParser(
       description="Create a Botpress user and optionally store secrets."
   )
   parser.add_argument("--name", required=True, help="Display name of the user.")
   parser.add_argument(
       "--id", required=True, help="User ID. If omitted, one is generated by the API."
   )
   parser.add_argument("--chat_api_id", help="ID for the Botpress Chat API integration. Taken from `.env` file if not provided.")
   parser.add_argument(
       "--no-secrets",
       action="store_true",
       help="Do not append to .streamlit/secrets.toml.",
   )

   args = parser.parse_args()

   print(f"Creating user: {args.name} (ID: {args.id or 'auto-generated'})")
   result = create_user(name=args.name, id=args.id, add_to_secrets=not args.no_secrets)

   print("✅ User created:")
   print(result)

只要你有Chat API ID，就可以运行：

python create_user.py –name YOUR_NAME –id SOME_USER_ID –chat_api_id YOUR_CHAT_API_ID

这会负责创建用户并将其添加到你的密钥文件中。

第五步：运行应用

当你的逻辑已经构建好，用户也已创建完成，现在是时候试用一下这个应用程序了。安装依赖项，然后运行：

streamlit run app.py

就是这样，你将看到我们的机器人完整地展现出来。

今天就运行一个 Streamlit 聊天机器人

如果你正在用 Streamlit 做原型开发，你会知道自定义不应该以牺牲便利性为代价。聊天机器人是用来解决问题的——不是制造问题的。

Botpress 提供可视化拖拽式构建器、数十种官方集成和易用的 API 接口。这样你可以在多个沟通渠道上构建、迭代和部署。

立即开始构建。永久免费。

常见问题

为什么我会选择 Streamlit 而不是其他前端框架来构建聊天机器人？

如果你想要一个基于 Python 的快速解决方案，可以让你无需前端专业知识就能快速制作交互式应用原型，Streamlit 是不错的选择，因为它用极少的代码就能处理聊天组件和状态管理等界面元素。

Streamlit 聊天机器人适合用于生产环境还是只适合做原型？

Streamlit 聊天机器人非常适合做原型和内部工具，但如果是面向公众的生产级应用，且有大量访问或需要高级样式，可能还需要额外的反向代理、安全加固，甚至更强大的前端框架来应对规模化需求。

使用 Streamlit 构建的聊天机器人的外观和体验可以自定义到什么程度？

Streamlit 支持调整基础样式，如颜色、字体和布局，但灵活性不如传统 Web 框架；如果需要完全自定义的设计，则需要嵌入自定义 HTML/CSS 或 JavaScript，这虽然可行，但比直接用 Streamlit 内置组件要复杂一些。

我可以将 Streamlit 聊天机器人集成到现有网站中，还是只能独立运行？

Streamlit 聊天机器人通常作为独立的 Web 应用在自己的 URL 上运行，但你可以通过 iframe 嵌入到现有网站中，不过这需要处理样式和安全等问题，以确保用户体验流畅。

将 Streamlit 聊天机器人部署为公开使用需要多少费用？

如果在本地或 Streamlit Community Cloud 上托管小型应用，部署 Streamlit 聊天机器人可以免费。但如果要大规模公开使用，根据访问量和在线时间要求，在 Heroku、AWS 或 DigitalOcean 等云平台上的费用大约为每月 5–50 美元。