据我所知，Streamlit 是启动可定制网络应用的最快方法。如果你想创建一个人工智能代理并将其部署在自己的前端，我想没有比这更好的选择了。
唯一的障碍是聊天元素库。它们专门针对OpenAI的应用程序接口和 Python 客户端进行了调整。
这很好--只需几行代码就能与一些最负盛名的技术进行交互，嗯......很好。
但这并不是全部。
如果您想对机器人进行更多控制，该怎么办？例如，您可能需要多步骤工作流或检索增强生成（RAG）。 这些新增的功能层通常意味着要将具有各种依赖关系的库整合在一起。
还是这样？
在本教程中，我将构建一个由 Streamlit 托管的聊天机器人客户端。我将向你展示一个用于快速迭代和高度自定义聊天机器人的界面。然后，你将学习如何使用定制的OpenAI Python 客户端集成聊天机器人。
如果您正在设计原型，那么依赖性和技术性问题不应该成为您的障碍。
此外，本着快速原型开发的精神，如果你想跳过教程开始动手，代码可以在GitHub上找到。
Bombs away 💣
第 1 步：构建聊天机器人逻辑
无论是工作流程自动化还是预约聊天机器人，这里都是你的天地。
如果您想寻找灵感，我建议您探索一下GenAI 聊天机器人的广泛用例。为了简单起见，我将带着我的侍酒师维诺娜（Winona）回到这里。
只需几步就能实现我们复杂、实用的小机器人。我的介绍会很简短，但你可以浏览许多超有用的冗长教程。
1.向它发出指令
在工作室中，我们将导航到左侧边栏的 "主页"。
您应该会看到 "说明 "部分位于正中间。点击它，添加或修改纯文本说明。
这为我们的机器人提供了指令、个性和防护。使用通俗易懂的语言，你可以非常有效地引导机器人做出预期的行为。让它听起来更人性化，并
2.建立流程
这是机器人个性的核心所在：获取特定信息、严格的步骤、代码执行等。
不要低估简单的力量。单个自治节点的功能可与推理代理相媲美。我的知识库（KB）就连接了一个节点。
3.添加知识库
如果说说明是关于氛围，那么知识库则是关于冷冰冰的事实。在我的案例中，所涉及的事实就是 "葡萄酒评论 "数据集中的葡萄酒，这是一份包含葡萄酒、描述和价格的列表。我将把它视为事实上的葡萄酒清单，供我们的机器人进行侍酒师化处理。
我会点击左侧面板中的 "表"，然后点击页面左上角的 "新建表"，并给它起一个描述性的名字。
点击右上角的垂直省略号 (⋮)并点击 "导入"。
将您的 .csv 拖入弹出的模式，然后按照屏幕上的步骤操作。
要让机器人访问表格，请导航到左侧边栏的知识库。
单击绿色表格小图标并选择相关来源。单击添加表格。
确保您的流程可以访问知识库，然后就可以使用了。
第 2 步：添加Chat 应用程序接口集成
机器人与本地客户端之间的联系点是Chat API。要将其添加到机器人中，我需要滚动到 "通信渠道"，然后点击...更多。
如果您愿意，请仔细阅读这些集成。我们的目标是 Chat.我翻了好久才找到。
点击集成，然后在弹出的模式窗口中点击 "安装集成"。
安装完成后，您会在webhook URL 的末尾看到Chat API ID。稍后您将需要它。
步骤 3：编写 Python 客户端
Chat API 公开了许多端点，用于对用户、会话和消息执行粗暴的操作。按照约定，我将把这些端点打包成一个 Python 客户端，以取代OpenAI 客户端。
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,
}
你可以在 .env 文件中添加Chat API ID，这有助于调试，但并非绝对必要。我们将在构建 Streamlit 应用程序时处理 API ID 和用户密钥。
我一直 BASE_URI 和 标题 在单独的 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（将在应用程序中以编程方式访问），并在发生时生成传入数据。
步骤 4：创建 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" 用户名
用户名 = "您的用户密钥"繁重的工作正在进行：
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 文件来反映这一点：
用户
键 = "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
这将用于创建用户并将其添加到秘密文件中。
步骤 5：运行应用程序
逻辑已经建立，用户也已创建，是时候体验一下这个应用程序了。安装依赖项，然后运行：
streamlit run app.py
就这样，你将看到我们的机器人的全部光彩。
立即运行 Streamlit 聊天机器人
如果您正在使用 Streamlit 进行原型开发，那么您就会知道，自定义功能不应以牺牲便利性为代价。聊天机器人是用来解决问题的，而不是制造问题。
Botpress 具有可视化拖放生成器、数十种官方集成和可访问的 API 端点。这样，您就可以在多种通信渠道中进行构建、迭代和部署。
今天就开始建设。它是免费的。
.svg)
