一个基于Streamlit开发的LLM小玩具

Yuan.Sn
  2025-10-11 16:01:29 2025-10-11 16:01:29 Created   2025-10-14 17:30:10 2025-10-14 17:30:10 Updated    1.2k Words  5 Mins  

最近在逛 Reddit 的时候, 偶然发现一个基于Python的框架 Streammit. 它能够用几行代码就能生成一个 可交互的，可Serveless的 专注于数据分析的前端网站

image
image

Streamlit 数据分析

简单看了下 Youtube上的 视频 和 文档 , 搓了一个简单的 Demo

image
image

简单的几行就能够写出一个完整的数据分析页面，还能部署到 Serverless服务器中，还挺酷🤗

基于Streamlit 的 LLM CHTA BOT

最近正好在研究 LLM 的 API 开发，正好拿这个来练练手。

API调用程序及其前端页面

a. 先创建一个 用于连接的客户端的函数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
def get_claude_client():
    """获取Claude客户端"""
    try:
        # 简化密钥获取：会话密钥 > 默认密钥（不使用配置文件）
        api_key = st.session_state.get("current_api_key") or DEFAULT_API_KEY
        
        # 检查密钥类型并给出警告
        if api_key.startswith("sk-ant-sid"):
            st.sidebar.warning("⚠️ Session key可能无法直接通过API调用，建议使用正式API密钥")
        
        # 创建客户端
        client = anthropic.Anthropic(api_key=api_key)
        
        # 简单验证 - 不进行实际API调用
        if api_key and (api_key.startswith("sk-ant-api") or api_key.startswith("sk-ant-sid")):
            return client
        else:
            return None

b. 写主页面及其api调用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
# 聊天输入
    if prompt := st.chat_input("输入你的消息..."):
        # 检查请求频率限制
        current_time = datetime.datetime.now()
        if (st.session_state.last_request_time and 
            current_time - st.session_state.last_request_time < MIN_TIME_BETWEEN_REQUESTS):
            
            remaining = MIN_TIME_BETWEEN_REQUESTS - (current_time - st.session_state.last_request_time)
            st.warning(f"请等待 {remaining.total_seconds():.1f} 秒后再发送消息")
            return
        
        # 添加用户消息
        user_msg = {
            "role": "user", 
            "content": prompt,
            "timestamp": current_time
        }
        st.session_state.messages.append(user_msg)
        
        # 显示用户消息
        with st.chat_message("user"):
            st.caption(f"🕒 {format_time(current_time)}")
            st.markdown(prompt)
        
        # 生成助手回复
        with st.chat_message("assistant"):
            timestamp_placeholder = st.empty()
            message_placeholder = st.empty()
            
            try:
                # 准备上下文（限制历史长度）
                context_messages = []
                recent_messages = st.session_state.messages[-HISTORY_LENGTH:]
                
                for msg in recent_messages:
                    if msg["role"] in ["user", "assistant"]:
                        context_messages.append({
                            "role": msg["role"],
                            "content": msg["content"]
                        })
                
                # 调用Claude API
                with st.spinner("🤔 Claude正在搬砖..."):
                    response = claude_client.messages.create(
                        model="claude-3-haiku-20240307",
                        max_tokens=2000,
                        system=INSTRUCTIONS,
                        messages=context_messages
                    )
                
                # 显示回复
                assistant_reply = response.content[0].text
                reply_time = datetime.datetime.now()
                
                timestamp_placeholder.caption(f"🕒 {format_time(reply_time)}")
                message_placeholder.markdown(assistant_reply)
                
                # 保存助手消息
                assistant_msg = {
                    "role": "assistant",
                    "content": assistant_reply,
                    "timestamp": reply_time
                }
                st.session_state.messages.append(assistant_msg)
                
                # 更新请求时间
                st.session_state.last_request_time = current_time
                
            except Exception as e:
                error_msg = f"抱歉，发生了错误: {str(e)}"
                timestamp_placeholder.caption(f"🕒 {format_time(datetime.datetime.now())}")
                message_placeholder.error(error_msg)

c.完善侧边栏API切换

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
def show_api_key_manager():
    """显示API密钥管理界面"""
    with st.sidebar:
        st.markdown("### 🔑 API密钥管理")
        
        # 显示当前使用的密钥（部分隐藏）
        current_key = st.session_state.get("current_api_key") or DEFAULT_API_KEY
        
        if current_key:
            masked_key = current_key[:12] + "..." + current_key[-4:] if len(current_key) > 16 else current_key
            key_type = "API密钥" if current_key.startswith("sk-ant-api") else "Session密钥"
            st.info(f"当前密钥: {masked_key}\n类型: {key_type}")
        
        # # 密钥类型说明
        # with st.expander("💡 Session Key获取说明"):
        #     st.markdown("""
        #     自己google，哈哈哈
        #     """)
        
        # 输入新密钥
        new_key = st.text_input(
            "输入你的API KEY",
            type="password",
            help="填入API(sk-ant-api...)"
        )
        
        col1, col2 = st.columns(2)
        
        with col1:
            if st.button("✅ 应用新密钥", disabled=not new_key):
                if new_key and (new_key.startswith("sk-ant-api") or new_key.startswith("sk-ant-sid")):
                    st.session_state.current_api_key = new_key
                    st.success("密钥已更新！")
                    st.rerun()
                else:
                    st.error("密钥格式不正确")
        
        with col2:
            if st.button("🔄 重置为默认"):
                if "current_api_key" in st.session_state:
                    del st.session_state.current_api_key
                st.success("已重置为默认密钥")
                st.rerun()

然后 streamlit run claude_chat.py 就能在本地运行啦~🤗

image
image

200行代码，能写出一个 前端的 CHAT BOT，真不错~~😄

然后这次试试把它通过Dockers 部署到自己的服务器上

Docker 部署

通过 文档 , 可以将应用部署到 Docker当中。
大致流程就是写一个 Dockerfile，在服务器中批处理运行就好了

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
FROM python:3.9-slim

WORKDIR /app

RUN apt-get update && apt-get install -y \
    build-essential \
    curl \
    software-properties-common \
    git \
    && rm -rf /var/lib/apt/lists/*

RUN git clone https://github.com/streamlit/streamlit-example.git .

RUN pip3 install -r requirements.txt

EXPOSE 8501

HEALTHCHECK CMD curl --fail http://localhost:8501/_stcore/health

ENTRYPOINT ["streamlit", "run", "streamlit_app.py", "--server.port=8501", "--server.address=0.0.0.0"]

需要注意的是，如果需要 nginx 反代域名的话，需要配置 WebSocket 的反向代理，就是在http 块中添加一个 map 指令

1
2
3
4
5
6
7
8
9
10
11
12
13
14
http {
    # ... 其他 http 配置 ...

    map $http_upgrade $connection_upgrade {
        default upgrade;
        ''      close;
    }

    server {
        # ... 你的 server 配置 ...
    }

    # ... 其他 http 配置 ...
}

然后！！！

LET'S CHAT!!!

酷吧🤗

推荐阅读
25.10.20_Presentation 25.10.20_Presentation 谈谈Nginx 谈谈Nginx Yuan's IMG HOST 部署日志 Yuan's IMG HOST 部署日志
推荐阅读
25.10.20_Presentation 25.10.20_Presentation 谈谈Nginx 谈谈Nginx
Comments
On this page
一个基于Streamlit开发的LLM小玩具
  1. Streamlit 数据分析
  2. 基于Streamlit 的 LLM CHTA BOT
    1. API调用程序及其前端页面
      1. a. 先创建一个 用于连接的客户端的函数
      2. b. 写主页面及其api调用
      3. c.完善侧边栏API切换
    2. Docker 部署
  3. LET'S CHAT!!!