Building a Watsonx.ai Chatbot RAG Server with MCP

使用 MCP 构建 Watsonx.ai 聊天机器人 RAG 服务

在本教程中，你将学习如何使用 IBM Watsonx.ai、用于向量索引的 ChromaDB，以及 Model Context Protocol（MCP）Python SDK，构建功能齐全的检索增强生成（RAG）服务，并通过 MCP 将其公开。完成后，你将拥有一个可处理 PDF 文档的独立服务，以及一个可针对这些文档进行查询的客户端，借助基于特定数据的大模型发挥强大能力。最后，我们将展示如何直接集成到 Claude Desktop 中。

Model Context Protocol（MCP） 对应用程序与大模型之间的接口进行了标准化。借助 MCP，你可以分离提供上下文、执行代码和管理用户交互等职责。MCP Python SDK 实现了完整的 MCP 规范，让你能够：

• **公开资源：**向大模型提供数据（类似 GET 端点）。
• **定义工具：**提供可执行操作或计算的功能（类似 POST 端点）。
• **创建提示词：**提供可重复使用的模板化交互。 我们的服务将实现一个客户端可以调用的 RAG“工具”。

前提条件

• 已安装 Python 3.8+（代码使用 typing.Union，与较旧版本兼容，但通常建议使用较新的 Python 版本）。
• 拥有 IBM Cloud 帐户，并已配置 Watsonx.ai 服务。你需要 API Key 和 Project ID。
• 用于安装 Python 软件包的 pip。

项目结构

watsonx-rag-mcp-server/
├── .env
├── requirements.txt
├── documents/         # Your PDF files live here
├── client.py
└── server.py

步骤 1：设置项目和环境

首先，为项目创建一个专用目录并进入该目录。

强烈建议使用虚拟环境：

python -m venv .venv
source .venv/bin/activate # On Windows use `.venv\Scripts\activate`

接下来，在项目根目录中创建一个 .env 文件，用于安全地存储 Watsonx.ai 凭据和配置。添加以下内容，并将占位值替换为你的实际凭据：

# .env
WATSONX_APIKEY=your_watsonx_api_key_here
WATSONX_URL=your_watsonx_instance_url_here
PROJECT_ID=your_watsonx_project_id_here

# Optional: Specify different folder/DB paths if needed
# DOCS_FOLDER=./my_manuals
# CHROMA_PERSIST_DIR=./vector_db_storage
# Optional: Specify a different Watsonx.ai model
# MODEL_ID=meta-llama/llama-3-70b-instruct

步骤 2：设置文档

RAG 服务需要文档来检索信息。创建一个目录（默认名称为 documents，与 .env 设置或 server.py 中的默认值一致），并将 PDF 文件放入其中。

mkdir documents
# --- Add your PDF files (e.g., drone manuals) into this 'documents' folder ---
# Example: cp /path/to/your/manual.pdf documents/

步骤 3：实现服务（server.py）

现在，我们将深入了解检索增强生成（RAG）聊天机器人的核心部分。这个 Python 脚本 server.py 负责处理 PDF 文档、管理向量数据库、与 IBM 的 Watsonx.ai 大语言模型（LLM）交互，并通过 MCP（Meta Compute Protocol）服务公开聊天功能。

我们的目标是创建一个系统，使其能够回答特定基于所提供 PDF 文档内容的问题（例如本例中的无人机手册）。下面我们来逐步了解 server.py 如何实现这一目标。

1. 做好准备：导入和配置

和任何优秀项目一样，我们先导入必要的工具并设置配置。

# server.py
import os
import sys
import logging
from pathlib import Path
import re
import textwrap
from dotenv import load_dotenv
from typing import Union # <--- IMPORT Union for older Python versions

# MCP Imports
from mcp.server.fastmcp import FastMCP

# IBM Watsonx.ai SDK Imports
from ibm_watsonx_ai import APIClient, Credentials
from ibm_watsonx_ai.foundation_models import ModelInference
from ibm_watsonx_ai.metanames import GenTextParamsMetaNames as GenParams

# Vector DB Imports
import chromadb
from pypdf import PdfReader

# --- Configuration and Setup ---

# Load .env variables
load_dotenv()

# Watsonx Credentials and Config
API_KEY      = os.getenv("WATSONX_APIKEY")
URL          = os.getenv("WATSONX_URL")
PROJECT_ID   = os.getenv("PROJECT_ID")
MODEL_ID     = os.getenv("MODEL_ID", "ibm/granite-13b-instruct-v2") # Default model

# Vector DB Config
DOCS_FOLDER = Path(os.getenv("DOCS_FOLDER", "documents")) # Where PDFs are stored
CHROMA_PERSIST_DIR = Path(os.getenv("CHROMA_PERSIST_DIR", "./chroma_db_data")) # Where DB data is saved
COLLECTION_NAME = "pdf_documents_chunked_watsonx" # DB Collection name
NUM_RESULTS_RAG = 3   # How many chunks to retrieve for context
CHUNK_SIZE = 700      # Target size for text chunks (in characters)
CHUNK_OVERLAP = 150   # Overlap between consecutive chunks

# --- Global Variables (Using typing.Union for compatibility) ---
# Holds the initialized ChromaDB collection
pdf_collection: Union[chromadb.Collection, None] = None # <--- FIX: Use Union
# Holds the persistent ChromaDB client instance
chroma_client: Union[chromadb.PersistentClient, None] = None # <--- FIX: Use Union

# --- Logging Configuration ---
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s"
)

# --- Environment Variable Validation ---
for name, val in [
    ("WATSONX_APIKEY", API_KEY),
    ("WATSONX_URL", URL),
    ("PROJECT_ID", PROJECT_ID)
]:
    if not val:
        raise RuntimeError(f"{name} is not set. Please add it to your .env file.")

说明：

• **导入：**我们引入标准库（os、sys、logging、pathlib、re、textwrap）、用于从 .env 文件加载敏感凭据的 dotenv、用于类型提示的 typing（具体来说，是用于兼容旧版 Python 的 Union）、FastMCP 服务类、Watsonx.ai SDK 组件（APIClient、Credentials、ModelInference）、用于向量数据库的 chromadb，以及用于读取 PDF 文件的 pypdf。
• **配置（load_dotenv、常量）：**我们使用 load_dotenv() 加载环境变量（例如 API 密钥）。我们为 Watsonx.ai 的详细信息、文档路径和数据库持久化目录、数据库集合名称，以及控制 RAG 流程的参数（结果数量、分块大小、重叠长度）定义常量。使用常量能让代码更易于阅读和修改。
• **全局变量（pdf_collection、chroma_client）：**这些变量在全局范围内声明，以便在启动时初始化一次，之后由 MCP 工具函数（chat_with_manual）访问。Union[..., None] 表示它们可能保存特定类型的值，也可能在初始化前为 None。
• **日志记录与验证：**我们设置基本日志，以了解服务正在执行的操作。至关重要的是，我们会检查环境中是否提供了必要的 Watsonx.ai 凭据；如果没有，脚本就会报错并停止。

2. 提取知识：处理 PDF（pdf_to_text）

在查询文档之前，我们需要先从中提取文本。

def pdf_to_text(folder: Path) -> dict:
    """Extract text from every PDF in `folder`, preserving structure."""
    texts = {}
    if not folder.is_dir():
        logging.error("Folder not found: %s", folder)
        return texts
    logging.info("Scanning folder '%s' for PDFs...", folder)
    pdf_files = list(folder.glob("*.pdf"))
    if not pdf_files:
        logging.warning("No PDF files found in folder '%s'.", folder)
        return texts
    for pdf_path in pdf_files:
        try:
            logging.info("Attempting to read %s", pdf_path.name)
            reader = PdfReader(str(pdf_path))
            # Handle encrypted PDFs (basic attempt)
            if reader.is_encrypted:
                logging.warning("PDF is encrypted: %s. Attempting default decryption.", pdf_path.name)
                try: reader.decrypt(''); logging.info("Successfully decrypted %s.", pdf_path.name)
                except Exception as de: logging.error("Failed decrypt %s: %s.", pdf_path.name, de); continue
            # Extract text page by page and join
            content = "\n".join(page.extract_text() or "" for page in reader.pages)
            # Basic cleaning: reduce excessive newlines
            content = re.sub(r'\n{3,}', '\n\n', content.strip())
            if content:
                texts[pdf_path.name] = content
                logging.info("Extracted %d chars from %s", len(content), pdf_path.name)
            else:
                logging.warning("Extracted no text from %s.", pdf_path.name)
        except Exception as e:
            logging.error("Error reading %s: %s", pdf_path.name, e)
    if not texts:
        logging.warning("No text extracted from any PDF.")
    return texts

说明：

• pdf_to_text 函数以 DOCS_FOLDER 路径作为输入。
• 它会扫描文件夹，查找所有以 .pdf 结尾的文件。
• 对于每个 PDF 文件，它会使用 pypdf.PdfReader 打开文件。
• 它会进行基本的加密检查，并尝试使用空密码解密（这是默认保护的常见情况）。
• 它会遍历每一页，使用 page.extract_text() 提取文本，并以换行符连接所有页面的文本。
• 它会使用简单的正则表达式（re.sub）清理多余的空行。
• 每个 PDF 提取出的文本都会存储在一个字典中，字典将文件名映射到文本内容。
• 文件读取问题会通过错误处理机制处理。

3. 拆分内容：文本分块（chunk_text）

大模型一次能够处理的文本量有限（上下文窗口）。此外，将文本拆分成较小、主题集中的片段，有助于更好地搜索特定信息。这正是分块的作用。

def chunk_text(text: str, chunk_size: int, chunk_overlap: int) -> list[str]:
    """Recursively splits text into chunks of a target size with overlap."""
    if not text: return []
    # Prioritized list of separators to split by
    separators = ["\n\n", "\n", ". ", " ", ""] # From paragraphs down to characters

    def split(text_to_split, current_separators):
        # Base case: If text is small enough or no more separators to try
        if not text_to_split or len(text_to_split) <= chunk_size:
             return [text_to_split] if text_to_split and text_to_split.strip() else []

        current_separator = current_separators[0]
        next_separators = current_separators[1:]

        # Handle splitting by character if all else fails
        if current_separator == "":
            # Simple character-based splitting with overlap
            chunks = []
            start = 0
            while start < len(text_to_split):
                end = start + chunk_size
                chunk = text_to_split[start:end]
                if chunk.strip():
                    chunks.append(chunk)
                # Move start, considering overlap
                start += chunk_size - chunk_overlap
                if start >= len(text_to_split): break # Prevent infinite loop
            return chunks

        # Try splitting by the current separator
        splits = [s for s in text_to_split.split(current_separator) if s.strip()]
        if not splits: # If this separator didn't work, try the next one
             return split(text_to_split, next_separators) if next_separators else split(text_to_split, [""])


        final_chunks = []
        current_chunk_parts = []
        current_length = 0
        separator_len = len(current_separator)

        # Try to combine smaller splits into chunks respecting chunk_size
        for i, part in enumerate(splits):
            part_len = len(part)
            potential_length = current_length + part_len + (separator_len if current_chunk_parts else 0)

            if potential_length <= chunk_size:
                # Add to current chunk
                current_chunk_parts.append(part)
                current_length = potential_length
            else:
                # Current chunk is full, or this part is too large
                if current_chunk_parts:
                    # Finalize the current chunk
                    chunk = current_separator.join(current_chunk_parts)
                    if chunk.strip(): final_chunks.append(chunk)

                    # --- Overlap Logic (Simplified Approach) ---
                    # For simplicity, we restart the next chunk with the current part.
                    # More complex overlap (like carrying over the end of the last chunk)
                    # can be implemented but adds complexity. A simple restart ensures
                    # the current part isn't lost.
                    current_chunk_parts = [part]
                    current_length = part_len

                    # If the current part *itself* is too big, split it further
                    if part_len > chunk_size:
                        final_chunks.extend(split(part, next_separators if next_separators else [""]))
                        current_chunk_parts = [] # Reset as it was handled recursively
                        current_length = 0

                else: # current_chunk_parts is empty: this 'part' alone is > chunk_size
                    final_chunks.extend(split(part, next_separators if next_separators else [""]))
                    current_chunk_parts = [] # Ensure reset
                    current_length = 0

            # Add the last assembled chunk if any parts remain
            if i == len(splits) - 1 and current_chunk_parts:
                chunk = current_separator.join(current_chunk_parts)
                if chunk.strip(): final_chunks.append(chunk)

        return [c for c in final_chunks if c.strip()] # Final cleanup

    # Start the splitting process with the initial text and separators
    return split(text, separators)

说明：

• chunk_text 函数采用递归策略拆分文本。
• 它会优先尝试使用有意义的分隔符拆分文本（段落 \n\n、行 \n、句子 . ），然后才使用空格或单个字符进行拆分。
• 它会尝试创建接近目标 CHUNK_SIZE 的文本块。
• CHUNK_OVERLAP 可确保文本块之间的上下文不会被突然截断，从而帮助检索流程找到可能跨越文本块边界的相关信息。如果一个句子在文本块 A 末尾开始，并在文本块 B 开头结束，重叠区域就有助于将其完整保留。
• 如果某个“部分”（例如，没有换行符的超长段落）仍然大于 CHUNK_SIZE，递归会进一步将其拆分。

4. 构建知识库：创建向量数据库（build_new_vector_db）

此函数负责协调整个流程：获取已提取并分块的文本，并将其索引到 ChromaDB 向量数据库中。仅当数据库不存在或为空时才会运行此函数。

def build_new_vector_db(client: chromadb.PersistentClient, docs_folder: Path) -> bool:
    """Extracts, chunks, and indexes PDFs into the specified collection."""
    global pdf_collection # To assign the newly created/updated collection

    logging.info("--- Starting Vector DB Creation/Update Process ---")
    try:
        # 1. Extract text from PDFs
        extracted = pdf_to_text(docs_folder)
        if not extracted:
            logging.error("No text extracted, cannot build vector DB.")
            return False

        # 2. Get or Create the ChromaDB Collection
        logging.info("Accessing collection '%s'...", COLLECTION_NAME)
        # get_or_create is safe: it gets if exists, creates if not
        collection = client.get_or_create_collection(name=COLLECTION_NAME)
        logging.info("Using collection '%s'. Current item count: %d", COLLECTION_NAME, collection.count())

        # 3. Chunk Text and Prepare for Indexing
        docs, metas, ids = [], [], []
        logging.info("Chunking text with target size ~%d chars, overlap %d chars...", CHUNK_SIZE, CHUNK_OVERLAP)
        total_chunks = 0
        for fname, text in extracted.items():
            if not text.strip(): continue # Skip empty documents
            file_chunks = chunk_text(text, CHUNK_SIZE, CHUNK_OVERLAP)
            logging.info("Generated %d chunks for %s.", len(file_chunks), fname)
            if not file_chunks: continue
            for idx, chunk in enumerate(file_chunks, start=1):
                docs.append(chunk) # The text chunk itself
                metas.append({"source": fname, "chunk_idx": idx}) # Metadata: filename and chunk number
                # Create a unique ID for each chunk
                safe_fname = re.sub(r'[^\w-]', '_', fname) # Make filename safe for ID
                unique_id = f"{safe_fname}_chunk_{idx}"
                ids.append(unique_id)
            total_chunks += len(file_chunks)

        if not docs:
            logging.warning("No text chunks generated for indexing.")
            pdf_collection = collection # Assign the (empty) collection handle
            return True # Succeeded, but nothing to add

        # 4. Add data to ChromaDB in Batches
        logging.info("Prepared %d total chunks for indexing.", total_chunks)
        batch_size = 100 # Process 100 chunks at a time
        num_batches = (len(docs) + batch_size - 1) // batch_size
        for i in range(0, len(docs), batch_size):
            batch_docs = docs[i:i+batch_size]
            batch_metas = metas[i:i+batch_size]
            batch_ids = ids[i:i+batch_size]
            try:
                # Use add() - Chroma handles embedding generation automatically
                collection.add(documents=batch_docs, metadatas=batch_metas, ids=batch_ids)
                logging.info("Indexed batch %d/%d (size %d chunks).",
                             (i // batch_size) + 1, num_batches, len(batch_docs))
            except Exception as e:
                logging.error("Error adding batch starting at index %d: %s", i, e)
                # Decide whether to continue or stop on batch error

        # 5. Finalize and assign global variable
        final_count = collection.count()
        logging.info("Finished indexing. Collection '%s' now contains %d items.",
                     collection.name, final_count)
        pdf_collection = collection # Make the populated collection globally available
        return True

    except Exception as e:
        logging.error("Failed during Vector DB build process: %s", e, exc_info=True)
        return False

说明：

• **工作流：**调用 pdf_to_text，然后遍历提取出的文本，为每份文档调用 chunk_text。
• **数据准备：**此函数会准备三个列表：docs（实际的文本块）、metas（与各文本块关联的元数据，例如源文件名）以及 ids（每个文本块的唯一标识符）。唯一 ID 对于日后更新或删除特定文本块非常重要。
• **ChromaDB 交互：**此函数使用 client.get_or_create_collection 安全地获取集合的句柄；如果是首次使用，则创建该集合。
• **批量索引：**此函数通过 collection.add() 将文档、元数据和 ID 添加到 ChromaDB 集合中。为提高效率，并避免一次性添加可能多达数千个文本块而导致数据库连接或内存不堪重负，添加操作会分批进行（batch_size = 100）。ChromaDB 会自动使用默认的嵌入模型，将文本块转换为向量嵌入。
• **全局赋值：**填充完毕后，collection 对象会被赋值给全局变量 pdf_collection。

5. 智能启动：初始化向量数据库（initialize_vector_db）

此函数对于数据持久化至关重要。我们不希望每次服务启动时都重建向量数据库，特别是数据库较大的时候。此函数会检查指定的持久化目录中是否已有数据库；如果存在，则加载数据库，否则触发构建流程。

def initialize_vector_db(persist_path: Path, docs_folder_path: Path):
    """Initializes a persistent ChromaDB client and ensures the collection exists."""
    global pdf_collection, chroma_client
    logging.info("--- Initializing Persistent Vector DB ---")
    logging.info("Storage directory: %s", persist_path)
    # Ensure the directory exists
    persist_path.mkdir(parents=True, exist_ok=True)

    try:
        # 1. Initialize Persistent Client
        # This tells ChromaDB to save data to the specified path
        chroma_client = chromadb.PersistentClient(path=str(persist_path))
        logging.info("ChromaDB Persistent Client initialized.")
    except Exception as e:
        logging.error("Fatal: Failed to initialize ChromaDB Persistent Client: %s", e, exc_info=True)
        return False # Cannot continue without a client

    try:
        # 2. Check if the Collection Already Exists
        logging.info("Checking for existing collection: '%s'", COLLECTION_NAME)
        existing_collections = chroma_client.list_collections()
        collection_exists = any(col.name == COLLECTION_NAME for col in existing_collections)

        if collection_exists:
            # 3a. Load Existing Collection
            logging.info("Collection '%s' found. Getting handle.", COLLECTION_NAME)
            collection = chroma_client.get_collection(name=COLLECTION_NAME)
            count = collection.count()
            if count > 0:
                # Collection exists and has data - load it
                logging.info("Existing collection has %d items. Using existing.", count)
                pdf_collection = collection # Assign to global variable
                return True # Success!
            else:
                # Collection exists but is empty - build it
                logging.warning("Found existing collection '%s', but it is empty. Will attempt to build.", COLLECTION_NAME)
                return build_new_vector_db(chroma_client, docs_folder_path)
        else:
            # 3b. Build New Collection
            logging.info("Collection '%s' not found. Proceeding to build a new one.", COLLECTION_NAME)
            return build_new_vector_db(chroma_client, docs_folder_path)

    except Exception as e:
        logging.error("Error during Vector DB initialization/check: %s", e, exc_info=True)
        return False # Failed to initialize or build

说明：

• 持久化：chromadb.PersistentClient(path=...) 是关键。它会告知 ChromaDB 将数据保存到指定目录（CHROMA_PERSIST_DIR）。
• **构建前检查：**此函数使用 chroma_client.list_collections() 检查持久化存储中是否已存在名为 COLLECTION_NAME 的集合。
• 加载或构建逻辑：
  • 如果集合存在且包含项目（count > 0），则使用 chroma_client.get_collection() 加载集合，并将其赋值给全局变量 pdf_collection。
  • 如果集合存在但为空，或者根本不存在，则调用 build_new_vector_db 创建并填充集合。
• **全局赋值：**初始化后的 chroma_client 也会存储为全局变量。

6. 连接大脑：初始化 Watsonx.ai

在这里，我们使用 SDK 和凭据设置与 IBM Watsonx.ai 服务的连接。

# --- Watsonx.ai Initialization ---
try:
    creds  = Credentials(url=URL, api_key=API_KEY)
    client = APIClient(credentials=creds, project_id=PROJECT_ID)
    model = ModelInference(
        model_id=MODEL_ID,
        credentials=creds,
        project_id=PROJECT_ID
    )
    logging.info(
        f"Initialized Watsonx.ai model '{MODEL_ID}' for project '{PROJECT_ID}'."
    )
except Exception as e:
    logging.error("Failed to initialize Watsonx.ai client/model: %s", e, exc_info=True)
    sys.exit(1) # Exit if we can't connect to the LLM service

说明：

• 此函数使用从 .env 文件加载的 API 密钥和 URL 创建 Credentials。
• 此函数会实例化一个 APIClient，用于常规交互（尽管此处不会直接用它来生成内容）。
• 此函数会创建一个 ModelInference 对象，并指定所需的 MODEL_ID（例如，“ibm/granite-13b-instruct-v2”）、凭据和 PROJECT_ID。此 model 对象将用于生成答案。
• 错误处理可确保如果无法初始化与 Watsonx.ai 的连接，服务就会退出。

7. 打开大门：设置 MCP 服务

我们需要一种方式，让客户端（例如命令行界面或 Web 应用）能够与后端逻辑通信。FastMCP 提供了一种简单的方式来公开 Python 函数。

# --- MCP Server Setup ---
mcp = FastMCP("Watsonx RAG Chatbot Server")

说明：

• 此处会创建一个 FastMCP 实例。字符串参数只是服务的名称，通常用于服务发现或日志记录。默认情况下，FastMCP 使用标准输入/输出（STDIO）进行通信，适用于以子进程方式启动服务的简单客户端-服务端设置。

8. 核心逻辑：RAG 工具（chat_with_manual）

这是 RAG 流程的核心，通过 MCP 暴露为可调用的工具。

# --- RAG Tool Definition ---
@mcp.tool()
def chat_with_manual(query: str) -> str:
    """
    Answers questions about drone manuals using RAG with Watsonx.ai.
    Relies on the globally initialized pdf_collection.
    """
    global pdf_collection # Access the globally initialized DB collection
    logging.info("Received RAG query: %r", query)

    # Ensure the vector DB is ready
    if pdf_collection is None:
        logging.error("Vector DB collection is not available. Initialization might have failed.")
        return "Error: The document database is not ready. Please check server logs."

    # 1. Retrieve: Query the Vector DB
    try:
        logging.info("Querying Vector DB ('%s') for top %d results...", pdf_collection.name, NUM_RESULTS_RAG)
        # Find chunks most similar to the user's query
        results = pdf_collection.query(
            query_texts=[query],          # The user's question
            n_results=NUM_RESULTS_RAG,    # How many results to fetch (configured earlier)
            include=['documents']         # We need the actual text content ('documents')
        )
    except Exception as e:
        logging.error("Error querying ChromaDB: %s", e, exc_info=True)
        return f"Error: Could not retrieve information from the document database."

    # 2. Augment: Prepare the Context
    retrieved_docs = results.get('documents', [[]])[0] # Extract the list of document texts
    logging.info(f"Retrieved documents: {retrieved_docs}") # Log the actual content

    if not retrieved_docs:
        # Handle case where no relevant documents were found
        logging.warning("No relevant documents found in Vector DB for query: %r", query)
        # Return a message indicating nothing was found (alternative: let the LLM try without context)
        return "I couldn't find specific information about that in the available documents."
    else:
        # Combine the retrieved chunks into a single context string
        logging.info("Retrieved %d document chunks.", len(retrieved_docs))
        context_string = "\n\n---\n\n".join(retrieved_docs) # Separate chunks clearly

    # 3. Generate: Construct Prompt and Call LLM
    # Create a prompt that instructs the LLM to use *only* the provided context
    prompt_template = f"""
    You are a helpful assistant that answers questions based *only* on the information provided from the manual below.

    --- Manual Context ---
    {context_string}
    --- End of Context ---

    Using only the context above, please answer the following question clearly and accurately.

    Question: {query}

    Answer:
    """
    logging.info("Constructed prompt for Watsonx.ai (length: %d chars)", len(prompt_template))
    # Use DEBUG level for logging full prompts to avoid cluttering INFO logs
    logging.debug("Prompt:\n%s", prompt_template)
    # Define LLM generation parameters
    params = {
        GenParams.DECODING_METHOD: "greedy", # Simple, deterministic output
        GenParams.MAX_NEW_TOKENS:  300,     # Max length of the generated answer
        GenParams.MIN_NEW_TOKENS:  10,      # Min length of the generated answer
        # GenParams.TEMPERATURE: 0.7, # Uncomment for more varied/creative answers
        GenParams.STOP_SEQUENCES: ["\n\n", "---", "Question:", "Context:"] # Stop generation if these appear
    }
    logging.info("Sending request to Watsonx.ai model '%s'...", MODEL_ID)

    # Call the Watsonx.ai model
    try:
        resp = model.generate_text(prompt=prompt_template, params=params, raw_response=True)
        # Carefully parse the response structure
        if resp and isinstance(resp, dict) and "results" in resp and isinstance(resp["results"], list) and len(resp["results"]) > 0:
            first_result = resp["results"][0]
            if isinstance(first_result, dict) and "generated_text" in first_result:
                answer = first_result["generated_text"].strip()
                logging.info("Received Watsonx.ai response: %r", answer)
                # Simple post-processing to remove trailing incomplete sentences
                if answer.endswith("..."):
                    last_period = answer.rfind('.')
                    if last_period != -1:
                        answer = answer[:last_period+1]
                return answer
            else:
                logging.error("Watsonx.ai response structure unexpected ('generated_text' missing): %s", first_result)
                return "Error: Received an unexpected response format from the AI model (Detail 1)."
        else:
            logging.error("Watsonx.ai response structure unexpected (main structure): %s", resp)
            return "Error: Received an unexpected response format from the AI model (Detail 2)."
    except Exception as e:
        logging.error("Watsonx.ai inference error: %s", e, exc_info=True)
        return f"Error: Failed to generate an answer due to an AI model issue."

说明：

• @mcp.tool()： 此装饰器会将 chat_with_manual 函数注册到 MCP 服务，使客户端可以调用它。
• 全局访问： 它会访问之前初始化的 pdf_collection。
• 检索： 它使用 pdf_collection.query() 查找 NUM_RESULTS_RAG 个文档块，这些文档块的向量嵌入与用户 query 的向量嵌入最接近（相似度最高）。它会明确要求在结果中包含 documents（文本）。
• 增强： 它会检查是否找到了文档。如果找到了，就将文档文本内容合并为一个 context_string，并以 \n\n---\n\n 分隔，方便阅读。如果没有找到文档，则返回一条特定消息。
• 生成：
  • 提示词工程： 会创建一条精心设计的提示词（prompt_template）。这对 RAG 至关重要。提示词会明确告知 LLM（Watsonx.ai 模型）其角色，并指示它只能根据提供的 context_string 回答用户的 query。这样可以防止 LLM 使用其通用知识，并强制它遵循文档内容。
  • 参数： 定义生成参数，例如 max_new_tokens（答案长度）、decoding_method（LLM 选择下一个词的方式——“greedy”是确定性的）和 stop_sequences（用于指示 LLM 停止生成的序列）。
  • LLM 调用： model.generate_text() 会将提示词和参数发送给 Watsonx.ai。
  • 响应处理： 代码会解析 Watsonx.ai 返回的 JSON 响应，提取 generated_text。基本的后处理会尝试清理可能突然中断的答案。对于 ChromaDB 查询和 Watsonx.ai API 调用，代码都包含了错误处理。

9. 开始运行：主执行代码块

最后，if __name__ == "__main__": 代码块可确保只有在直接运行脚本时（而不是将其作为模块导入时），才会执行以下代码。

# --- Main Execution ---
if __name__ == "__main__":
    # 1. Initialize/Load the Persistent Vector DB *before* starting the server
    # This ensures the RAG tool has data to query
    if not initialize_vector_db(CHROMA_PERSIST_DIR, DOCS_FOLDER):
         logging.error("CRITICAL: Failed to initialize Vector DB. Exiting.")
         sys.exit(1) # Stop server if DB isn't ready

    # 2. Start the MCP server (this call is blocking)
    # It will listen for client connections (on STDIO by default)
    # and dispatch calls to registered tools like chat_with_manual
    logging.info("Starting MCP server on STDIO transport...")
    mcp.run()

    # This line is reached only when the server stops (e.g., Ctrl+C)
    logging.info("MCP server stopped.")

说明：

• 先初始化： 首先调用 initialize_vector_db。这至关重要——在确认向量数据库已就绪（已加载或刚刚构建完成）之前，服务不应开始接收请求。如果初始化失败，脚本就会退出。
• 启动服务： 如果数据库已就绪，mcp.run() 就会启动 MCP 服务。此函数通常会阻塞运行，也就是说，脚本会在此等待，监听传入的客户端请求，并将其路由到相应的 @mcp.tool 函数（本例中为 chat_with_manual）。
• 关闭： 只有在服务关闭后（例如，客户端正常断开连接，或收到 Ctrl+C 等中断信号），脚本才会继续执行 mcp.run() 之后的代码。

这种模块化结构清楚地展示了数据如何从原始 PDF 转化为 LLM 生成的上下文相关答案。下一步是创建一个客户端应用，连接到此 MCP 服务并与 chat_with_manual 工具交互。

第 4 步：与 RAG 机器人对话：client.py 脚本

在上一节中，我们构建了功能强大的 RAG 聊天机器人后端 server.py。但我们要如何实际使用它呢？这正是 client.py 发挥作用的地方。此脚本充当用户界面（目前是一个简单的命令行界面），负责启动服务、发送问题，并显示从 RAG 系统收到的答案。

它使用 Meta Compute Protocol（MCP）客户端库，专门配置为通过标准输入/输出（STDIO）进行通信。这非常适合与我们在 server.py 中设置的 FastMCP 服务交互。

以下是完整的 client.py 代码：

# client.py
import asyncio
import logging
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

# Configure basic logging for the client
logging.basicConfig(level=logging.INFO, format='%(asctime)s [Client] %(message)s')

async def main():
    # Command to start the server.py script using the python interpreter
    server_params = StdioServerParameters(command="python", args=["server.py"])
    logging.info("Attempting to start and connect to MCP server ('server.py')...")

    try:
        # 1. Start the server and establish connection via STDIO
        async with stdio_client(server_params) as (reader, writer):
            logging.info("Connected to MCP server. Initializing session...")
            # 2. Establish MCP Session using the reader/writer streams
            async with ClientSession(reader, writer) as session:
                await session.initialize() # Perform MCP handshake
                logging.info("MCP session initialized.")

                # --- Interaction Logic ---
                # Define the question to ask the RAG system
                user_msg = "How many flight modes it has and explain them?"
                # Example alternative question:
                # user_msg = "What is the maximum flight time?"

                logging.info("Calling tool 'chat_with_manual' with query: %r", user_msg)

                try:
                    # 3. Call the remote tool on the server
                    response = await session.call_tool(
                        "chat_with_manual",          # Name must match the @mcp.tool function in server.py
                        arguments={"query": user_msg} # Arguments expected by the tool
                    )
                    logging.info("Received response from server.")

                    # 4. Process and Display the Response
                    # MCP might wrap responses; attempt to extract raw text if needed
                    final_answer = response # Default to the raw response
                    if hasattr(response, 'content') and response.content and isinstance(response.content, list):
                        # Check if response has structured content
                        first_content = response.content[0]
                        if hasattr(first_content, 'text'):
                            final_answer = first_content.text # Extract text if available

                    # Print clearly formatted question and answer
                    print("\n" + "="*20 + " Query " + "="*20)
                    print(f"Your Question: {user_msg}")
                    print("\n" + "="*20 + " Answer " + "="*20)
                    print(f"Bot Answer:\n{final_answer}")
                    print("\n" + "="*50)

                except Exception as e:
                    # Handle errors during the tool call specifically
                    logging.error("Error calling tool 'chat_with_manual': %s", e, exc_info=True)
                    print(f"\nError interacting with the chatbot: {e}")

    except Exception as e:
        # Handle errors during connection or session setup
        logging.error("Failed to connect or communicate with the MCP server: %s", e, exc_info=True)
        print(f"\nCould not connect to or run the server: {e}")

if __name__ == "__main__":
    # Run the asynchronous main function
    asyncio.run(main())

拆解 client.py

下面来逐步了解这个客户端的工作原理：

1. 导入和日志记录：

   • asyncio：必不可少，因为 MCP 客户端操作（连接、发送、接收）都是异步的。
   • mcp：我们导入 ClientSession 来管理通信协议，导入 StdioServerParameters 来定义如何启动服务，以及导入 stdio_client。后者是一个上下文管理器，负责启动服务并提供通信流。
   • logging：设置基本日志记录，用于显示客户端活动。

2. main 异步函数： 客户端的核心逻辑就在这里。

   • 服务定义（StdioServerParameters）： 我们告诉客户端如何运行服务。StdioServerParameters(command="python", args=["server.py"]) 的意思是“执行命令 python server.py”。客户端会管理这个子进程。
   • 连接（stdio_client）： async with stdio_client(server_params) as (reader, writer): 代码块负责处理繁重工作：
     • 在后台启动 python server.py 进程。
     • 捕获服务的标准输入和标准输出。
     • 提供 reader 和 writer 对象，让客户端能够向服务的 stdin 发送数据，并异步读取服务 stdout 中的数据。
     • 重要的是，退出 async with 代码块时（正常退出或发生错误），它还会确保服务进程正确终止。
   • MCP 会话（ClientSession）： 在 stdio_client 代码块中，async with ClientSession(reader, writer) as session: 会将原始的读写流封装为 MCP 会话处理器。该对象知道如何按照 MCP 协议格式化消息。
   • 初始化（session.initialize()）： 在发起调用之前，客户端和服务会进行一次简短的握手，以确保双方已准备好通过 MCP 协议进行通信。
   • 定义查询（user_msg）： 一个简单的字符串变量，用于保存我们要向 RAG 机器人提出的问题。
   • 调用工具（session.call_tool）： 这是关键的交互点。
     • await session.call_tool(...)：通过已建立的会话向服务发送请求。
     • "chat_with_manual"：这个字符串必须与 server.py 中使用 @mcp.tool() 装饰的函数名称完全一致。
     • arguments={"query": user_msg}：这个字典包含要传递给服务端函数的参数。键 "query" 必须与服务端定义 chat_with_manual(query: str) 中的参数名称一致。
     • 然后客户端会使用 await 等待服务端响应。
   • 处理响应： response 变量保存 chat_with_manual 函数在服务端返回的内容（应为包含答案的字符串）。代码中包含一个检查（if hasattr(response, 'content')...），用于处理 MCP 可能将响应包装在更复杂对象中的情况，并尝试提取原始文本。最后，代码会以格式化的方式打印原始问题和收到的答案。
   • 错误处理： 使用 try...except 代码块捕获可能出现的问题，例如服务启动失败、连接问题，或服务端在调用工具时发生的错误（这些错误会被传回）。

3. 运行客户端（if __name__ == "__main__":）

   • 这是标准的 Python 结构，用于确保其中的代码仅在直接执行脚本时运行。
   • asyncio.run(main())：由于 main 是一个 async 函数，我们需要使用 asyncio 运行其事件循环并执行客户端的异步逻辑。

完成整个流程

这个 client.py 脚本成功搭建了用户与功能强大的 server.py 后端之间的桥梁。运行 python client.py 后，用户即可触发完整的 RAG 流程：客户端启动服务端，服务端初始化向量数据库（加载或构建数据库），客户端发送查询，服务端从数据库中检索相关上下文，将上下文和查询发送给 Watsonx.ai，获取生成的答案，再将答案发送回客户端，最后由客户端向用户显示答案。

步骤 5：设置依赖项

为确保安装了所有必需的库，请创建一个 requirements.txt 文件：

# requirements.txt
ibm-watsonx-ai>=1.0.0
chromadb>=0.4.0 # Use a recent version
pypdf>=3.0.0
python-dotenv>=1.0.0
mcp-sdk>=0.3.0
# tiktoken might be needed by chromadb's default embedding model
tiktoken>=0.5.0

安装依赖项（确保已激活虚拟环境）：

pip install -r requirements.txt

步骤 6：运行 RAG 服务端和客户端

你也可以选择启动服务端

python server.py

客户端脚本会自动启动服务端。你只需运行客户端。

打开终端（确保当前位于 watsonx_rag_mcp_server 目录中，且虚拟环境处于激活状态），然后运行：

python client.py

运行过程：

1. client.py 启动。
2. 它会作为子进程执行 python server.py。
3. server.py 启动并初始化与 Watsonx.ai 的连接。
4. server.py 检查 ChromaDB 数据库。
   • 首次运行： 找不到数据库，因此会处理 documents 文件夹中的 PDF，将其分块，并把嵌入向量保存到 ./chroma_db_data 目录中。所需时间取决于 PDF 的大小和数量。
   • 后续运行： 找到 ./chroma_db_data 目录及其中的集合后，会快速加载并跳过处理步骤。
5. server.py 启动 MCP 服务，并通过 STDIO 进行监听。
6. client.py 通过 STDIO 连接服务端并初始化 MCP 会话。
7. client.py 向 chat_with_manual 工具发送预设查询（“How many flight modes...”）。
8. server.py 收到请求后查询 ChromaDB，获取相关文本块，结合上下文构建提示词，调用 Watsonx.ai，获取答案并将其发回客户端。
9. client.py 接收答案并将其打印到终端。

步骤 7：查询 RAG 服务端

运行 client.py 时，它会提出硬编码的问题：

Your Question: How many flight modes does the drone have and explain them?

你应会看到 Watsonx.ai 生成的回答，该回答仅基于你提供的 PDF 文档中的信息。回答的质量和完整性很大程度上取决于相关信息是否存在于文档中，以及向量搜索能否有效检索到这些信息。

核心组件说明

• IBM Watsonx.ai 集成： 提供大语言模型（LLM），用于理解文档提供的上下文，并针对查询生成自然流畅的回答。我们使用 ibm-watsonx-ai SDK 与其交互。
• 用于向量索引的 ChromaDB： 充当 RAG 系统的记忆。它会将 PDF 中的文本块转换为数值表示（向量嵌入），并使用嵌入模型进行存储（如果未指定，ChromaDB 会使用默认模型）。收到查询后，它会查找嵌入向量与查询嵌入向量最接近（语义最相似）的文本块。我们使用 chromadb.PersistentClient，确保数据库在多次运行之间得以保留。
• 模型上下文协议（MCP）： 为客户端和服务端提供标准化的通信方式。我们无需定义自定义 API 端点，而是使用 MCP 的概念：
  • 工具： 我们的 chat_with_manual 函数作为 MCP 工具公开。客户端调用此工具来执行 RAG 操作。这类似于专用函数调用，或 Web 领域中的 POST 请求。
  • (资源和提示词)： 虽然本例中没有大量使用，但 MCP 也允许将数据源直接公开为资源（类似 GET 请求），并定义可重复使用的提示词。

提示和故障排除

• 检查日志： server.py 和 client.py 都会生成详细日志。如果出现问题，请检查终端输出中的 [INFO]、[WARNING] 或 [ERROR] 消息。必要时提高日志级别（例如 logging.DEBUG），以获取更多详细信息。
• .env 配置： 仔细检查 .env 文件中的 API 密钥、URL 和 Project ID 是否正确，并确认文件已加载（文件名没有拼写错误）。
• PDF 可访问性： 确保 documents 文件夹中的 PDF 文件可读且未受密码保护（或者 pdf_to_text 中的简单解密尝试能够正常工作）。损坏的 PDF 可能会导致错误。
• 依赖项： 确保已在当前虚拟环境中正确安装 requirements.txt 中列出的所有库。
• 强制重建： 如果更新了 PDF 并希望服务重新为其建立索引，则需要在再次运行 client.py 之前删除 chroma_db_data 目录。之后，服务会检测到数据库不存在，并触发 build_new_vector_db 进程。
• Watsonx.ai 模型： 默认使用 ibm/granite-13b-instruct-v2。使用其他模型（如 Llama 3）时，结果可能不同，或者需要调整参数。你可以在 .env 文件中更改 MODEL_ID。

前端

让我们创建一个简单的 Flask Web 界面，提供类似 ChatGPT 的聊天体验，并与后端 server.py 交互。

该方案将：

1. 使用 Flask 作为 Web 框架。
2. 使用提供的 base.html 和 Watsonx 壁纸。
3. 使用 chat.html 模拟聊天界面。
4. 根据每条用户消息启动并与 server.py 通信，使用根据 client.py 改写的逻辑。**注意：**这种方式不适合生产环境（因为会反复重新初始化服务／数据库连接），但在标准 Flask 设置中无需复杂的异步管理，实现起来更简单。
5. 在 Flask 会话中维护对话历史。
6. 根据字符数（作为令牌数的近似值）裁剪对话历史，使其保持在可管理的范围内。

项目结构：

请确保文件按以下方式组织：

/chatbot
├── static/
│   └── assets/
│       └── watsonx-wallpaper.jpg  <-- Place your wallpaper image here
├── templates/
│   ├── base.html
│   └── chat.html
├── frontend.py            <-- The Flask app we will create
├── server.py              <-- Your existing server script
├── documents/             <-- Folder needed by server.py
│   └── ... (your PDFs)
├── .env                   <-- File needed by server.py
└── chroma_db_data/        <-- Directory created/used by server.py

1. 基础模板（templates/base.html）

这是你提供的模板，用于设置背景和 Tailwind CSS。

<!doctype html>
<html lang="en">
<head>
  <meta charset="utf-8">
  <title>{% block title %}Watsonx RAG Chat{% endblock %}</title>
  <link href="https://cdn.jsdelivr.net/npm/[email protected]/dist/tailwind.min.css" rel="stylesheet">
  <style>
    html, body {
      height: 100%;
      margin: 0;
    }
    body {
      background-image: url('{{ url_for("static", filename="assets/watsonx-wallpaper.jpg") }}');
      background-size: cover;
      background-position: center center;
      background-repeat: no-repeat;
      background-attachment: fixed; /* Keep wallpaper fixed during scroll */
      display: flex;
      flex-direction: column; /* Stack content vertically */
    }
    /* Ensure content block takes available space */
    .content-wrapper {
       flex-grow: 1;
       display: flex;
       align-items: center; /* Center content vertically */
       justify-content: center; /* Center content horizontally */
       width: 100%;
       padding: 1rem; /* Add some padding */
    }
  </style>
</head>
<body class="bg-gray-900">  {# Default background if image fails #}
  <div class="content-wrapper">
    {% block content %}{% endblock %}
  </div>
  {# Optional: Add scripts here if needed globally #}
  <script>
      // Auto-scroll to bottom of chat window
      function scrollToBottom() {
          const chatbox = document.getElementById('chatbox');
          if (chatbox) {
              chatbox.scrollTop = chatbox.scrollHeight;
          }
      }
      // Scroll on initial load
      window.onload = scrollToBottom;
      // If using HTMX or similar for partial updates, you might need
      // to call scrollToBottom() after content is swapped.
  </script>
</body>
</html>

（注：更新了 Tailwind CDN 链接，并添加了 background-attachment: fixed 和 Flexbox 布局，以便在固定背景下更好地实现内容居中和滚动。）

2. 聊天界面模板（templates/chat.html）

该模板扩展 base.html，并创建聊天 UI。

{% extends 'base.html' %}

{% block title %}Watsonx RAG Chat{% endblock %}

{% block content %}
<div class="bg-white bg-opacity-80 backdrop-filter backdrop-blur-sm rounded-lg shadow-xl w-full max-w-2xl h-full max-h-[85vh] flex flex-col">

  <div class="p-4 border-b border-gray-300 flex justify-between items-center">
    <h1 class="text-xl font-semibold text-gray-800">Chat with PDF Manuals</h1>
    <a href="{{ url_for('clear_chat') }}" class="text-sm text-blue-600 hover:text-blue-800">Clear Chat</a>
  </div>

  <div id="chatbox" class="flex-grow p-4 space-y-4 overflow-y-auto">
    {% if not history %}
      <div class="text-center text-gray-500">Ask a question about the PDF documents!</div>
    {% endif %}
    {% for message in history %}
      {% if message.role == 'user' %}
        <div class="flex justify-end">
          <div class="bg-blue-500 text-white p-3 rounded-lg max-w-xs lg:max-w-md shadow">
            {{ message.content }}
          </div>
        </div>
      {% elif message.role == 'assistant' %}
        <div class="flex justify-start">
          <div class="bg-gray-200 text-gray-800 p-3 rounded-lg max-w-xs lg:max-w-md shadow">
            {# Use safe filter or pre tags if response might contain HTML/code #}
            <pre class="whitespace-pre-wrap font-sans">{{ message.content }}</pre>
          </div>
        </div>
      {% elif message.role == 'error' %}
        <div class="flex justify-start">
          <div class="bg-red-100 text-red-700 p-3 rounded-lg max-w-xs lg:max-w-md shadow">
            <strong>Error:</strong> {{ message.content }}
          </div>
        </div>
      {% endif %}
    {% endfor %}
  </div>

  <div class="p-4 border-t border-gray-300">
    <form method="POST" action="{{ url_for('chat') }}" class="flex space-x-2">
      <input type="text" name="message" placeholder="Type your question..." autocomplete="off" required
             class="flex-grow px-3 py-2 border border-gray-300 rounded-md focus:outline-none focus:ring-2 focus:ring-blue-500">
      <button type="submit"
              class="bg-blue-600 text-white px-4 py-2 rounded-md hover:bg-blue-700 focus:outline-none focus:ring-2 focus:ring-blue-500 focus:ring-offset-2">
        Send
      </button>
    </form>
  </div>

</div>
{% endblock %}

（注：为助手消息添加了 pre 标签以保留换行等格式；在 base.html 中添加了 id="chatbox" 和用于自动滚动的基础 JavaScript；还添加了清除聊天链接和基础错误显示。）

3. Flask 应用（frontend.py）

这是 Web 前端的主 Python 脚本。

import os
import sys
import asyncio
import logging
import uuid # To generate unique session IDs if needed, but Flask session handles this
from flask import Flask, render_template, request, redirect, url_for, session

# --- MCP Client Imports ---
# Assuming mcp library is installed: pip install meta-compute-protocol
# Make sure server.py is in the same directory or Python path
try:
    from mcp import ClientSession, StdioServerParameters
    from mcp.client.stdio import stdio_client
    # from mcp.common.content import TextContent # Might be needed depending on exact response wrapping
except ImportError:
    print("ERROR: 'meta-compute-protocol' library not found.")
    print("Please install it: pip install meta-compute-protocol")
    sys.exit(1)

# --- Configuration ---
logging.basicConfig(level=logging.INFO, format='%(asctime)s [Frontend] %(levelname)s: %(message)s')

# --- Flask App Setup ---
app = Flask(__name__)
# IMPORTANT: Change this to a random secret key for production!
app.secret_key = os.environ.get("FLASK_SECRET_KEY", "dev-secret-key-replace-me")

# --- Conversation Memory Configuration ---
MAX_HISTORY_CHARS = 4000 # Approximate token limit (adjust as needed) ~1000 tokens

# --- Helper Function for MCP Interaction ---

async def async_mcp_call(user_query: str):
    """
    Starts server.py, connects via MCP, calls the tool, and returns the response.
    This runs the full client logic for each call - inefficient but simpler for Flask.
    """
    server_params = StdioServerParameters(command=sys.executable, args=["server.py"]) # Use sys.executable for portability
    logging.info("Attempting to start and connect to MCP server ('server.py')...")
    response_text = None
    error_message = None

    try:
        async with stdio_client(server_params) as (reader, writer):
            logging.info("Connected to MCP server. Initializing session...")
            async with ClientSession(reader, writer) as mcp_session:
                await mcp_session.initialize()
                logging.info("MCP session initialized.")
                logging.info("Calling tool 'chat_with_manual' with query: %r", user_query)

                try:
                    response = await mcp_session.call_tool(
                        "chat_with_manual",
                        arguments={"query": user_query}
                    )
                    logging.info("Received response from server.")

                    # Process response (handle potential wrapping)
                    final_answer = response
                    # Example check if response is wrapped like TextContent (adjust based on actual MCP response)
                    if hasattr(response, 'content') and response.content and isinstance(response.content, list):
                         first_content = response.content[0]
                         if hasattr(first_content, 'text'):
                             final_answer = first_content.text
                    elif isinstance(response, str): # If it's already a string
                         final_answer = response

                    # Ensure we have a string before returning
                    response_text = str(final_answer) if final_answer is not None else "Received an empty response."


                except Exception as tool_call_e:
                    logging.error("Error calling tool 'chat_with_manual': %s", tool_call_e, exc_info=True)
                    error_message = f"Error calling backend tool: {tool_call_e}"

    except Exception as connection_e:
        logging.error("Failed to connect or communicate with the MCP server: %s", connection_e, exc_info=True)
        error_message = f"Could not connect to or run the server: {connection_e}"

    # Ensure server process is terminated (stdio_client context manager handles this)
    logging.info("MCP client connection closed.")
    return response_text, error_message


def get_rag_response(user_query: str) -> tuple[str | None, str | None]:
    """
    Synchronous wrapper to run the asynchronous MCP call.
    Returns (response_text, error_message)
    """
    # --- Running asyncio logic within Flask ---
    # Note: Running asyncio.run() inside a synchronous Flask route handler
    # is generally discouraged for production scalability, but works for simple demos.
    # More robust solutions involve running the server independently (e.g., via TCP)
    # or using async Flask extensions.
    try:
        # For Python 3.7+
        response_text, error_message = asyncio.run(async_mcp_call(user_query))
        return response_text, error_message
    except RuntimeError as e:
        # Handle cases like "asyncio.run() cannot be called from a running event loop"
        # This might happen in certain deployment scenarios or with specific Flask extensions.
        logging.error("Asyncio runtime error: %s. Trying get_event_loop().run_until_complete()", e)
        try:
            loop = asyncio.get_event_loop()
            response_text, error_message = loop.run_until_complete(async_mcp_call(user_query))
            return response_text, error_message
        except Exception as fallback_e:
             logging.error("Fallback asyncio execution failed: %s", fallback_e)
             return None, f"Internal server error during async execution: {fallback_e}"
    except Exception as e:
        logging.error("Unexpected error during RAG response retrieval: %s", e)
        return None, f"Unexpected error: {e}"


# --- Conversation History Management ---

def trim_history(history: list, max_chars: int) -> list:
    """Removes oldest messages if total character count exceeds max_chars."""
    current_chars = sum(len(msg.get('content', '')) for msg in history)
    # Keep removing the oldest messages (index 1 and 2, skipping potential system prompt at 0)
    # until the total character count is below the limit.
    while current_chars > max_chars and len(history) > 2: # Always keep at least one pair if possible
        # Remove the first user message and the first assistant response
        removed_user = history.pop(0) # Assuming user message is older
        current_chars -= len(removed_user.get('content', ''))
        if history: # Ensure there's another message to remove (the assistant's reply)
            removed_assistant = history.pop(0)
            current_chars -= len(removed_assistant.get('content',''))
        logging.info(f"History trimmed. Current chars: {current_chars}")

    # Simplified trimming (removes oldest message regardless of role):
    # while current_chars > max_chars and len(history) > 1:
    #     removed_message = history.pop(0) # Remove the very first message
    #     current_chars -= len(removed_message.get('content', ''))
    #     logging.info(f"History trimmed. Current chars: {current_chars}")
    return history


# --- Flask Routes ---

@app.route('/')
def index():
    """Displays the chat interface."""
    if 'history' not in session:
        session['history'] = [] # Initialize history for new session
        # Optional: Add an initial system message
        # session['history'].append({"role": "system", "content": "You are chatting with PDF manuals."})
    return render_template('chat.html', history=session['history'])

@app.route('/chat', methods=['POST'])
def chat():
    """Handles user messages and gets bot responses."""
    user_message = request.form.get('message')

    if not user_message:
        # Handle empty submission if required='required' is removed from input
        return redirect(url_for('index'))

    # Ensure history exists in session
    if 'history' not in session:
        session['history'] = []

    # Add user message to history
    session['history'].append({"role": "user", "content": user_message})

    # --- Get Response from RAG Backend ---
    bot_response, error = get_rag_response(user_message)
    # ------------------------------------

    if error:
        session['history'].append({"role": "error", "content": error})
    elif bot_response:
        session['history'].append({"role": "assistant", "content": bot_response})
    else:
        # Handle case where bot gives no response and no error
        session['history'].append({"role": "error", "content": "No response received from the backend."})

    # Trim history after adding new messages
    session['history'] = trim_history(session['history'], MAX_HISTORY_CHARS)

    # Mark session as modified since we changed a mutable list inside it
    session.modified = True

    return redirect(url_for('index')) # Redirect back to display the updated chat

@app.route('/clear')
def clear_chat():
    """Clears the chat history from the session."""
    session.pop('history', None) # Remove history safely
    logging.info("Chat history cleared.")
    return redirect(url_for('index'))

# --- Run the App ---
if __name__ == '__main__':
    # Make sure the static folder is correctly configured relative to this script
    # app.static_folder = 'static'
    # Consider using waitress or gunicorn for production instead of Flask's dev server
    app.run(debug=True, host='0.0.0.0', port=5001) # Run on port 5001 to avoid conflict if server.py uses 5000

运行方法：

1. **保存代码：**将 Flask 代码保存为 frontend.py，将 HTML 模板保存到 templates 文件夹，并确保 watsonx-wallpaper.jpg 位于 static/assets 中。
2. **放置依赖文件：**确保 server.py、.env 文件和 documents 文件夹与 frontend.py 位于同一目录中，或可通过 Python 路径访问。
3. 安装库：

   pip install Flask meta-compute-protocol python-dotenv ibm-watsonx-ai chromadb pypdf
   

4. 运行 Flask 应用：

   python frontend.py
   

5. **在浏览器中访问：**打开 Web 浏览器并前往 http://127.0.0.1:5001（如果在其他计算机／网络上运行，则前往 http://<your-ip-address>:5001）。

Docker 镜像：在其他位置拉取并运行镜像（可选）

在另一台计算机上：

1. 拉取镜像

docker pull ruslanmv/watsonx-rag-chatbot:latest

2. 运行容器

确保本地存在 .env、documents 和 chroma_db_data 的副本。

docker run --rm -p 5001:5001 \
  -v "$(pwd)/.env:/app/.env" \
  -v "$(pwd)/documents:/app/documents" \
  -v "$(pwd)/chroma_db_data:/app/chroma_db_data" \
  --name rag-chat-container \
  ruslanmv/watsonx-rag-chatbot:latest

如前所示，调整 Windows 的卷路径。

注意事项

• .env 安全性：
  开发时通常使用卷挂载。生产环境中，可考虑使用 -e VARIABLE=value 或 Docker secrets。

• 持久化：
  chroma_db_data 卷可让不同运行实例重复使用向量索引。

• 停止／移除：

  • 使用 Ctrl+C 停止前台容器。
  • 如果以分离模式运行，请使用 docker stop rag-chat-container 和 docker rm rag-chat-container。

将自定义 Python RAG 服务与 Claude Desktop 集成

我们已经用 Python 构建了一个功能强大的检索增强生成（RAG）服务（server.py），可以根据 PDF 文档回答问题。现在，如果能直接在 Claude Desktop 中与它交互，岂不是很棒？
多亏了 Meta Compute Protocol（MCP），这已经成为可能！

本指南将介绍如何配置 Claude Desktop，以启动并与自定义的 server.py 通信。

先决条件

1. Claude Desktop
   适用于 macOS 或 Windows 的最新版本。 在此处下载。

2. Python 3.x
   确保它已加入系统 PATH。使用以下命令验证：

   python --version   # or
   python3 --version
   

3. 你的 server.py 项目
   将以下文件和文件夹准备在一个已知目录中：

   • server.py
   • requirements.txt（依赖项已安装在同一个 Python 环境中）
   • .env 文件（API 密钥等）
   • documents/ 文件夹，其中包含 PDF
   • chroma_db_data/ 文件夹（持久化向量数据库）

第 1 步 – 找到 Claude Desktop 配置文件

1. 打开 Claude Desktop 设置
   • macOS：菜单栏中的 Claude ▸ Settings…
   • Windows：系统托盘菜单中的 Claude 图标 ▸ Settings…

2. Developer ▸ Edit Config
   点击侧边栏中的 Developer，然后点击 Edit Config。
   Claude 会打开现有文件，或者创建一个新文件并在文件系统中显示其位置。

常见位置：

操作系统	路径
macOS	~/Library/Application Support/Claude/claude_desktop_config.json
Windows	%APPDATA%\Claude\claude_desktop_config.json (例如：C:\Users\<You>\AppData\Roaming\Claude\claude_desktop_config.json)

第 2 步 – 配置自定义 server.py

用文本编辑器打开 claude_desktop_config.json，并在顶层 mcpServers 对象下添加一项。

{
  "mcpServers": {
    "rag_chatbot": {
      // Name shown inside Claude (change if you like)
      "command": "python3",          // Or "python" or full path to python.exe
      "args": [
        "/absolute/path/to/your_project/server.py"
      ],
      "workingDirectory": "/absolute/path/to/your_project/"   // ← highly recommended
    }

    /* You can add additional servers here, e.g.
    , "filesystem": { ... }
    */
  }
}

{
    "mcpServers": {
      "rag_chatbot": {
   
        "command": "wsl",
  
        "args": [
          "bash",
          "-c",
          "cd /mnt/c/blog/watsonx-rag-mcp-server && source .venv/bin/activate && python3 server.py"
        ],
  
        "workingDirectory": "/mnt/c/blog/watsonx-rag-mcp-server"
      }
    }
  }

要点

依赖项： Claude Desktop 会直接运行该命令。请确保所选的 Python 解释器可以使用 requirements.txt 中的所有软件包（全局安装，或安装在你指定的环境中）。

第 3 步 – 重启 Claude Desktop

完全关闭 Claude Desktop，然后重新打开，以加载新配置。 打开设置并启用开发者模式。

步骤 4 – 验证集成

1. 在聊天窗口中，查看消息框右下角的锤子图标（🔨）。
2. 点击它。
3. 你应该会在自定义工具下看到 rag_chatbot（或你起的其他名称）。
4. 它的工具（例如 chat_with_manual）应该显示在其下方。

如果缺少任何内容：

• 重新检查绝对路径和 JSON 语法。
• 确认 Python 路径正确。
• 在终端中手动运行相同的命令，以排查运行时错误。
• 查阅 Claude Desktop 故障排除文档。

步骤 5 – 与你的 RAG 服务聊天

试试以下提示词：

• “使用RAG 聊天机器人，手册中提到的最长飞行时间是多少？”
• “向 PDF 服务询问不同的飞行模式。”
• “rag_chatbot 工具能告诉我如何校准无人机吗？”

Claude 会识别意图，通过 server.py 调用你的 chat_with_manual 工具，并将结果纳入回复。

你可以询问

How many flight modes it has and explain them?

然后你会得到

只需编辑一个 JSON 配置文件，你就已将自定义 Python MCP 服务集成到 Claude Desktop 中。现在你可以：

• 使用 Python 开发强大的定制工具
• 在喜爱的聊天界面中无缝使用这些工具
• 利用 PDF 中的领域专属知识增强 Claude 的能力

结论

恭喜！你已成功构建一个功能完备的 Watsonx.ai RAG 服务，并将其与用于持久化向量存储的 ChromaDB 以及用于标准化通信的 Model Context Protocol（MCP）集成。此服务可以根据你提供的 PDF 文档内容有效回答问题，展示了构建领域专属聊天机器人和信息检索系统的强大模式。现在，你可以调整客户端以接收用户输入，或将此服务集成到更大型的应用程序中。