Python 助手（后端）

一、引言

1.1 系统目的

本系统为 Python 开发助手，通过与后端的交互，利用 AnythingLLM 的知识库和 /v1/workspace/{slug}/stream-chat API 提供实时的、流式的代码生成、纠错等功能。

1.2 架构概览 (VS Code 集成)

  • 后端代理 (Flask): (同前) 负责安全调用 AnythingLLM 并转发 SSE 流。
  • AnythingLLM 服务: (同前) 提供 AI 能力。

1.3 技术栈

  • 后端: Python 3, Flask, Requests, python-dotenv(用于管理环境变量)
  • 通信协议: HTTP/S, Server-Sent Events (SSE，用于实现流式响应)

二、开发环境设置

2.1 环境要求

  • Python 3.12 和 pip

  • 一个正在运行且网络可达的 AnythingLLM 实例

2.2 项目设置

  • 创建后端项目: 创建 Flask 后端，并配置 (如app.py, .env)。

2.3 安装依赖

  • 后端依赖:
pip install Flask requests python-dotenv

三、配置说明

3.1 后端配置 (.env)

在后端项目根目录下创建 .env 文件，用于存储敏感信息和配置。此文件不应提交到版本控制系统。

# .env

# ---- AnythingLLM ----
# AnythingLLM 实例地址
ANYTHINGLLM_BASE_URL = "http://XXX.XXX.XXX.XXX:XXXX"
# API 密钥
ANYTHINGLLM_API_KEY = "XXXXXXX-XXXXXXX-XXXXXXX-XXXXXXX"
# 要交互的工作区标识符 slug
ANYTHINGLLM_WORKSPACE_SLUG = "test1"

# ---- Flask 配置信息 ----
FLASK_APP=app.py
FLASK_DEBUG=True    # 开发阶段设为True，使用阶段设为False

3.2 获取配置值

  • ANYTHINGLLM_BASE_URL: AnythingLLM 实例的可访问地址。
  • ANYTHINGLLM_API_KEY: 在 AnythingLLM 的设置 (Settings) -> API Keys 中创建和获取。
  • ANYTHINGLLM_WORKSPACE_SLUG: 在 AnythingLLM 中进入对应工作区。

四、后端 (基于 Flask 框架) 详解

4.1 项目结构

  • 以下是基本结构：

在这里插入图片描述

4.2 依赖

  • Flask: Web 框架，用于创建 API 接口和处理 HTTP 请求/响应。
  • Requests: 强大的 HTTP 客户端库，用于向 AnythingLLM 发送请求。
  • python-dotenv: 用于从 .env 文件加载环境变量，方便配置管理。

4.3 API 接口: POST /stream-ask

向前端提供的一个的核心接口，用于处理流式聊天请求（用于代码片段的生成和纠错）。

4.3.1 请求格式

  • 方法: POST

  • 路径: /stream-ask

  • Headers:

    • Content-Type: application/json

  • Body:

{
  "question": "用户的提问内容"
}
4.3.2 成功响应 (SSE 流)

  • Status Code: 200 OK

  • Headers:

  • Content-Type: text/event-stream

  • Cache-Control: no-cache

  • Connection: keep-alive

  • Body (SSE Stream): 一个遵循 Server-Sent Events 格式的文本流。包含一系列事件，每条事件以 \n\n 分隔。常见的事件格式：

    • 数据块事件 (默认 message 或自定义 chunk 事件):
data: {"type": "chunk", "content": "这是返回的文本片段"}\n\n

或 (如果后端仅转发 AnythingLLM 的纯文本块)

data: "这是返回的文本片段"\n\n

注意: data: 行后面的 JSON 结构完全取决于 AnythingLLM /stream-chat 接口的实际输出以及后端 event_stream 函数的处理方式。开发者必须根据实际情况调整前端的解析逻辑。

  • 错误事件 (自定义 error 事件):
event: error
data: {"message": "处理过程中发生错误，例如无法连接知识库。"}\n\n
  • 结束事件 (自定义 final_end 事件):
event: final_end
data: {} \n\n

(data 内容可以为空或包含最终信息，如来源总结)

4.3.3 错误响应 (非流式)

在请求阶段发生错误（如请求格式无效、配置错误）时，接口会返回标准的 JSON 错误响应，而不是 SSE 流。

  • Status Code: 400 Bad Request, 500 Internal Server Error, etc.

  • Headers:

  • Content-Type: application/json

  • Body (JSON):

{
  "error": "具体的错误描述信息"
}

4.4 代码详解 (app.py)

4.4.1 初始化与配置加载
import os
import requests
import json
from flask import Flask, request, jsonify, Response, stream_with_context
from dotenv import load_dotenv

# 从 .env 文件加载环境变量
load_dotenv()

# 初始化 Flask 应用
app = Flask(__name__)

# --- 从环境变量（.env文件）加载配置 ---
ANYTHINGLLM_BASE_URL = os.getenv("ANYTHINGLLM_BASE_URL")
ANYTHINGLLM_API_KEY = os.getenv("ANYTHINGLLM_API_KEY")
ANYTHINGLLM_WORKSPACE_SLUG = os.getenv("ANYTHINGLLM_WORKSPACE_SLUG")

# --- 输入验证 ---
# 检查必要的环境变量是否都已设置
if not all([ANYTHINGLLM_BASE_URL, ANYTHINGLLM_API_KEY, ANYTHINGLLM_WORKSPACE_SLUG]):
    app.logger.error("错误：缺少 AnythingLLM 环境变量。")
  • 导入所需库，加载环境变量，初始化 Flask 应用和日志记录。
4.4.2 请求处理与验证
@app.route('/stream-ask', methods=['POST'])
def handle_stream_ask():
    """
    接收问题，调用 AnythingLLM 的 stream-chat 接口，并将 SSE 流转发给前端。
    """
    app.logger.info("收到 /stream-ask 接口的请求")

    # 1. 获取并验证前端请求数据
    try:
        data = request.form
        if not data or 'question' not in data:
            app.logger.warning("错误的请求：JSON 负载中缺少 'question'")
            # 对于流式接口，如果请求本身就有问题，直接返回错误 JSON
            return jsonify({"error": "请求体中缺少 'question' 字段"}), 400
        question = data['question']
        app.logger.info(f"收到的问题 (流式): {question}")
    except Exception as e:
        app.logger.error(f"解析请求 JSON 时出错: {e}")
        return jsonify({"error": "请求体中的 JSON 格式无效"}), 400

    # 2. 检查配置
    if not all([ANYTHINGLLM_BASE_URL, ANYTHINGLLM_API_KEY, ANYTHINGLLM_WORKSPACE_SLUG]):
         app.logger.error("无法处理请求：AnythingLLM 配置缺失。")
         return jsonify({"error": "内部服务器配置错误。"}), 500
  • 定义 /stream-ask 路由，限定 POST 方法。使用 request.json 获取并验证请求体，处理错误并返回 400/500 响应给前端。
4.4.3 调用 AnythingLLM 流式接口
	# 3. 准备调用 AnythingLLM 的 stream-chat 接口
    anythingllm_stream_endpoint = f"{ANYTHINGLLM_BASE_URL}/api/v1/workspace/{ANYTHINGLLM_WORKSPACE_SLUG}/stream-chat"
    headers = {
        "Authorization": f"Bearer {ANYTHINGLLM_API_KEY}",
        "Content-Type": "application/json",
        "Accept": "text/event-stream" # 期望得到 SSE 流
    }
    payload = {
        "message": question,
        "mode": "chat"
    }
  • 构建目标 URL、 headers 和请求 payload。
  • 在 headers 中设置 Accept: text/event-stream 是向 AnythingLLM 表示期望 SSE 响应。
  • 在 requests.post 调用中设置 stream=True ，阻止 requests 立即下载整个响应体。
  • 使用 with 语句管理 requests 的响应对象，确保资源释放。
  • 调用 response.raise_for_status() 检查初始 HTTP 连接是否成功。
4.4.4 SSE 处理与转发生成器 (event_stream)
# 4. 定义一个生成器函数来处理和转发 SSE 流
    def event_stream():
        try:
            with requests.post(
                anythingllm_stream_endpoint,
                headers=headers,
                json=payload,
                stream=True, # 启用流式请求
                timeout=180 # 流式连接可能需要更长的超时时间
            ) as response:
                app.logger.info(f"已连接到 AnythingLLM stream-chat: {response.status_code}")
                # 检查初始连接是否成功 (例如，认证失败会在这里报 4xx)
                response.raise_for_status()

                # 迭代处理从 AnythingLLM 收到的每一行数据
                # decode_unicode=True 自动将字节解码为字符串
                for line in response.iter_lines(decode_unicode=True):
                    if line: # 过滤掉空行
                        app.logger.debug(f"收到 SSE 行: {line}") # 调试时可以取消注释
                        if line.startswith("data:"):
                            # 提取 data: 后面的 JSON 字符串
                            data_content = line[len("data:"):].strip()
                            is_complete_signal = False
                            try:
                                parsed_data = json.loads(data_content)

                                # 如果不是结束信号，直接转发数据给前端
                                if not is_complete_signal:
                                     # 格式化为 SSE 事件并发送给前端
                                     repaired_data = data_content.encode("latin-1")
                                     corrected_data = repaired_data.decode("utf-8")
                                     # yield (f"data: {data_content}\n\n").encode("utf-8")
                                     yield (f"data: {corrected_data}\n\n").encode("utf-8")

                            except json.JSONDecodeError:
                                app.logger.warning(f"无法解析收到的 SSE data: {data_content}")
                            except Exception as e:
                                app.logger.error(f"处理 SSE 数据时发生意外错误: {e}")
                                # 可以发送一个错误事件给前端
                                yield f"event: error\ndata: {{\"message\": \"Error processing stream data\"}}\n\n"
                                break # 出错时可能需要中断流

                        elif line.startswith("event:"):
                             # 可以选择性地处理或直接转发其他 SSE 事件
                             yield f"{line}\n" # 如果需要转发 event 类型

                app.logger.info("AnythingLLM 流结束 (iter_lines 完成)")

        # 处理请求 AnythingLLM 期间发生的错误 (连接错误、HTTP错误等)
        except requests.exceptions.Timeout:
            app.logger.error("错误：连接 AnythingLLM stream-chat 超时。")
            yield f"event: error\ndata: {{\"message\": \"Connection to knowledge base timed out.\"}}\n\n"
        except requests.exceptions.HTTPError as e:
            error_message = f"错误：AnythingLLM API 返回状态 {e.response.status_code}。"
            # 尝试获取错误详情
            try:
                error_details = e.response.json()
                error_message += f" Details: {error_details.get('error', e.response.text)}"
            except: # noqa
                 error_message += f" Response body: {e.response.text}"
            app.logger.error(error_message)
            yield f"event: error\ndata: {{\"message\": \"Failed to connect to the knowledge base ({e.response.status_code}).\"}}\n\n"
        except requests.exceptions.RequestException as e:
            app.logger.error(f"错误：访问 AnythingLLM 的网络请求失败: {e}")
            yield f"event: error\ndata: {{\"message\": \"Could not connect to the knowledge base.\"}}\n\n"
        except Exception as e:
            app.logger.error(f"处理流时发生意外错误: {e}", exc_info=True)
            yield f"event: error\ndata: {{\"message\": \"An internal server error occurred while processing the stream.\"}}\n\n"
        finally:
            app.logger.info("流处理生成器结束。")
            # 可以在这里 yield 一个最终的自定义事件，确保前端知道流已关闭
            yield ("event: final_end\ndata: {}\n\n").encode("utf-8")

说明:

  • 生成器函数: 使用 def event_stream(): 定义了一个生成器。

  • response.iter_lines(decode_unicode=True): 高效地逐行读取来自 AnythingLLM 的响应流，并自动解码为 UTF-8 字符串。

  • SSE 解析: 循环内部代码检查每行是否以 data: 或 event: 开头，提取相应内容。

  • 重要: 此示例直接转发 data: 后面的内容。实际应用中，你必须根据 AnythingLLM 返回的具体 JSON 结构来决定是否需要先 json.loads() 解析 data_content，然后可能根据 type 字段或其他逻辑决定转发什么内容，或者处理如 sources 等特殊数据块。

  • yield f"data: {data_content}\n\n": 将处理（或直接转发）后的数据块格式化为标准的 SSE data 事件，并通过 yield 发送出去。Flask 会捕获这些 yield 的值并将其流式传输给客户端。\n\n 是强制性的消息分隔符。

  • yield "event: final_end\ndata: {}\n\n": 在流正常结束时（iter_lines 完成），发送一个自定义的 final_end 事件，告知前端流已结束。

  • 错误事件 (yield f"event: error..."): 在 except 块中捕获到错误时，发送一个自定义的 error 事件，并将错误信息包装在 data: 负载中（推荐使用 JSON 格式）。

4.4.5 错误处理机制
  • 分层处理:

    • 请求验证层: 处理无效的客户端输入。

    • 连接层: 处理与 AnythingLLM 建立连接时的错误，在生成器开始时 yield 错误事件。

    • 流处理层: 处理在读取和解析 AnythingLLM 流过程中的错误，在生成器内部 yield 错误事件。

    • 通过 SSE 传递错误: 对于流式接口，开始流式传输之后，就不能再返回普通的 HTTP 错误码。因此，通过 yield 自定义的 event: error 并携带错误信息 JSON，将运行时错误返回给前端。

4.4.6 返回流式响应
    # 5. 返回一个流式响应给前端
    # 设置 mimetype='text/event-stream'
    # 使用 stream_with_context 确保在请求上下文之外也能访问 request 等对象
    return Response(stream_with_context(event_stream()), mimetype='text/event-stream')
  • 创建 Flask 的 Response 对象。
  • 第一个参数是调用我们的 event_stream() 生成器函数。stream_with_context 确保生成器在执行时能访问请求上下文。
  • mimetype='text/event-stream' 是必须设置的，使得前端将此响应作为 SSE 流处理。
Logo
DeepSeek技术社区

欢迎加入DeepSeek 技术社区。在这里，你可以找到志同道合的朋友，共同探索AI技术的奥秘。

更多推荐

cover

【华为开发者空间 x DeepSeek】基于华为开发者空间云主机DeepSeek助力电商企业AI海报文案驱动的最佳实践落地

avatar DeepSeek技术社区
cover

【实战利器】大模型开源项目全盘点！超详细，一定记得收藏！

avatar DeepSeek技术社区
cover

【GitHub精华】《大模型项目实战》项目笔记大公开！实战经验全收录！

avatar DeepSeek技术社区