據我所知，Streamlit 是讓可客製化的 Web 應用程式落地的最快方法。如果您想要建立一個 AI 代理並部署在自己的前端，我想不出有比這更好的選擇了。
唯一的缺點是聊天元素函式庫。它們是特別針對OpenAI的 API 和 Python 用戶端調整的。
這很不錯，只需要幾行程式碼就能與一些最負盛名的技術互動，這真是......太棒了。
但這並非一切。
如果您想要對您的機器人有更多的控制權呢？舉例來說，您可能想要一個多步驟的工作流程，或檢索增量生成(RAG)。 這些新增的功能層級通常意味著要將各式各樣依賴的函式庫整合在一起。
或者有嗎？
在本教程中，我將建立一個 Streamlit 託管的聊天機器人客戶端。我會向您展示一個快速迭代和高度客製化的聊天機器人介面。接著，您將學習如何使用自訂的OpenAI Python 用戶端整合聊天機器人。
如果您正在進行原型設計，依賴性和技術問題不應該阻礙您。
此外，基於快速原型設計的精神，如果您想跳過教學開始修補，程式碼就在GitHub上。
轟炸 💣
步驟 1：建立聊天機邏輯
無論是工作流程自動化或約會預約聊天機器人，這裡都是您的天地。
如果您正在尋找靈感，我建議您探索GenAI 聊天機器人的廣泛使用案例。為了簡單起見，我將帶著我的侍酒師Winona 回來。
只需幾個步驟，就可以完成我們這個精密、有用的小機器人。我會簡短說明，但您可以瀏覽許多冗長、超有用的教學。
1.給它指示
在工作室中，我們將導航到左側邊欄中的首頁。
您應該會看到「說明」部分置於正中央。按一下即可新增或修改純文字說明。
這給予我們的機器人指令、個性和警戒線。使用簡單的語言，您就可以非常有效地將機器人引導至想要的行為。讓它聽起來更人性化，而且
2.建立流程
這是機器人個性的核心所在：存取特定資訊、僵化的步驟、執行程式碼等。
不要低估簡單的力量。單一自主節點的功能可與推理代理相媲美。我的知識庫 (KB) 就有一個獨立節點。
3.新增知識庫
如果說明是關於氣氛，KB 則是關於冷冰冰的事實。在我的案例中，有關的事實就是Wine Reviews 資料集中的葡萄酒，這是一份包含葡萄酒、描述和價格的清單。我將把它視為事實上的葡萄酒清單，供我們的機器人進行 Sommelier 化。
我會按一下左側面板中的表格，然後按下頁面左上方的「新增表格」，並為它取一個描述性的名稱。
按一下右上方的垂直省略號 (⋮)，然後按一下匯入。
將您的 .csv 拖曳至彈出的模組，並依照螢幕上的步驟操作。
若要讓您的機器人可以存取表格，請導航至左側側邊欄的知識庫。
按一下綠色表格小圖示，然後選擇相關來源。按一下新增表格。
確保您的流程可以存取知識庫，您就可以使用了。
步驟 2：新增Chat API 整合
機器人與本機用戶端之間的接觸點是Chat API。要將其加入我們的機器人，我會捲動到通訊管道，然後按下 ... 更多。
如果您願意，請仔細閱讀這些整合。我們的目標是 Chat.我必須捲動一下才能找到。
按一下整合，然後在彈出的模組中按下安裝整合。
安裝完成後，您會在webhook URL 結尾看到Chat API ID。您稍後需要使用該 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 和 標題 在單獨的 常數.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 監聽器
這是 hackery 的範圍。為了聆聽會話更新並將其循環到 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"
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 檔案來反映：
[[使用者]]
鍵 = "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)
.webp)
.webp)
