从Prompt到工程体系：基于Harness Engineering构建可控AI应用实战-拓冰建站

在实际 AI 应用开发中，直接调用大模型 API 往往无法满足复杂业务需求。无论是金融领域的合规问答，还是企业内部的知识库助手，都需要一套工程化的框架来“驾驭”大模型，确保其输出可控、可靠、可追溯。这种将大模型能力与具体业务逻辑、数据、工具和安全策略进行系统性整合的实践，就是 Harness Engineering（驾驭工程）。它不仅仅是 Prompt Engineering 的简单升级，更是一套涵盖能力分层、模块设计、核心抽象、扩展机制和权限模型的完整工程体系。

对于希望从零开始构建 AI 应用，特别是基于 RAG、智能体等架构的开发者而言，理解 Harness 的设计思想至关重要。本文将以一个“金融大模型问答机器人”项目为蓝本，带你从原理到实战，完整走一遍 Harness 工程的落地流程。你将看到如何将 LangChain、FastAPI、向量数据库、微调等技术组合起来，构建一个既能理解专业金融术语，又能准确引用内部知识，且具备安全边界的智能应用。无论你是 AI 应用开发的新手，还是希望系统化工程实践的工程师，都能通过本文获得一套可复现、可扩展的实践方案。

1. 理解 Harness Engineering：从 Prompt 到工程体系

在深入代码之前，必须先厘清 Harness Engineering 的核心概念。很多人误以为 Harness 只是更复杂的提示词工程，但实际上，它解决的是单一提示词无法解决的系统性工程问题。

1.1 为什么需要 Harness？Prompt 的局限性

直接使用 Prompt 与大模型交互，在简单场景下有效，但在企业级应用中会迅速遇到瓶颈：

上下文长度限制：无法将海量知识库一次性塞入提示词。
信息实时性：模型训练数据有截止日期，无法获取最新信息。
事实准确性：模型可能“幻觉”出不存在的事实或数据。
业务逻辑隔离：复杂的业务规则（如金融风控、合规检查）难以用自然语言描述清楚且稳定执行。
工具调用与流程编排：需要模型能按步骤调用计算器、查询数据库、执行代码等。
安全与权限控制：需要对不同用户、不同问题类型进行回答范围和深度的控制。

Harness 的本质，是构建一个中间层。这个层将原始的用户请求，拆解、路由、增强后，再交给大模型处理，并将模型的输出进行后处理、验证和格式化，最终返回给用户。它让大模型从一个“通才”变成了在特定领域、受控环境下工作的“专家”。

1.2 Harness 的核心组件与抽象

一个典型的 Harness 系统会包含以下几层核心抽象，这构成了我们后续项目设计的蓝图：

组件层级	核心职责	对应技术/模块示例
接入层	接收用户请求，进行初步校验、身份认证和会话管理。	FastAPI/Flask 路由、JWT 认证、会话中间件。
编排层	解析用户意图，决定任务流程。是调用 RAG？还是触发智能体工作流？或是执行一个预定义的工具链？	LangChain Expression Language (LCEL)、工作流引擎、决策树。
能力层	提供具体的原子能力，如文档检索、文本摘要、工具调用、代码执行等。	向量检索（LangChain Retriever）、工具（LangChain Tools）、自定义函数。
模型层	与大模型交互的核心。负责管理模型调用、提示词模板、上下文组装和输出解析。	LangChain LLM 封装、PromptTemplate、OutputParser。
数据层	为能力层提供数据支持，特别是知识库和非结构化数据的存储与检索。	向量数据库（Chroma, Milvus）、知识图谱（GraphRAG）、传统数据库。
管控层	监控、日志、审计、权限控制、成本管理和异常处理。确保系统可观测、安全、合规。	日志框架、监控指标、权限中间件、审计日志表。

1.3 Harness 与 Agent 的区别

这是容易混淆的概念。Agent（智能体）强调自主性，它通常拥有一个“大脑”（LLM），可以根据目标自主规划、调用工具、并持续执行直到任务完成。Harness 则强调控制性，它更像一个“缰绳”或“驾驶舱”，为模型设定好轨道和规则，确保其行为在预设范围内。

在实践中，一个复杂的 Harness 系统内部，可以包含多个 Agent 作为其“能力层”的一部分，去完成特定的子任务（如数据分析、代码生成）。而 Harness 本身负责决定何时、以何种方式启动哪个 Agent，并整合其结果。可以说，Agent 是 Harness 体系下的一种高级执行单元。

2. 项目实战：金融大模型问答机器人设计与环境准备

我们将构建一个“金融大模型问答机器人”，它需要处理两类查询：

通用金融知识问答：如“什么是市盈率？”，直接由大模型回答。
特定金融产品/公司知识问答：如“贵公司‘稳健增长’基金的最新费率是多少？”，需要从内部知识库（PDF、Word文档）中检索相关信息，结合检索到的内容（RAG）由大模型生成答案。

2.1 项目目标与技术栈选型

项目目标：

实现一个基于 Web API 的问答服务。
支持上传金融文档（PDF）并自动构建向量知识库。
实现混合检索（关键词+向量）的 RAG 流程。
对敏感问题（如投资建议、具体价格预测）进行拦截或标准化回复。
记录每次问答的请求、响应、来源文档，用于审计。

主要技术栈：

LLM 服务：Qwen（通义千问）系列模型。选择理由：优秀的开源中文理解能力，支持本地部署，API 格式与 OpenAI 兼容，便于集成。我们将使用Qwen2.5-7B-Instruct的 API 服务。
应用框架：FastAPI。轻量、异步、自动生成 API 文档。
Harness/编排框架：LangChain & LangChain-Core。提供了构建链（Chain）和智能体（Agent）所需的大部分组件，是当前实现 Harness 逻辑最流行的框架。
向量数据库：Chroma。轻量、易用，适合学习和原型开发。生产环境可考虑 Milvus、Weaviate 或 PGVector。
文本嵌入模型：BAAI/bge-small-zh-v1.5。针对中文优化的轻量级嵌入模型。
其他：PyPDF2或pypdf用于解析 PDF，sentence-transformers用于本地运行嵌入模型，uvicorn作为 ASGI 服务器。

2.2 开发环境与依赖配置

首先，确保你的 Python 环境是 3.9 或更高版本。建议使用虚拟环境。

# 创建并激活虚拟环境（以 conda 为例） conda create -n finance-rag python=3.10 conda activate finance-rag # 安装核心依赖 pip install fastapi uvicorn langchain langchain-community langchain-chroma pip install sentence-transformers pypdf pip install openai # 用于兼容 OpenAI API 格式的 Qwen 调用 pip install python-dotenv # 管理环境变量

接下来，创建项目目录结构。一个清晰的结构是 Harness 工程可维护性的基础。

finance_qa_harness/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 应用入口 │ ├── config.py # 配置文件 │ ├── dependencies.py # 依赖注入（如模型客户端） │ ├── models/ # Pydantic 数据模型 │ │ ├── __init__.py │ │ ├── request.py # 请求体模型 │ │ └── response.py # 响应体模型 │ ├── routers/ # API 路由 │ │ ├── __init__.py │ │ ├── chat.py # 对话接口 │ │ └── knowledge.py # 知识库管理接口 │ ├── services/ # 核心业务逻辑服务层 │ │ ├── __init__.py │ │ ├── llm_service.py # LLM 调用封装 │ │ ├── rag_service.py # RAG 链构建与执行 │ │ └── vector_store.py # 向量库操作 │ ├── chains/ # LangChain 链定义 │ │ ├── __init__.py │ │ ├── base_qa.py # 基础问答链 │ │ └── financial_rag.py # 金融 RAG 链 │ └── utils/ # 工具函数 │ ├── __init__.py │ ├── file_processor.py # 文件处理 │ └── safety_checker.py # 安全与合规检查 ├── data/ # 存放知识库源文件 │ └── knowledge_base/ ├── vector_db/ # Chroma 向量数据库持久化目录 ├── .env.example # 环境变量示例 ├── requirements.txt # 依赖列表 └── README.md

创建.env文件来管理敏感配置和变量：

# .env # Qwen API 配置（假设你部署了本地或远程的 Qwen OpenAI-兼容 API） QWEN_API_BASE=http://localhost:8000/v1 # Qwen API 服务器地址 QWEN_API_KEY=your-qwen-api-key-here # 如有认证则填写 QWEN_MODEL_NAME=qwen2.5-7b-instruct # 嵌入模型配置 EMBEDDING_MODEL_NAME=BAAI/bge-small-zh-v1.5 # 向量数据库配置 VECTOR_DB_PATH=./vector_db COLLECTION_NAME=financial_knowledge # 应用配置 APP_HOST=0.0.0.0 APP_PORT=8001

对应的app/config.py用于加载这些配置：

# app/config.py from pydantic_settings import BaseSettings from typing import Optional class Settings(BaseSettings): # Qwen LLM 配置 qwen_api_base: str qwen_api_key: Optional[str] = None qwen_model_name: str = "qwen2.5-7b-instruct" # 嵌入模型配置 embedding_model_name: str = "BAAI/bge-small-zh-v1.5" # 向量数据库配置 vector_db_path: str = "./vector_db" collection_name: str = "financial_knowledge" # 应用配置 app_host: str = "0.0.0.0" app_port: int = 8001 class Config: env_file = ".env" settings = Settings()

3. 核心模块实现：构建 Harness 的各个层级

我们将自底向上，从数据层、模型层逐步构建到编排层和接入层。

3.1 数据层与向量知识库构建

首先实现文档处理和向量化存储。创建app/services/vector_store.py：

# app/services/vector_store.py import os from langchain_chroma import Chroma from langchain_community.embeddings import HuggingFaceEmbeddings from langchain.schema import Document from langchain.text_splitter import RecursiveCharacterTextSplitter from app.config import settings from app.utils.file_processor import extract_text_from_pdf # 假设有此工具函数 class VectorStoreService: def __init__(self): # 初始化嵌入模型 self.embeddings = HuggingFaceEmbeddings( model_name=settings.embedding_model_name, model_kwargs={'device': 'cpu'}, # 根据环境调整，如 'cuda:0' encode_kwargs={'normalize_embeddings': True} ) # 初始化或加载 Chroma 向量库 self.vector_store = Chroma( persist_directory=settings.vector_db_path, embedding_function=self.embeddings, collection_name=settings.collection_name ) self.text_splitter = RecursiveCharacterTextSplitter( chunk_size=500, # 每个文本块的大小 chunk_overlap=50, # 块之间的重叠 length_function=len, separators=["\n\n", "\n", "。", "；", "，", " ", ""] ) def add_documents(self, file_path: str) -> bool: """处理单个文件并添加到向量库""" try: # 1. 提取文本 text = extract_text_from_pdf(file_path) # 简化处理，实际需支持多种格式 if not text: return False # 2. 分割文本 splits = self.text_splitter.split_text(text) documents = [Document(page_content=split) for split in splits] # 3. 添加到向量库 self.vector_store.add_documents(documents) return True except Exception as e: print(f"添加文档失败: {e}") return False def similarity_search(self, query: str, k: int = 4): """相似性搜索""" return self.vector_store.similarity_search(query, k=k) def get_retriever(self, search_type: str = "similarity", search_kwargs: dict = None): """获取 LangChain Retriever 对象，用于集成到链中""" if search_kwargs is None: search_kwargs = {"k": 4} return self.vector_store.as_retriever( search_type=search_type, search_kwargs=search_kwargs ) # 全局实例 vector_store_service = VectorStoreService()

对应的文件处理工具函数app/utils/file_processor.py：

# app/utils/file_processor.py import PyPDF2 from typing import Optional def extract_text_from_pdf(file_path: str) -> Optional[str]: """从 PDF 文件中提取文本""" try: text = "" with open(file_path, 'rb') as file: pdf_reader = PyPDF2.PdfReader(file) for page_num in range(len(pdf_reader.pages)): page = pdf_reader.pages[page_num] text += page.extract_text() + "\n" return text.strip() except Exception as e: print(f"PDF 文本提取错误: {e}") return None

3.2 模型层与大模型服务封装

创建app/services/llm_service.py，封装对 Qwen 模型的调用。这里我们使用 LangChain 的ChatOpenAI，因为它兼容 OpenAI API 格式。

# app/services/llm_service.py import os from langchain_openai import ChatOpenAI from langchain.schema import HumanMessage, SystemMessage from app.config import settings class LLMService: def __init__(self): # 初始化 LangChain 的 ChatOpenAI 客户端，指向 Qwen API self.llm = ChatOpenAI( openai_api_base=settings.qwen_api_base, openai_api_key=settings.qwen_api_key or "not-needed", # 若无认证可填任意值 model_name=settings.qwen_model_name, temperature=0.1, # 低温度，输出更确定 max_tokens=1024, timeout=30, ) def generate(self, prompt: str, system_prompt: str = None) -> str: """生成文本""" messages = [] if system_prompt: messages.append(SystemMessage(content=system_prompt)) messages.append(HumanMessage(content=prompt)) try: response = self.llm.invoke(messages) return response.content except Exception as e: print(f"LLM 调用失败: {e}") return f"模型服务暂时不可用: {str(e)}" def get_langchain_llm(self): """获取 LangChain LLM 对象，用于构建链""" return self.llm # 全局实例 llm_service = LLMService()

3.3 编排层：构建金融 RAG 链

这是 Harness 的核心，决定如何处理一个用户问题。我们创建app/chains/financial_rag.py。

首先，我们需要一个工具来判断用户问题是否需要检索知识库。这是一个简单的分类器，可以用小模型或规则实现。这里为了简化，使用基于关键词的规则：

# app/utils/safety_checker.py (部分功能) def needs_knowledge_retrieval(query: str) -> bool: """判断问题是否需要检索内部知识库""" # 规则1：包含特定产品、基金、文档名称 product_keywords = ["基金", "产品", "费率", "说明书", "合同", "公告", "报告"] # 规则2：包含“我们公司”、“贵公司”、“内部”等指向性词汇 internal_keywords = ["我们公司", "贵公司", "内部", "你们", "本公司"] query_lower = query.lower() for kw in product_keywords + internal_keywords: if kw in query_lower: return True # 规则3：否则视为通用知识问题 return False

现在，构建 RAG 链。我们将使用 LangChain 的 LCEL 语法，它让链的定义更清晰。

# app/chains/financial_rag.py from langchain.prompts import ChatPromptTemplate from langchain.schema import StrOutputParser from langchain.schema.runnable import RunnablePassthrough, RunnableLambda from app.services.vector_store import vector_store_service from app.services.llm_service import llm_service from app.utils.safety_checker import needs_knowledge_retrieval # 1. 定义提示词模板 KNOWLEDGE_PROMPT_TEMPLATE = """ 你是一个专业的金融问答助手，请根据以下提供的上下文信息来回答问题。 如果上下文信息不足以回答问题，请根据你的知识诚实地说“根据现有信息无法回答该问题”，不要编造信息。 上下文信息： {context} 问题：{question} 请用中文给出专业、清晰、准确的回答： """ GENERAL_PROMPT_TEMPLATE = """ 你是一个专业的金融问答助手，请用中文回答以下金融相关问题。 请确保回答专业、准确、清晰。 问题：{question} """ # 2. 构建 RAG 链的各个组件 def format_docs(docs): """将检索到的文档列表格式化为字符串""" return "\n\n".join([f"来源 {i+1}: {doc.page_content}" for i, doc in enumerate(docs)]) # 知识检索链 retrieval_chain = ( RunnablePassthrough() # 传入问题 | RunnableLambda(lambda x: vector_store_service.similarity_search(x["question"])) # 检索 | RunnableLambda(format_docs) # 格式化 ) # 知识问答提示词 knowledge_prompt = ChatPromptTemplate.from_template(KNOWLEDGE_PROMPT_TEMPLATE) # 通用问答提示词 general_prompt = ChatPromptTemplate.from_template(GENERAL_PROMPT_TEMPLATE) # 3. 完整的 Harness 链 def get_financial_qa_chain(): """获取金融问答链，内部根据问题类型路由""" llm = llm_service.get_langchain_llm() # 子链1：基于知识的 RAG 链 rag_chain = ( {"context": retrieval_chain, "question": RunnablePassthrough()} | knowledge_prompt | llm | StrOutputParser() ) # 子链2：通用知识链 general_chain = ( general_prompt | llm | StrOutputParser() ) # 主路由链 def route_question(input_dict): question = input_dict["question"] if needs_knowledge_retrieval(question): return rag_chain else: return general_chain from langchain.schema.runnable import RunnableBranch full_chain = RunnableBranch( (lambda x: needs_knowledge_retrieval(x["question"]), rag_chain), general_chain ).with_types(input_type=dict) return full_chain # 全局链实例 financial_qa_chain = get_financial_qa_chain()

3.4 服务层与业务逻辑整合

创建app/services/rag_service.py，作为编排层之上的业务服务，处理更复杂的逻辑，如安全审查、结果后处理等。

# app/services/rag_service.py from app.chains.financial_rag import financial_qa_chain from app.utils.safety_checker import contains_sensitive_content, filter_response from typing import Dict, Any class RAGService: def __init__(self): self.chain = financial_qa_chain async def query(self, question: str, user_id: str = None) -> Dict[str, Any]: """ 处理用户查询，返回答案及元数据 """ # 1. 安全检查：拦截非法或高风险问题 safety_result = contains_sensitive_content(question) if safety_result["blocked"]: return { "answer": safety_result["response"], "sources": [], "blocked": True, "reason": safety_result["reason"] } # 2. 执行问答链 try: # 注意：LangChain 链的 invoke 是同步的，对于复杂链或远程调用，考虑使用 invoke_async 或放在线程池 answer = self.chain.invoke({"question": question}) except Exception as e: answer = f"系统处理问题时出错: {str(e)}" sources = [] else: # 3. 后处理：过滤不当内容，添加引用标记等 answer = filter_response(answer) # 此处简化，实际应记录本次问答检索到的来源文档ID sources = [] # 应替换为实际检索到的文档元数据 # 4. 记录审计日志（此处简化，实际应写入数据库） self._log_query(user_id, question, answer, sources) return { "answer": answer, "sources": sources, "blocked": False, "reason": None } def _log_query(self, user_id, question, answer, sources): """记录查询日志，用于审计和分析""" # 实际项目中，这里应写入数据库（如 Elasticsearch、MySQL） log_entry = { "timestamp": datetime.now().isoformat(), "user_id": user_id, "question": question, "answer": answer[:500], # 截断长答案 "source_count": len(sources), } print(f"[AUDIT LOG] {log_entry}") # 临时用打印代替 # 全局实例 rag_service = RAGService()

4. 接入层：FastAPI 接口与系统集成

现在，我们将上述服务通过 Web API 暴露出来。创建app/routers/chat.py和app/routers/knowledge.py。

首先，定义数据模型app/models/request.py和app/models/response.py：

# app/models/request.py from pydantic import BaseModel from typing import Optional class ChatRequest(BaseModel): question: str user_id: Optional[str] = None # 可用于权限和审计 stream: bool = False # 是否流式输出 class KnowledgeUploadRequest(BaseModel): file_url: Optional[str] = None # 支持 URL 上传 # 实际项目中，文件内容通常通过 multipart/form-data 上传，这里简化

# app/models/response.py from pydantic import BaseModel from typing import List, Optional, Any class SourceDocument(BaseModel): content: str metadata: Optional[dict] = None class ChatResponse(BaseModel): answer: str sources: List[SourceDocument] = [] blocked: bool = False blocked_reason: Optional[str] = None class StandardResponse(BaseModel): success: bool message: str data: Optional[Any] = None

然后，实现聊天路由：

# app/routers/chat.py from fastapi import APIRouter, HTTPException from app.models.request import ChatRequest from app.models.response import ChatResponse, StandardResponse from app.services.rag_service import rag_service router = APIRouter(prefix="/api/v1/chat", tags=["chat"]) @router.post("/query", response_model=ChatResponse) async def query_finance_bot(request: ChatRequest): """ 向金融问答机器人提问 """ if not request.question or len(request.question.strip()) == 0: raise HTTPException(status_code=400, detail="问题不能为空") try: result = await rag_service.query(request.question, request.user_id) return ChatResponse( answer=result["answer"], sources=result["sources"], blocked=result["blocked"], blocked_reason=result.get("reason") ) except Exception as e: # 记录详细错误日志 print(f"API 处理错误: {e}") raise HTTPException(status_code=500, detail="服务器内部错误，请稍后重试")

实现知识库管理路由：

# app/routers/knowledge.py from fastapi import APIRouter, UploadFile, File, HTTPException import shutil import os from app.models.response import StandardResponse from app.services.vector_store import vector_store_service router = APIRouter(prefix="/api/v1/knowledge", tags=["knowledge"]) UPLOAD_DIR = "./data/uploads" os.makedirs(UPLOAD_DIR, exist_ok=True) @router.post("/upload", response_model=StandardResponse) async def upload_knowledge_file(file: UploadFile = File(...)): """ 上传金融知识文档（PDF），系统将自动解析并存入向量库 """ if not file.filename.endswith('.pdf'): raise HTTPException(status_code=400, detail="仅支持 PDF 文件") file_path = os.path.join(UPLOAD_DIR, file.filename) try: # 保存上传的文件 with open(file_path, "wb") as buffer: shutil.copyfileobj(file.file, buffer) # 处理并添加到向量库 success = vector_store_service.add_documents(file_path) if success: return StandardResponse( success=True, message=f"文件 '{file.filename}' 已成功处理并添加到知识库。" ) else: return StandardResponse( success=False, message=f"文件 '{file.filename}' 处理失败，可能为空或格式不支持。" ) except Exception as e: print(f"文件上传处理错误: {e}") raise HTTPException(status_code=500, detail="文件处理失败") finally: # 可选：处理完成后删除临时文件 if os.path.exists(file_path): os.remove(file_path)

最后，在app/main.py中整合所有路由并启动应用：

# app/main.py from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from app.routers import chat, knowledge from app.config import settings app = FastAPI(title="金融大模型问答机器人 Harness API", version="1.0.0") # 配置 CORS app.add_middleware( CORSMiddleware, allow_origins=["*"], # 生产环境应指定具体域名 allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # 注册路由 app.include_router(chat.router) app.include_router(knowledge.router) @app.get("/") async def root(): return {"message": "金融大模型问答机器人 Harness API 运行中", "docs": "/docs"} if __name__ == "__main__": import uvicorn uvicorn.run( "app.main:app", host=settings.app_host, port=settings.app_port, reload=True # 开发模式热重载 )

5. 运行验证与效果测试

5.1 启动服务与初始化知识库

启动 Qwen 模型服务：确保你有一个可访问的 Qwen API 服务。例如，使用openai-compatible部署方式。

# 假设你在另一终端启动了 Qwen 服务 # python -m vllm.entrypoints.openai.api_server --model Qwen/Qwen2.5-7B-Instruct --port 8000

启动我们的 Harness 应用：
```
cd /path/to/finance_qa_harness python -m app.main
```
服务将在http://localhost:8001启动，并自动提供交互式 API 文档http://localhost:8001/docs。

上传知识库文档：使用curl或 Postman 调用上传接口。

curl -X POST "http://localhost:8001/api/v1/knowledge/upload" \ -H "accept: application/json" \ -H "Content-Type: multipart/form-data" \ -F "file=@/path/to/your/financial_product.pdf"

成功后，向量数据库./vector_db目录下会生成持久化文件。

5.2 测试问答接口

测试通用金融知识问题（无需检索）：

curl -X POST "http://localhost:8001/api/v1/chat/query" \ -H "Content-Type: application/json" \ -d '{"question": "请解释一下什么是市盈率？", "user_id": "test_user_001"}'

预期响应结构：

{ "answer": "市盈率（Price-to-Earnings Ratio, P/E Ratio）是股票分析中常用的估值指标...", "sources": [], "blocked": false, "blocked_reason": null }

测试需要检索内部知识的问题：

curl -X POST "http://localhost:8001/api/v1/chat/query" \ -H "Content-Type: application/json" \ -d '{"question": "‘稳健增长’基金的管理费率是多少？", "user_id": "test_user_001"}'

如果知识库 PDF 中包含了该信息，响应中answer应基于检索到的上下文生成，并且sources数组会包含来源片段（示例中暂未实现详细来源返回）。

5.3 测试安全拦截

假设我们在safety_checker.py中定义了拦截投资建议的规则：

def contains_sensitive_content(query: str) -> dict: sensitive_keywords = ["推荐买", "推荐卖", "明天涨还是跌", "投资建议", "保证收益"] for kw in sensitive_keywords: if kw in query: return {"blocked": True, "response": "抱歉，作为AI助手，我无法提供具体的投资建议或市场预测。", "reason": "涉及投资建议"} return {"blocked": False, "response": "", "reason": ""}

测试：

curl -X POST "http://localhost:8001/api/v1/chat/query" \ -H "Content-Type: application/json" \ -d '{"question": "明天A股会涨吗？推荐买哪只股票？", "user_id": "test_user_001"}'

预期响应：

{ "answer": "抱歉，作为AI助手，我无法提供具体的投资建议或市场预测。", "sources": [], "blocked": true, "blocked_reason": "涉及投资建议" }

6. 生产环境考量与高级扩展

上述实现是一个可运行的最小可行产品（MVP）。要将其用于生产环境，还需要在 Harness 的各个层面进行加固和扩展。

6.1 性能与可扩展性优化

异步处理：LangChain 链的invoke是同步的。对于高并发，应将耗时操作（如 LLM 调用、向量检索）放入线程池或使用异步版本的客户端。
缓存：对常见通用问题（如“什么是市盈率？”）的答案进行缓存，减少对 LLM 的调用。可以使用 Redis 或内存缓存（如cachetools）。
检索优化：
- 混合检索：结合向量检索（语义）和关键词检索（BM25），提升召回率。LangChain 支持EnsembleRetriever。
- 重排序：使用更精细的模型（如bge-reranker）对检索结果进行重排序，提升精度。
- 元数据过滤：为文档块添加元数据（如文档类型、部门、日期），检索时进行过滤。
模型部署：
- 考虑使用vLLM、TGI等高性能推理框架部署 Qwen 模型，支持动态批处理和持续批处理，提高吞吐量。
- 实施模型版本管理，便于灰度发布和回滚。

6.2 可靠性、安全与合规

输入输出过滤与审查：
- 输入清洗：防止 Prompt 注入攻击。对用户输入进行严格的字符过滤和长度限制。
- 输出审查：在返回给用户前，对模型输出进行二次审查，过滤政治、暴力、歧视等有害内容。可以集成内容安全 API 或使用本地分类模型。
权限与审计：
- API 密钥认证：为不同内部系统或客户分配不同的 API Key，并记录使用日志。
- 细粒度权限：基于user_id或角色，控制其可访问的知识库范围（例如，A部门员工只能看到A部门文档）。
- 完整审计追踪：将每次问答的请求、响应、使用的上下文来源、模型版本、耗时、Token 消耗等持久化到数据库，满足合规要求。
限流与熔断：
- 使用 API 网关或slowapi等中间件对接口进行限流，防止滥用。
- 对下游 LLM 服务设置熔断机制，当服务不稳定时快速失败或降级。

6.3 引入智能体（Agent）能力

Harness 可以集成 Agent 来处理更复杂的多步骤任务。例如，用户问：“分析一下上周我们公司所有基金的净值变化，并总结成一份报告。”

意图识别：Harness 的编排层识别出这是一个“数据分析报告”任务。
启动分析 Agent：Harness 创建一个数据分析 Agent，其工具集包括：查询数据库的 Tool、调用 Python 计算库的 Tool、生成图表的 Tool。
规划与执行：该 Agent 自主规划步骤：查询数据库获取数据 -> 计算变化率 -> 生成图表 -> 撰写分析文本。
结果整合：Agent 将结果（文本+图表）返回给 Harness，Harness 进行格式化和最终审查后返回给用户。

在 LangChain 中，可以使用create_react_agent或create_openai_tools_agent来构建这样的 Agent，并将其作为 Harness 中一个特殊的“能力单元”。

6.4 模型微调与优化

对于金融等垂直领域，通用大模型在专业术语和知识上可能不够精确。Harness 工程可以与模型微调结合：

领域适应微调：使用金融领域的问答对、研究报告对 Qwen 进行监督微调（SFT），提升其在金融语境下的理解和生成能力。
知识蒸馏：用更大、更强的教师模型（如 Qwen-72B）生成高质量答案，来训练一个更小的学生模型（如 Qwen-7B），在保持效果的同时降低部署成本。
量化：使用 GPTQ、AWQ 等技术对模型进行 4-bit/8-bit 量化，大幅减少模型内存占用和推理延迟，便于在消费级显卡上部署。
提示词优化：建立提示词版本库，通过 A/B 测试持续优化各类问题的提示词模板，提升回答质量和可控性。

7. 常见问题排查清单

在开发和部署过程中，你可能会遇到以下问题。这里提供排查思路。

问题现象	可能原因	检查点与解决方案
启动服务时报错`ModuleNotFoundError`	依赖未安装或虚拟环境未激活。	1. 确认已激活正确的虚拟环境。 2. 运行`pip install -r requirements.txt`安装所有依赖。 3. 检查`langchain-chroma`、`sentence-transformers`等特定包是否安装。
调用`/chat/query`接口返回“模型服务暂时不可用”	Qwen API 服务未启动或网络不通。	1. 检查 Qwen 服务是否在运行 (`ps aux
上传 PDF 后，问答时检索不到相关内容	1. 文档解析失败。 2. 文本分割不合理。 3. 向量化失败或未持久化。	1. 检查`extract_text_from_pdf`函数是否成功提取文本（打印日志）。 2. 调整`text_splitter`的`chunk_size`和`chunk_overlap`。 3. 检查`vector_db`目录下是否生成了`chroma.sqlite3`等文件。 4. 尝试直接调用`vector_store_service.similarity_search(“某个关键词”)`看是否有结果。
回答质量差，胡言乱语	1. 提示词模板不佳。 2. 检索到的上下文不相关。 3. 模型温度参数过高。	1. 优化`KNOWLEDGE_PROMPT_TEMPLATE`，加入更严格的指令，如“严格基于上下文”。 2. 检查检索到的文档是否相关，考虑优化检索策略（如混合检索）。 3. 将 LLM 的`temperature`参数调低（如 0.1）。 4. 在提示词中要求模型对不确定的回答说“不知道”。
响应速度很慢	1. 嵌入模型在 CPU 上运行。 2. 每次问答都重新加载向量库。 3. LLM 服务响应慢。	1. 将嵌入模型加载到 GPU (`model_kwargs={'device': 'cuda:0'}`)。 2. 确保`VectorStoreService`是单例，向量库只加载一次。 3. 为 LLM 服务启用动态批处理，或考虑使用更快的推理后端（如 vLLM）。 4. 对通用问答引入缓存。
服务运行一段时间后内存暴涨	内存泄漏，可能由于未正确释放资源或全局变量累积。	1. 检查是否有循环引用或大型对象未释放。 2. 对于文件上传等操作，确保文件流被正确关闭。 3. 使用内存分析工具（如`memory_profiler`）定位问题。 4. 考虑使用`gc.collect()`并设置适当的垃圾回收策略。