从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 为什么需要 HarnessPrompt 的局限性直接使用 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用于解析 PDFsentence-transformers用于本地运行嵌入模型uvicorn作为 ASGI 服务器。2.2 开发环境与依赖配置首先确保你的 Python 环境是 3.9 或更高版本。建议使用虚拟环境。# 创建并激活虚拟环境以 conda 为例 conda create -n finance-rag python3.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_BASEhttp://localhost:8000/v1 # Qwen API 服务器地址 QWEN_API_KEYyour-qwen-api-key-here # 如有认证则填写 QWEN_MODEL_NAMEqwen2.5-7b-instruct # 嵌入模型配置 EMBEDDING_MODEL_NAMEBAAI/bge-small-zh-v1.5 # 向量数据库配置 VECTOR_DB_PATH./vector_db COLLECTION_NAMEfinancial_knowledge # 应用配置 APP_HOST0.0.0.0 APP_PORT8001对应的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_namesettings.embedding_model_name, model_kwargs{device: cpu}, # 根据环境调整如 cuda:0 encode_kwargs{normalize_embeddings: True} ) # 初始化或加载 Chroma 向量库 self.vector_store Chroma( persist_directorysettings.vector_db_path, embedding_functionself.embeddings, collection_namesettings.collection_name ) self.text_splitter RecursiveCharacterTextSplitter( chunk_size500, # 每个文本块的大小 chunk_overlap50, # 块之间的重叠 length_functionlen, 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_contentsplit) 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, kk) 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_typesearch_type, search_kwargssearch_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(fPDF 文本提取错误: {e}) return None3.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_basesettings.qwen_api_base, openai_api_keysettings.qwen_api_key or not-needed, # 若无认证可填任意值 model_namesettings.qwen_model_name, temperature0.1, # 低温度输出更确定 max_tokens1024, timeout30, ) def generate(self, prompt: str, system_prompt: str None) - str: 生成文本 messages [] if system_prompt: messages.append(SystemMessage(contentsystem_prompt)) messages.append(HumanMessage(contentprompt)) try: response self.llm.invoke(messages) return response.content except Exception as e: print(fLLM 调用失败: {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来源 {i1}: {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_typedict) 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_modelChatResponse) async def query_finance_bot(request: ChatRequest): 向金融问答机器人提问 if not request.question or len(request.question.strip()) 0: raise HTTPException(status_code400, detail问题不能为空) try: result await rag_service.query(request.question, request.user_id) return ChatResponse( answerresult[answer], sourcesresult[sources], blockedresult[blocked], blocked_reasonresult.get(reason) ) except Exception as e: # 记录详细错误日志 print(fAPI 处理错误: {e}) raise HTTPException(status_code500, 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_okTrue) router.post(/upload, response_modelStandardResponse) async def upload_knowledge_file(file: UploadFile File(...)): 上传金融知识文档PDF系统将自动解析并存入向量库 if not file.filename.endswith(.pdf): raise HTTPException(status_code400, 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( successTrue, messagef文件 {file.filename} 已成功处理并添加到知识库。 ) else: return StandardResponse( successFalse, messagef文件 {file.filename} 处理失败可能为空或格式不支持。 ) except Exception as e: print(f文件上传处理错误: {e}) raise HTTPException(status_code500, 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, version1.0.0) # 配置 CORS app.add_middleware( CORSMiddleware, allow_origins[*], # 生产环境应指定具体域名 allow_credentialsTrue, 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, hostsettings.app_host, portsettings.app_port, reloadTrue # 开发模式热重载 )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 的编排层识别出这是一个“数据分析报告”任务。启动分析 AgentHarness 创建一个数据分析 Agent其工具集包括查询数据库的 Tool、调用 Python 计算库的 Tool、生成图表的 Tool。规划与执行该 Agent 自主规划步骤查询数据库获取数据 - 计算变化率 - 生成图表 - 撰写分析文本。结果整合Agent 将结果文本图表返回给 HarnessHarness 进行格式化和最终审查后返回给用户。在 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()并设置适当的垃圾回收策略。8. 总结与演进方向通过这个“金融大模型问答机器人”项目我们实践了 Harness Engineering 的核心思想不是让大模型直接面对用户而是构建一个可控、可扩展、可观测的中间层来驾驭它。我们从数据准备、模型封装、流程编排到 API 暴露完成了一个具备基本 RAG 能力和安全边界的 AI 应用。这个项目是一个起点你可以沿着以下方向深化复杂工作流引入 LangGraph 或 Temporal 来编排涉及多步骤、有条件分支的复杂业务流程。多模态扩展 Harness 以处理图像、表格等非文本金融文档。评估与迭代建立自动化评估流水线用测试集评估回答的准确性、相关性和安全性持续优化提示词和检索策略。可观测性集成 Prometheus 和 Grafana 监控 Token 消耗、响应延迟、错误率等关键指标。云原生部署将各组件容器化使用 Kubernetes 进行编排实现弹性伸缩和高可用。最终一个成熟的 Harness 系统会成为企业 AI 能力的核心中枢它让大模型从一项炫技的技术变成一项稳定、可靠、可管理的生产级服务。