如何安装第三方软件包

AI应用实战

案例一：AI智能对话工具

from http.client import responses

import json
import streamlit as st
import os
import openai
from datetime import datetime

openai.api_key = "sk-"
openai.base_url = "https://api.deepseek.com"

#设置页面的配置项
st.set_page_config(
    page_title="ai_partner",
    page_icon="🔣",
    layout="wide",
    initial_sidebar_state="expanded",
    menu_items={}
)

#logo
st.logo("logo.png")

#大标题
st.title("AI智能对话")

#生成会话标识
def generate_session_name():
    return datetime.now().strftime("%Y-%m-%d_%H-%M-%S")

#保存会话信息
def save_session():
    if st.session_state.current_session:
        # 构建新的会话对象
        session_data = {
            "nick_name": st.session_state.nick_name,
            "nature": st.session_state.nature,
            "current_session": st.session_state.current_session,
            "messages": st.session_state.messages
        }

        # 如果sessions目录不存在，则创建
        if not os.path.exists("AI智能伴侣/sessions"):
            os.mkdir("AI智能伴侣/sessions")

        # 保存会话数据
        with open("sessions/" + st.session_state.current_session + ".json", "w", encoding="utf-8") as f:
            json.dump(session_data, f, ensure_ascii=False, indent=4)

#加载所有会话列表信息
def load_sessions():
    session_list = []
    #加载session目录下的列表
    if os.path.exists("AI智能伴侣/sessions"):
        file_list = os.listdir("AI智能伴侣/sessions")
        for filename in file_list:
            if filename.endswith(".json"):
                session_list.append(filename[:-5])
    session_list.sort(reverse=True) #排序，降序排序
    return session_list

#加载指定的会话信息
def load_session(session_name):
    try:
        if os.path.exists(f"AI智能伴侣/sessions/{session_name}.json"):
            #读取会话
            with open("sessions/" + session_name + ".json", "r", encoding="utf-8") as f:
                session_data = json.load(f)
                st.session_state.nick_name = session_data["nick_name"]
                st.session_state.nature = session_data["nature"]
                st.session_state.messages  = session_data["messages"]
                st.session_state.current_session = session_name
    except Exception:
        st.error("加载会话失败")

#删除会话信息
def delete_session(session_name):
    try:
        if os.path.exists(f"AI智能伴侣/sessions/{session_name}.json"):
            os.remove(f"AI智能伴侣/sessions/{session_name}.json") #删除文件
            #如果删除的是当前会话，则需要更新消息列表
            if session_name == st.session_state.current_session:
                st.session_state.messages = []
                st.session_state.current_session = generate_session_name()
    except Exception:
        st.error("删除会话失败")

#系统提示词
system_prompt = """
    你叫%s，现在是我的真实伴侣，请完全代入伴侣角色。：
    规则：
        1. 每次只回1条消息
        2. 禁止任何场景或状态描述性文字
        3. 匹配用户的语言
        4. 回复简短，像微信聊天一样
        5. 有需要的话可以用❤️🌸等emoji表情
        6. 用符合伴侣性格的方式对话
        7. 回复的内容，要充分体现伴侣的性格特征
    
    伴侣性格：
        %s
    你必须严格遵守上述规则来回复用户。
"""

#初始化聊天信息
if "messages" not in st.session_state:
    st.session_state.messages = [] #所有需要展示的信息（用户的提示词，大模型的回复都存储在messages列表中）
#昵称
if "nick_name" not in st.session_state:
    st.session_state.nick_name = "小之之"
#性格
if "nature" not in st.session_state:
    st.session_state.nature = "温柔可爱的南方姑娘"
#会话标识
if "current_session" not in st.session_state:
    st.session_state.current_session = generate_session_name()

#展示聊天信息（遍历messages列表）
st.text(f"会话名称:{st.session_state.current_session}")
for message in st.session_state.messages: # {"role":"user","content":"prompt"}
    st.chat_message(message["role"]).write(message["content"])

#左侧侧边栏--with：streamlit中上下文管理
with st.sidebar:
    #会话信息
    st.subheader("控制面板")

    #新建会话
    if st.button("新建会话", width="stretch", icon="✏"):
        #1.保存当前会话信息
        save_session()

        #2.新建新的会话
        if st.session_state.messages: #如果聊天信息非空，true
            st.session_state.messages = []
            st.session_state.current_session = generate_session_name()
            save_session()
            st.rerun() #重新运行当前页面

    #会话历史
    st.text("会话历史")
    session_list = load_sessions()
    for session in session_list:
        col1, col2 = st.columns([4,1])
        with col1:
            #加载会话信息
            #三元运算符:条件为真返回第一个表达式的值，否则返回第二个表达式的值  语法：值1 if 条件 else 值2
            if st.button(session, width="stretch", icon="📃", type = "primary" if session == st.session_state.current_session else "secondary"):
                load_session(session)
        with col2:
            #删除会话信息
            if st.button("", width="stretch", icon="❌", key=session):
                delete_session(session)
                st.rerun()

    #分割线
    st.divider()

    st.subheader("伴侣信息")
    #昵称输入框
    nick_name = st.text_input("昵称", placeholder="请输入昵称", value=st.session_state.nick_name)
    if nick_name:
        st.session_state.nick_name = nick_name
    #性格输入框
    nature = st.text_area("性格", placeholder="请输入性格", value=st.session_state.nature)
    if nature:
        st.session_state.nature = nature

#消息输入框
prompt = st.chat_input("请输入您要问的问题")
if prompt: #字符串会自动转换为bool值，如果字符串为空，则为true，字符串非空，则为false
    st.chat_message("user").write(prompt)#char_message展示消息 user assistance
    print("-----------提示词：",prompt)#print会输出在终端中，便于调试使用

    #保存用户输入的提示词
    st.session_state.messages.append({"role":"user", "content":prompt.replace("~","~\u200B")})

    #调用AI大模型
    responses = openai.chat.completions.create(
        model="deepseek-v4-pro",  # 或 "deepseek-reasoner" 用于推理模型
        messages=[
            {"role": "system", "content": system_prompt % (st.session_state.nick_name, st.session_state.nature)},
            *st.session_state.messages
        ],
        stream=True
    )

    #输出大模型返回结果（非流式输出）
    #print("大模型返回结果:",responses.choices[0].message.content)
    #st.chat_message("assistant").write(responses.choices[0].message.content)

    # 输出大模型返回结果（流式输出）
    #创建一个空的组件，用于展示大模型返回的结果
    responses_message = st.empty()

    fun_response = ""
    for chunk in responses:
        if chunk.choices[0].delta.content is not None:
            content = chunk.choices[0].delta.content.replace("~","~\u200B")
            fun_response += content
            responses_message.chat_message("assistant").write(fun_response)

    #保存大模型返回的结果
    st.session_state.messages.append({"role": "assistant", "content": fun_response.replace("~","~\u200B")})

    #保存会话信息
    save_session()

案例二：汉字猜谜AI对话游戏

#FileResponse  从服务端发送文件内容到客户端
import os
import json

from git import exc
from openai import OpenAI
from typing import Any
from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
from fastapi.responses import FileResponse
from datetime import datetime
from pydantic import BaseModel
import logging

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

#日志记录：为了能够灵活的控制项目中日志的输出，通过官方提供的logging模块来输出日志
#日志级别：常见的有：DEBUG,INFO,WARNING,ERROR,FATAL(日志级别依次升高)
# asctime 时间   filename 文件名   lineno 行号   message 日志信息
#配置日志的基本信息
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(filename)s:%(lineno)d - %(message)s",
)

client = OpenAI(
    api_key = "sk-",
    base_url = "https://api.deepseek.com"
)


#创建FastAPI实例
app = FastAPI(title="汉字谜盒")

#使用StaticFiles从目录中提供静态文件
app.mount("/static", StaticFiles(directory="static"), name="static")

#创建会话存放的目录 sessions
if not os.path.exists("./sessions"):
    os.mkdir("./sessions")

#生成会话标识
def generate_session_id():
    return datetime.now().strftime("%Y%m%d_%H%M%S")


#根据session_id获取文件名
def get_session_file_name(session_id):
    return f"sessions/{session_id}.json"


# 系统提示词
SYSTEM_PROMPT = """
你是一个专门玩猜字谜的AI小助手，只进行字谜互动，不闲聊无关内容，全程纯文本交互。
请严格遵守以下规则：

一、出题规则
1. 开场先友好打招呼，并随机出一道常见、简单、适合大众的字谜，不生僻、不低俗、不使用网络烂梗。
2. 题目格式："谜面"（打一字）。
3. 每次出题必须完全随机，禁止重复使用相同题目；你需要在对话上下文中主动记录已使用过的谜语，确保同一会话内绝对不重复。
4. 避免使用高频重复的经典老谜语，尽量选择多样化的简单常见谜语。

二、判断题规则（最重要！）
1. 判断题，只看用户输入中的核心汉字，忽略无关内容：
   - 比如用户输入"江字""江""jiang"，都视为答案是「江」；
   - 用户输入"是江吗？""应该是江"，也视为答案是「江」。
2. 核心字与正确答案完全一致 → 判为正确，回复："太棒了！答对了！就是'XX'字！要不要再来一题？"
3. 核心字与正确答案不一致 → 判为错误，回复："不对哦，再想想～给你个小提示：[简短线索，不泄露答案]"
4. 用户说"不知道""公布答案"：先揭晓谜底和解释，再问"要不要再来一题？"

三、互动流程
1. 用户答对：夸奖 + 确认正确 + 询问"要不要再来一题？"
2. 用户答错：告知不对 + 简单提示 + 鼓励继续猜
3. 用户说"提示一下"：给出简短线索，不公布答案
4. 用户说"公布答案"或"不知道"：揭晓谜底并解释 + 询问"要不要再来一题？"
5. 用户说"换一题"或"再来一题"：立即更换新字谜

四、其他要求
1. 语气轻松有趣，简洁明快，不啰嗦。
2. 全程只围绕字谜，不回答其他问题，不聊无关话题。
3. 不使用多余表情符号，保持简洁。
4. 若用户答案与正确答案仅差一字或笔画，请仔细核对是否正确。

请严格按以上规则回复，优先保证谜语的随机性和多样性。
"""


#数据模型
class ApiResponse(BaseModel):
    code: int
    message: str
    data: Any

#数据模型
class ChatRequest(BaseModel):
    session_id: str
    message: str

#定义路径操作函数
@app.get("/")
def root():
    logging.info("访问项目首页")
    #使用FileResponse引用index.html渲染到页面
    return FileResponse("static/index.html")


#创建会话
#[get查询 post新增 delete put修改]
@app.post("/api/sessions")
def create_session():
    logging.info("创建会话")
    #1.生成会话的标识(名字)(当前时间)
    session_id = generate_session_id()

    #2.组装会话信息，保存到文件
    session_data = {
        "current_session": session_id,
        "messages": []
    }
    with open(f"sessions/{session_id}.json", "w", encoding="utf-8") as f:
        json.dump(session_data, f, ensure_ascii=False, indent=2)

    #3.返回数据,以对象的形式返回
    return ApiResponse(code=200,message="创建会话成功",data=session_id)

#与AI交互
@app.post("/api/chat")
def chat(request: ChatRequest) -> ApiResponse:
    logging.info(f"与AI交互: {request.session_id} : {request.message}")

    # 逻辑实现  --->  与AI大模型交互
    # 加载json文件中的会话数据->构建AI大模型交互的消息数据->调用AI大模型 deepseek->获取响应数据->更新消息列表中的消息->保存会话信息到json文件->响应数据
    #1.加载json文件中的会话数据
    session_path = get_session_file_name(request.session_id)
    with open(session_path, "r", encoding="utf-8") as f:
        session_data = json.load(f)

    #2.构建AI大模型交互的消息数据
    messages = [{"role": "system", "content": SYSTEM_PROMPT}]
    for message in session_data["messages"]:
        messages.append(message)
    messages.append({"role": "user", "content": request.message})

    #3.调用AI大模型 deepseek
    logging.info(f"请求的会话信息:, {messages}")
    response = client.chat.completions.create(
        model="deepseek-chat",  # 或 "deepseek-reasoner" 用于推理模型
        messages=messages,
        stream=False, #非流式输出
        temperature=1.5 #大模型的温度，越大生成的内容越有创造性、多样性、创意
    )

    #4.获取响应数据
    ai_response = response.choices[0].message.content
    logging.info(f"ai大模型相应的数据:, {ai_response}")

    #5.更新消息列表中的消息
    messages.pop(0)
    messages.append({"role": "assistant", "content": ai_response})
    session_data["messages"] = messages
    logging.info(f"更新后的会话信息:, {messages}")

    #6.保存会话信息到json文件
    with open(session_path, "w", encoding="utf-8") as f:
        json.dump(session_data, f, ensure_ascii=False, indent=2)

    # 7.响应数据
    return ApiResponse(code=200, message="请求成功", data=ai_response)

#获取会话列表
@app.get("/api/sessions")
def get_sessions() -> ApiResponse:
    logging.info("获取会话列表")
    # 1.获取sessions目录下的所有文件名
    session_files = os.listdir("sessions")

    #2.获取文件名中的会话id
    session_ids = [file.split(".")[0] for file in session_files]
    session_ids.sort(reverse=True)

    #返回数据
    return ApiResponse(code=200,message="获取会话列表成功",data=session_ids)

#获取指定的会话信息  [路径参数 用{},为通配符]
@app.get("/api/sessions/{session_id}")
def get_session(session_id: str) -> ApiResponse:
    logging.info(f"获取指定的会话信息：{session_id}")

    #1.获取会话文件名
    session_file = get_session_file_name(session_id)

    #2.读取会话文件
    with open(session_file, "r", encoding="utf-8") as f:
        session_data = json.load(f)

    #3.返回数据
    return ApiResponse(code=200, message="获取会话信息成功",data=session_data)

#删除指定的会话信息
@app.delete("/api/sessions/{session_id}")
def delete_session(session_id: str) -> ApiResponse:
    logging.info(f"获取指定的会话信息：{session_id}")

    #1.获取会话文件名
    session_file = get_session_file_name(session_id)

    #2.删除会话文件
    if os.path.exists(session_file):
        os.remove(session_file)

    #3.返回数据
    return ApiResponse(code=200, message="删除会话成功",data=None)


#统一异常处理
@app.exception_handler(Exception)
def handle_exception(request: Request, exc: Exception):
    logging.error(f"处理异常，请求路径：{request.url},异常信息：{exc}")
    return JSONResponse(content={"code": 500, "message": "服务器内部错误"})


#启动项目
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000) #accsss_log=False : 关闭访问日志