【案例】Python + Streamlit 搭建私有聊天机器人

"""最小代码：用官方 ollama 库跟本地模型说一句话。

    pip install ollama
    python3 min_ollama.py

和直接发 HTTP 相比，这个库只是替你把 URL、JSON 序列化、响应解析包掉了，
底下走的还是同一个 /api/chat 接口。
"""
import os

import ollama

MODEL = os.environ.get("OLLAMA_MODEL", "qwen2:1.5b")

# 情况一：Ollama 就在本机、端口也没改，直接用模块级函数
response = ollama.chat(
    model=MODEL,
    messages=[{"role": "user", "content": "从前有座山，山里有个庙，续写一下"}],
)
# 返回是个字典，模型说的话在 message.content 里
print(response["message"]["content"])

# 情况二：模型跑在别的机器上（或改过端口），就显式建一个 Client
host = os.environ.get("OLLAMA_HOST_ADDR", "127.0.0.1")
port = os.environ.get("OLLAMA_PORT", "11434")
client = ollama.Client(host="http://%s:%s" % (host, port))

response2 = client.chat(
    model=MODEL,
    messages=[{"role": "user", "content": "用一句话说明什么是私有化部署"}],
    # options 和 HTTP 接口里的完全一致
    options={"temperature": 0.3, "num_predict": 128},
)
print(response2["message"]["content"])

"""后端工具箱：只负责「把消息交给模型、把答案拿回来」。

刻意和界面代码分开：
  - 界面换成 Flask、命令行、企业微信机器人，这个文件都不用动
  - 想换模型、加超时、加日志，也只改这一个文件

单独运行可以脱离界面自测：
    python3 my_utils.py
"""
import os

import ollama

MODEL = os.environ.get("OLLAMA_MODEL", "qwen2:1.5b")

# 上下文窗口有限，历史无限增长会越来越慢，最后被 num_ctx 截断。
# 只把最近 N 条发给模型；N 取多少取决于 num_ctx 和单条消息长度。
MAX_MESSAGES = 50

SYSTEM_PROMPT = "你是黑马智聊机器人，用中文回答，简洁准确，不确定就直说。"


def get_response(prompt):
    """prompt 既可能是一个字符串（单轮），也可能是 messages 列表（多轮）。

    之所以两种都收：界面早期版本传的是一句话，加上记忆之后传的是整段历史。
    与其改调用方，不如在这里判一下类型，接口保持向后兼容。
    """
    if isinstance(prompt, str):
        messages = [{"role": "user", "content": prompt}]
    else:
        # 取最近 MAX_MESSAGES 条，避免上下文无限膨胀
        messages = list(prompt)[-MAX_MESSAGES:]

    # system 始终放在最前面，而且不参与裁剪
    if not messages or messages[0].get("role") != "system":
        messages = [{"role": "system", "content": SYSTEM_PROMPT}] + messages

    response = ollama.chat(
        model=MODEL,
        messages=messages,
        options={"temperature": 0.3},
    )
    return response["message"]["content"]


def get_response_stream(prompt):
    """流式版本：返回一个生成器，界面拿去逐片渲染。"""
    messages = [{"role": "user", "content": prompt}] if isinstance(prompt, str) else list(prompt)[-MAX_MESSAGES:]
    if not messages or messages[0].get("role") != "system":
        messages = [{"role": "system", "content": SYSTEM_PROMPT}] + messages

    for chunk in ollama.chat(model=MODEL, messages=messages, stream=True):
        part = chunk.get("message", {}).get("content", "")
        if part:
            yield part


if __name__ == "__main__":
    # 脱离界面自测：这一步通了，再去调界面；否则分不清是界面问题还是模型问题
    print(get_response("一周有几天？"))

"""界面层：Streamlit 聊天页面，只管显示和收集输入。

    pip install streamlit ollama
    streamlit run my_chat.py

必须理解的机制：用户每敲一次回车，Streamlit 就把这个文件**从第一行整个重跑一遍**。
所以普通变量存不住东西，聊天记录只能放进 st.session_state。
"""
import streamlit as st

from my_utils import get_response

st.set_page_config(page_title="黑马智聊机器人", page_icon="💬")
st.title("黑马智聊机器人")
st.caption("模型跑在本机，数据不出内网")

# ---------- ① 初始化：只在首次进入时执行一次 ----------
# 这个 if 是整个程序的关键。没有它，每次重跑都会把历史清空，
# 机器人就永远只记得你最后说的那一句。
if "messages" not in st.session_state:
    st.session_state["messages"] = [
        {"role": "assistant", "content": "你好，我是黑马智聊机器人，有什么可以帮你的？"},
    ]

# ---------- ② 重放：把历史全部重新画一遍 ----------
# 因为页面是整个重绘的，之前的气泡不会自动留在屏幕上，必须自己循环画出来
for message in st.session_state["messages"]:
    with st.chat_message(message["role"]):
        st.markdown(message["content"])

# ---------- ③ 输入框 ----------
# chat_input 固定在页面底部；没输入时返回 None，脚本执行到这里就结束了
prompt = st.chat_input("请输入你要咨询的问题：")

if prompt:
    # 先把用户这句存进历史，再画出来——顺序反了会导致刷新后丢失
    st.session_state["messages"].append({"role": "user", "content": prompt})
    st.chat_message("user").markdown(prompt)

    # ---------- ④ 调模型 ----------
    with st.chat_message("assistant"):
        with st.spinner("正在思考…"):
            # 把整段历史都传过去，模型才有上下文
            content = get_response(st.session_state["messages"])
        st.markdown(content)

    # ---------- ⑤ 模型的回答也要存进历史 ----------
    # 漏掉这一步，下一轮模型就不记得自己刚才说过什么
    st.session_state["messages"].append({"role": "assistant", "content": content})

# ---------- ⑥ 侧边栏：清空对话 ----------
with st.sidebar:
    if st.button("清空对话"):
        st.session_state["messages"] = [
            {"role": "assistant", "content": "已清空，我们重新开始。"},
        ]
        # 手动触发一次重跑，页面立刻刷新
        st.rerun()

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""cli_chat.py —— 同一个工具层，换一个界面

用途：证明分层不是嘴上说说。这个文件把 Streamlit 界面换成纯命令行，
     而 my_utils.py 一个字都没改，直接 import 过来用。

用法：
    python3 cli_chat.py
    OLLAMA_MODEL=qwen2:7b python3 cli_chat.py

为什么值得写这一份：
  * 服务器上没有浏览器时，它就是最快的验收手段；
  * 页面出问题时，用它对照一下就知道是界面的锅还是模型的锅；
  * 接企业微信机器人、定时任务、批处理脚本时，起点也是这一份。

对话中可用的命令：
    /clear  清空历史，重新开始
    /save   把当前对话存成 JSON 文件
    /stat   看当前历史条数与大致字数
    /bye    退出
"""

import json
import os
import sys
import time

from my_utils import MAX_MESSAGES, SYSTEM_PROMPT, get_response_stream

SAVE_DIR = os.environ.get("CHAT_SAVE_DIR", ".")


def new_history():
    """每次开新对话都从这里开始：只有一条 system。"""
    return [{"role": "system", "content": SYSTEM_PROMPT}]


def show_stat(messages):
    """历史有多长，心里要有数——越长越慢是必然的。"""
    chars = sum(len(m["content"]) for m in messages)
    turns = len([m for m in messages if m["role"] == "user"])
    print("  历史消息 %d 条（含 system），完成 %d 轮问答，约 %d 字"
          % (len(messages), turns, chars))
    print("  工具层保留上限 MAX_MESSAGES = %d 条" % MAX_MESSAGES)
    if len(messages) > MAX_MESSAGES:
        print("  已超过上限，最早的消息不会再发给模型（system 除外）")


def save_history(messages):
    """存成 JSON，方便事后复盘模型到底看到了什么。"""
    name = time.strftime("chat-%Y%m%d-%H%M%S.json")
    path = os.path.join(SAVE_DIR, name)
    with open(path, "w", encoding="utf-8") as fp:
        json.dump(messages, fp, ensure_ascii=False, indent=2)
    print("  已保存到 %s" % path)


def ask(messages, question):
    """发一轮问答：流式打印，返回完整答案。

    这里不用 print(end="") 之后再统一 flush，是因为不 flush 的话
    输出会被缓冲住，流式效果白做了——看起来还是等到最后一次性出现。
    """
    messages.append({"role": "user", "content": question})
    print("助手：", end="")
    sys.stdout.flush()

    started = time.time()
    pieces = []
    try:
        for part in get_response_stream(messages):
            pieces.append(part)
            print(part, end="")
            sys.stdout.flush()
    except KeyboardInterrupt:
        print("\n  （已中断本轮生成）")
        messages.pop()  # 这一轮不完整，把问题也撤回，避免污染历史
        return None
    except Exception as exc:
        print("\n  调用模型失败：%s" % exc)
        print("  先确认服务在跑：ollama ps；再确认模型名写全了带冒号版本。")
        messages.pop()
        return None

    answer = "".join(pieces)
    elapsed = time.time() - started
    print("\n  （耗时 %.1fs，%d 字）" % (elapsed, len(answer)))

    # 关键一步：模型的回答必须回填进历史，否则下一轮它不记得自己说过什么
    messages.append({"role": "assistant", "content": answer})
    return answer


def main():
    print("=" * 56)
    print("命令行版聊天机器人（界面换了，工具层没动）")
    print("模型：%s" % os.environ.get("OLLAMA_MODEL", "qwen2:1.5b"))
    print("命令：/clear 清空  /save 保存  /stat 看历史  /bye 退出")
    print("=" * 56)

    messages = new_history()

    while True:
        try:
            question = input("\n你：").strip()
        except (EOFError, KeyboardInterrupt):
            print("\n再见。")
            return 0

        if not question:
            continue
        if question == "/bye":
            print("再见。")
            return 0
        if question == "/clear":
            messages = new_history()
            print("  已清空，我们重新开始。")
            continue
        if question == "/stat":
            show_stat(messages)
            continue
        if question == "/save":
            save_history(messages)
            continue
        if question.startswith("/"):
            print("  未知命令；可用：/clear /save /stat /bye")
            continue

        ask(messages, question)


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""chat_smoke_test.py —— 交付之前的自动化验收

用途：把「肉眼点一遍页面」这件事写成脚本。重点验的不是「有没有回答」，
     而是这个案例真正的验收点——**多轮记忆到底通没通**。

用法：
    python3 chat_smoke_test.py
    OLLAMA_MODEL=qwen2:7b python3 chat_smoke_test.py

它验五件事：
    1. 工具层能不能调通模型（单轮）
    2. 传整段历史时，模型记不记得前面说过的名字
    3. 只传当前这一句时，模型确实不记得（反向验证）
    4. 流式接口能不能分多片返回
    5. 历史裁剪之后，system 人设还在不在

退出码 0 = 全通过，1 = 有失败项，可以接进发版流程。
"""

import sys
import time

import my_utils

PASS = "[ PASS ]"
FAIL = "[ FAIL ]"

results = []


def record(name, ok, detail=""):
    results.append((name, ok))
    line = "%s %s" % (PASS if ok else FAIL, name)
    if detail:
        line += " —— " + detail
    print(line)
    return ok


def t1_single_turn():
    """最基本的一步：工具层能不能拿到回答。"""
    try:
        answer = my_utils.get_response("一周有几天？只回答数字。")
    except Exception as exc:
        return record("1. 工具层单轮调用", False,
                      "%s（先确认 ollama ps 里有这个模型）" % exc)
    if not answer or not answer.strip():
        return record("1. 工具层单轮调用", False, "返回了空字符串")
    return record("1. 工具层单轮调用", True, "回答 %r" % answer.strip()[:20])


def t2_memory_with_history():
    """传整段历史，模型应该记得名字。这是本案例的核心验收点。"""
    messages = [
        {"role": "user", "content": "记住，我叫小明。"},
        {"role": "assistant", "content": "好的，小明。"},
        {"role": "user", "content": "我叫什么名字？只回答名字。"},
    ]
    try:
        answer = my_utils.get_response(messages)
    except Exception as exc:
        return record("2. 传整段历史时有记忆", False, str(exc))
    ok = "小明" in answer
    detail = "回答 %r" % answer.strip()[:24]
    if not ok:
        detail += "；检查界面层是否传了整个 messages 而不是单句"
    return record("2. 传整段历史时有记忆", ok, detail)


def t3_no_memory_without_history():
    """反向验证：只传当前这一句，模型不应该知道名字。

    这一项失败并不总是代码问题（小模型可能瞎猜一个名字），
    所以失败时给的是提示而不是断言，避免误导。
    """
    try:
        answer = my_utils.get_response("我叫什么名字？")
    except Exception as exc:
        return record("3. 不传历史时确实没有记忆", False, str(exc))
    ok = "小明" not in answer
    detail = "回答 %r" % answer.strip()[:24]
    if not ok:
        detail += "；模型凭空说出了名字，属于异常，检查是否误用了共享历史"
    return record("3. 不传历史时确实没有记忆", ok, detail)


def t4_stream_chunks():
    """流式应当分多片返回；只收到一片说明中间有人做了缓冲。"""
    started = time.time()
    chunks = 0
    text = ""
    try:
        for part in my_utils.get_response_stream("从 1 数到 10，只输出数字。"):
            chunks += 1
            text += part
    except Exception as exc:
        return record("4. 流式接口分片返回", False, str(exc))
    if chunks <= 1:
        return record("4. 流式接口分片返回", False,
                      "只收到 %d 片，检查网关 proxy_buffering" % chunks)
    return record("4. 流式接口分片返回", True,
                  "%d 片，耗时 %.1fs" % (chunks, time.time() - started))


def t5_system_survives_trim():
    """构造一段超长历史，确认裁剪之后 system 仍然在最前面。

    这里不发请求，只检查工具层的拼装逻辑——真正容易错的就是这一步。
    """
    long_history = []
    for i in range(my_utils.MAX_MESSAGES + 20):
        role = "user" if i % 2 == 0 else "assistant"
        long_history.append({"role": role, "content": "第 %d 条消息" % i})

    # 复刻 get_response 里的拼装逻辑，验证裁剪顺序
    trimmed = list(long_history)[-my_utils.MAX_MESSAGES:]
    if not trimmed or trimmed[0].get("role") != "system":
        trimmed = [{"role": "system", "content": my_utils.SYSTEM_PROMPT}] + trimmed

    ok = trimmed[0]["role"] == "system" and len(trimmed) <= my_utils.MAX_MESSAGES + 1
    detail = "裁剪后 %d 条，首条 role=%s" % (len(trimmed), trimmed[0]["role"])
    if not ok:
        detail += "；system 被裁掉会导致聊久了人设消失"
    return record("5. 裁剪后 system 仍在最前", ok, detail)


def main():
    print("模型：%s" % my_utils.MODEL)
    print("历史上限 MAX_MESSAGES：%d" % my_utils.MAX_MESSAGES)
    print("-" * 56)

    if not t1_single_turn():
        print("-" * 56)
        print("最基本的调用都没通，后面的用例没有意义。")
        print("排查顺序：ollama ps → curl /api/version → python3 my_utils.py")
        return 1

    t2_memory_with_history()
    t3_no_memory_without_history()
    t4_stream_chunks()
    t5_system_survives_trim()

    print("-" * 56)
    failed = [name for name, ok in results if not ok]
    print("通过 %d/%d" % (len(results) - len(failed), len(results)))
    if failed:
        for name in failed:
            print("  未通过：%s" % name)
        return 1
    print("全部通过，可以交付。")
    return 0


if __name__ == "__main__":
    sys.exit(main())

"""骨架模板：私有聊天机器人，复制改 TODO 就能变成你自己的业务助手。

    pip install streamlit ollama
    streamlit run chatbot_skeleton.py

已经替你处理好的四件事：人设可配、历史可裁剪、流式输出、异常兜底。
"""
import os

import ollama
import streamlit as st

# ---------- 配置区：改这里就够了 ----------
# TODO: 换成你要用的模型全名（ollama list 里查得到的那个）
MODEL = os.environ.get("OLLAMA_MODEL", "qwen2:1.5b")

# TODO: 模型服务地址；跑在别的机器上就改这里，或者设环境变量
HOST = os.environ.get("OLLAMA_BASE", "http://127.0.0.1:11434")

# TODO: 换成你的业务人设，这一段直接决定机器人像不像样
SYSTEM_PROMPT = """
你是某公司的内部助手。
只依据用户提供的信息回答，不确定就说不确定，不要编造。
回答用中文，先结论后细节。
"""

# TODO: 页面标题与开场白
PAGE_TITLE = "内部智能助手"
WELCOME = "你好，我是内部助手，有什么可以帮你的？"

# TODO: 保留多少条历史。调大更能记事，但每轮请求都变长、变慢
MAX_MESSAGES = 40

client = ollama.Client(host=HOST)


def call_model(messages):
    """流式调用，返回一个生成器供 st.write_stream 渲染。"""
    payload = [{"role": "system", "content": SYSTEM_PROMPT}] + messages[-MAX_MESSAGES:]
    for chunk in client.chat(model=MODEL, messages=payload, stream=True,
                             options={"temperature": 0.3}):
        part = chunk.get("message", {}).get("content", "")
        if part:
            yield part


# ---------- 页面 ----------
st.set_page_config(page_title=PAGE_TITLE, page_icon="💬")
st.title(PAGE_TITLE)

if "messages" not in st.session_state:
    st.session_state["messages"] = [{"role": "assistant", "content": WELCOME}]

for m in st.session_state["messages"]:
    with st.chat_message(m["role"]):
        st.markdown(m["content"])

prompt = st.chat_input("请输入问题：")
if prompt:
    st.session_state["messages"].append({"role": "user", "content": prompt})
    st.chat_message("user").markdown(prompt)

    with st.chat_message("assistant"):
        try:
            # write_stream 会边收边渲染，并把完整文本作为返回值给你
            answer = st.write_stream(call_model(st.session_state["messages"]))
        except Exception as exc:  # 模型没起、端口不通、超时都会走到这里
            answer = "调用模型失败：%s" % exc
            st.error(answer)

    st.session_state["messages"].append({"role": "assistant", "content": answer})

with st.sidebar:
    st.write("当前模型：`%s`" % MODEL)
    st.write("服务地址：`%s`" % HOST)
    if st.button("清空对话"):
        st.session_state["messages"] = [{"role": "assistant", "content": WELCOME}]
        st.rerun()

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""chat_logger.py —— 给工具层加一层日志与统计

用途：回答「到底有没有人在用」「慢在哪」「失败率多少」这三个问题。
     日志加在工具层，因为它是所有界面共用的唯一入口——
     加在界面层的话，换一个界面就要重加一次。

用法（在 my_utils 里包一层）：
    from chat_logger import log_call

    @log_call
    def get_response(prompt):
        ...

单独查看统计：
    python3 chat_logger.py            # 打印汇总
    python3 chat_logger.py --tail 20  # 看最近 20 条

日志落在 CHAT_LOG_PATH 指定的文件，默认 ./chat_calls.log，
每行一条 JSON，方便用 grep / jq 直接查，也方便后续接进日志系统。

刻意不记录用户问题的完整原文，只记长度和哈希——
内部助手的对话里经常有敏感信息，日志是最容易被忽略的泄漏口。
"""

import functools
import hashlib
import json
import os
import sys
import time

LOG_PATH = os.environ.get("CHAT_LOG_PATH", "chat_calls.log")


def _digest(text):
    """只留短哈希，用于判断「是不是同一个问题反复问」，但还原不出原文。"""
    return hashlib.sha256(text.encode("utf-8")).hexdigest()[:12]


def _write(record):
    """一行一条 JSON；失败不能影响主流程，所以整段包住。"""
    try:
        with open(LOG_PATH, "a", encoding="utf-8") as fp:
            fp.write(json.dumps(record, ensure_ascii=False) + "\n")
    except Exception:
        # 日志写不进去不该拖垮聊天功能，静默放过
        pass


def _summarize_prompt(prompt):
    """兼容工具层的两种入参：字符串或 messages 列表。"""
    if isinstance(prompt, str):
        return {"turns": 1, "chars": len(prompt), "hash": _digest(prompt)}
    messages = list(prompt)
    chars = sum(len(m.get("content", "")) for m in messages)
    last_user = ""
    for m in reversed(messages):
        if m.get("role") == "user":
            last_user = m.get("content", "")
            break
    return {
        "turns": len([m for m in messages if m.get("role") == "user"]),
        "chars": chars,
        "hash": _digest(last_user) if last_user else "",
    }


def log_call(func):
    """装饰器：记录每次模型调用的耗时、规模与成败。"""

    @functools.wraps(func)
    def wrapper(prompt, *args, **kwargs):
        info = _summarize_prompt(prompt)
        started = time.time()
        try:
            answer = func(prompt, *args, **kwargs)
        except Exception as exc:
            _write({
                "ts": time.strftime("%Y-%m-%d %H:%M:%S"),
                "ok": False,
                "elapsed": round(time.time() - started, 2),
                "in_turns": info["turns"],
                "in_chars": info["chars"],
                "q": info["hash"],
                "err": exc.__class__.__name__,
            })
            raise
        elapsed = time.time() - started
        out_chars = len(answer) if isinstance(answer, str) else 0
        _write({
            "ts": time.strftime("%Y-%m-%d %H:%M:%S"),
            "ok": True,
            "elapsed": round(elapsed, 2),
            "in_turns": info["turns"],
            "in_chars": info["chars"],
            "out_chars": out_chars,
            "q": info["hash"],
        })
        return answer

    return wrapper


def load_records(path=None):
    path = path or LOG_PATH
    if not os.path.exists(path):
        return []
    records = []
    with open(path, "r", encoding="utf-8") as fp:
        for line in fp:
            line = line.strip()
            if not line:
                continue
            try:
                records.append(json.loads(line))
            except json.JSONDecodeError:
                continue  # 半行写坏了就跳过，不因为一行毁掉整份统计
    return records


def summarize(records):
    if not records:
        print("还没有任何调用记录：%s" % LOG_PATH)
        return 0

    total = len(records)
    failed = [r for r in records if not r.get("ok")]
    ok_records = [r for r in records if r.get("ok")]
    elapsed_list = sorted(r.get("elapsed", 0) for r in ok_records)

    print("日志文件      : %s" % LOG_PATH)
    print("总调用        : %d 次" % total)
    print("失败          : %d 次（%.1f%%）"
          % (len(failed), 100.0 * len(failed) / total))

    if elapsed_list:
        avg = sum(elapsed_list) / len(elapsed_list)
        p50 = elapsed_list[len(elapsed_list) // 2]
        p95 = elapsed_list[min(len(elapsed_list) - 1,
                               int(len(elapsed_list) * 0.95))]
        print("耗时 平均/中位/P95 : %.1fs / %.1fs / %.1fs" % (avg, p50, p95))
        print("最慢一次      : %.1fs" % elapsed_list[-1])

    if ok_records:
        avg_turns = sum(r.get("in_turns", 0) for r in ok_records) / len(ok_records)
        avg_in = sum(r.get("in_chars", 0) for r in ok_records) / len(ok_records)
        print("平均轮数      : %.1f 轮" % avg_turns)
        print("平均送入字数  : %.0f 字" % avg_in)
        if avg_in > 4000:
            print("  提示：送入字数偏大，检查历史裁剪是否生效")

    if failed:
        kinds = {}
        for r in failed:
            kinds[r.get("err", "?")] = kinds.get(r.get("err", "?"), 0) + 1
        print("失败类型      : %s"
              % ", ".join("%s×%d" % (k, v) for k, v in kinds.items()))

    # 反复问同一个问题，通常说明答案没让人满意
    counter = {}
    for r in records:
        key = r.get("q") or ""
        if key:
            counter[key] = counter.get(key, 0) + 1
    repeated = [(k, v) for k, v in counter.items() if v >= 3]
    if repeated:
        repeated.sort(key=lambda x: x[1], reverse=True)
        print("高频重复提问  : %d 个（最高 %d 次）——答案可能没解决问题"
              % (len(repeated), repeated[0][1]))
    return 0


def main():
    args = sys.argv[1:]
    records = load_records()
    if "--tail" in args:
        idx = args.index("--tail")
        n = int(args[idx + 1]) if len(args) > idx + 1 else 10
        for r in records[-n:]:
            print(json.dumps(r, ensure_ascii=False))
        return 0
    return summarize(records)


if __name__ == "__main__":
    sys.exit(main())