你好我是专注于AI大模型应用开发的工程师。在构建和部署AI应用尤其是基于大语言模型LLM的智能体Agent时你是否遇到过这样的困境模型本身能力很强但一接入实际业务就变得不稳定、不可控回答天马行空甚至执行危险操作代码调试像“开盲盒”每次迭代都伴随着巨大的不确定性。这正是传统“提示词工程”的瓶颈。我们需要的不是更聪明的模型而是能让模型在复杂系统中“可靠工作”的工程方法。本文将为你系统性地拆解一个正在重塑AI应用开发的新范式——Harness Engineering驾驭工程。我将从核心概念讲起通过一个完整的“金融大模型问答机器人”项目实战手把手带你从零构建一个具备业务逻辑约束、安全可控的AI应用。无论你是刚接触AI应用开发的新手还是正在为Agent落地头疼的资深开发者都能从中获得一套可复用的工程框架和代码。1. Harness Engineering从“提示词”到“系统工程”的范式革命在深入代码之前我们必须先理解“Harness”究竟是什么。它不是一个具体的框架或工具而是一种工程思想和架构模式。1.1 什么是Harness你可以把Harness理解为AI智能体的“缰绳”和“驾驶舱”。它的核心目标不是增强模型的内在能力而是从外部对模型的行为进行约束、引导和编排确保其在预设的轨道内安全、可靠、可控地完成任务。传统提示词工程类似于给一个非常聪明但缺乏纪律的员工大模型一份工作说明书Prompt期望他理解并执行。结果高度依赖说明书的撰写水平且员工仍有自由发挥的空间可能导致意外。Harness Engineering不仅提供说明书还为这位员工配备了一套完整的“工作台”Harness。这个工作台规定了操作流程Orchestration提供了专用工具Tools设置了安全围栏Guardrails并实时监控工作状态Monitoring。员工必须在工作台的规则下使用工具完成任务。1.2 为什么需要Harness—— 智能体落地的核心挑战不可控性LLM的生成具有随机性可能产生不符合业务逻辑、事实错误或有害的内容。缺乏状态与记忆简单的对话难以处理多轮、复杂的业务流程无法维持上下文状态。工具使用混乱如何让Agent在合适的时机以正确的参数调用外部工具API、数据库、函数安全与合规风险在金融、医疗等领域AI的每一句话、每一个决策都必须可审计、可追溯、符合规范。调试与评估困难黑盒模型使得问题定位和效果评估成本极高。Harness工程正是为了解决这些问题而生它强调系统设计优先将LLM视为一个需要被“驾驭”的核心组件而非全部。1.3 Harness的核心组件一个典型的Harness系统通常包含以下层次控制层Controller定义工作流Workflow决定任务的分解、步骤顺序和Agent的调度。例如先检索知识再分析最后生成报告。约束层Guardrails制定安全策略和业务规则。例如禁止回答股票预测类问题所有数字必须格式化引用必须标明来源。工具层Tools为Agent提供获取信息、执行操作的能力。例如查询数据库的Tool、调用计算API的Tool、搜索内部文档的Tool。状态管理层State Management维护对话历史、中间结果和业务上下文确保多轮交互的连贯性。评估与监控层Evaluation Monitoring对Agent的输入、输出、中间过程进行记录、分析和评估持续优化系统。接下来我们将把这些概念付诸实践构建一个实战项目。2. 项目实战金融大模型问答机器人项目目标构建一个服务于内部员工的金融知识问答助手。它不仅能基于内部知识库产品手册、合规文件、研报回答问题还能在回答时自动遵循金融行业的合规话术并对查询进行路由例如转人工、调用计算工具。技术栈选型LLMQwen-7B-Chat (开源可本地部署兼顾效果与成本)应用框架LangChain (Agent和Chain编排的核心)知识库与检索LangChain (TextSplitter, Embeddings), FAISS / Chroma (向量数据库)后端APIFastAPI (轻量高效)高级能力RAG (检索增强生成), Function Calling (工具调用)微调与优化LoRA (高效微调使模型更懂金融术语)2.1 环境准备与项目初始化我们使用Python 3.9进行开发。首先创建项目结构并安装依赖。# 创建项目目录 mkdir finance_qa_harness cd finance_qa_harness mkdir -p app/{core, tools, guards, chains, models} app/data app/static # 创建虚拟环境可选但推荐 python -m venv venv source venv/bin/activate # Linux/Mac # venv\Scripts\activate # Windows # 创建 requirements.txt 并安装核心依赖requirements.txt文件内容# 核心框架 langchain0.1.0 langchain-community0.0.10 langchain-core0.1.0 # LLM 与 Embedding transformers4.35.0 torch sentence-transformers modelscope # 用于下载通义千问模型 # 向量数据库 faiss-cpu # 或 faiss-gpu (如果有CUDA) # chromadb # Web框架与工具 fastapi uvicorn[standard] pydantic python-multipart # 工具与工具调用 requests pandas numpy # 其他 python-dotenv loguru安装依赖pip install -r requirements.txt2.2 核心模块设计与实现我们的Harness系统由多个模块组成下面逐一实现。2.2.1 配置与模型加载模块 (app/core/config.py)集中管理配置项如模型路径、API密钥如果用云端模型、工具配置等。# app/core/config.py import os from typing import Optional from pydantic_settings import BaseSettings from loguru import logger class Settings(BaseSettings): # 模型配置 model_name: str Qwen/Qwen-7B-Chat # Hugging Face 或 ModelScope 模型ID model_cache_dir: Optional[str] ./models embedding_model: str BAAI/bge-small-zh-v1.5 # 中文Embedding模型 # 向量数据库配置 vector_store_path: str ./data/faiss_index knowledge_base_dir: str ./data/knowledge # 服务器配置 host: str 0.0.0.0 port: int 8000 # 安全/合规关键词示例 forbidden_keywords: list [推荐股票, 保证收益, 稳赚不赔, 内幕消息] mandatory_disclaimer: str 【免责声明】本回答基于公开信息生成不构成任何投资建议。市场有风险投资需谨慎。 class Config: env_file .env settings Settings() # 初始化日志 logger.add(logs/finance_qa.log, rotation10 MB, levelINFO)2.2.2 工具层实现 (app/tools/)工具是Agent的手和脚。我们创建几个金融领域常用的工具。1. 金融计算器工具 (app/tools/calculator.py)# app/tools/calculator.py from langchain.tools import BaseTool from pydantic import BaseModel, Field from typing import Type import math class CalculatorInput(BaseModel): expression: str Field(description一个合法的数学表达式例如(12)*3/4) class FinanceCalculatorTool(BaseTool): name finance_calculator description 用于执行金融相关的数学计算如年化收益率、现值、终值等。输入应为数学表达式。 args_schema: Type[BaseModel] CalculatorInput def _run(self, expression: str) - str: 执行计算 try: # 警告直接eval在生产环境是危险的此处仅作演示。 # 真实场景应使用更安全的评估库如 ast.literal_eval或解析器。 # 这里可以扩展为专门的金融公式解析器如 numpy_financial result eval(expression, {__builtins__: {}}, {math: math}) return f计算结果: {result} except Exception as e: return f计算错误: {e} async def _arun(self, expression: str): raise NotImplementedError(异步调用暂不支持) # 示例可以扩展更多工具如 # - StockPriceTool: 查询实时股价需接入API # - NewsSearchTool: 搜索金融新闻 # - ReportQueryTool: 查询内部研报数据库2. 知识库检索工具 (app/tools/retriever.py) 这是RAG检索增强生成的核心负责从向量数据库中查找相关信息。# app/tools/retriever.py from langchain.vectorstores import FAISS from langchain.embeddings import HuggingFaceEmbeddings from app.core.config import settings import os class KnowledgeRetriever: def __init__(self): # 加载嵌入模型 self.embeddings HuggingFaceEmbeddings( model_namesettings.embedding_model, model_kwargs{device: cpu}, # 可改为 cuda encode_kwargs{normalize_embeddings: True} ) self.vector_store None self._load_vector_store() def _load_vector_store(self): 加载已有的向量数据库 if os.path.exists(settings.vector_store_path): self.vector_store FAISS.load_local( settings.vector_store_path, self.embeddings, allow_dangerous_deserializationTrue # 注意安全提示 ) print(f向量数据库已从 {settings.vector_store_path} 加载) else: print(警告向量数据库不存在请先运行知识库构建脚本。) self.vector_store None def retrieve(self, query: str, top_k: int 3): 检索相关文档 if self.vector_store is None: return [] docs self.vector_store.similarity_search(query, ktop_k) return docs def get_relevant_context(self, query: str) - str: 将检索到的文档合并为上下文文本 docs self.retrieve(query) context \n\n.join([doc.page_content for doc in docs]) return context # 全局检索器实例 retriever KnowledgeRetriever()2.2.3 约束层Guardrails实现 (app/guards/)Guardrails是Harness的“安全围栏”用于在输入/输出层面进行过滤和修正。1. 关键词过滤守卫 (app/guards/keyword_guard.py)# app/guards/keyword_guard.py from app.core.config import settings from loguru import logger from typing import Tuple class KeywordGuard: 检查用户输入或模型输出中是否包含违禁关键词 def __init__(self): self.forbidden_keywords settings.forbidden_keywords self.disclaimer settings.mandatory_disclaimer def check_input(self, user_input: str) - Tuple[bool, str]: 检查用户输入返回 (是否通过, 消息) for keyword in self.forbidden_keywords: if keyword in user_input: logger.warning(f用户输入触发违禁词: {keyword}) return False, f您的问题涉及{keyword}根据合规要求我无法回答该类问题。请咨询持牌投资顾问。 return True, def enforce_output(self, agent_output: str) - str: 对模型输出进行后处理添加免责声明等 # 简单的后处理确保输出包含免责声明 if not agent_output.endswith(self.disclaimer): agent_output f\n\n{self.disclaimer} # 这里可以添加更多规则如格式化数字、替换敏感词等 return agent_output # 全局守卫实例 keyword_guard KeywordGuard()2.2.4 智能体Agent与链Chain编排 (app/chains/)这是Harness的“控制层”我们将使用LangChain来组装各个部件。1. 构建带有工具和上下文的Agent (app/chains/qa_chain.py)# app/chains/qa_chain.py from langchain.agents import AgentExecutor, create_react_agent from langchain.memory import ConversationBufferMemory from langchain.prompts import PromptTemplate from langchain.tools import Tool from langchain_community.llms.huggingface_pipeline import HuggingFacePipeline from transformers import AutoModelForCausalLM, AutoTokenizer, pipeline from app.core.config import settings from app.tools.calculator import FinanceCalculatorTool from app.tools.retriever import retriever import torch def create_finance_agent(): 创建金融问答Agent # 1. 加载本地Qwen模型 print(正在加载Qwen模型...) tokenizer AutoTokenizer.from_pretrained( settings.model_name, trust_remote_codeTrue, cache_dirsettings.model_cache_dir ) model AutoModelForCausalLM.from_pretrained( settings.model_name, torch_dtypetorch.float16, # 半精度节省显存 device_mapauto, # 自动分配设备 trust_remote_codeTrue, cache_dirsettings.model_cache_dir ) # 创建文本生成管道 pipe pipeline( text-generation, modelmodel, tokenizertokenizer, max_new_tokens512, temperature0.1, # 低温度使输出更确定 do_sampleTrue, ) llm HuggingFacePipeline(pipelinepipe) # 2. 定义工具列表 tools [ FinanceCalculatorTool(), Tool( nameknowledge_base_search, funclambda q: retriever.get_relevant_context(q), description在内部金融知识库中搜索相关信息。输入是一个查询问题。 ), # 可以在此添加更多工具... ] # 3. 设计Agent的提示词模板 (ReAct格式) prompt_template 你是一个专业的金融领域AI助手必须严格遵守以下规则 1. 回答必须基于提供的事实和知识。 2. 如果用户询问投资建议、个股推荐或市场预测必须拒绝并引导用户咨询持牌顾问。 3. 可以使用工具来获取信息或进行计算。 4. 所有数字结论必须说明计算过程或数据来源。 历史对话 {history} 当前问题{input} 你拥有以下工具 {tools} 请按以下格式思考 思考我需要先理解问题并决定是否需要使用工具。 行动要使用的工具名必须是[{tool_names}]中的一个。 行动输入工具的输入 观察工具返回的结果 ...这个思考-行动-观察循环可以重复多次 最终答案基于所有观察给出最终回答。 开始 {agent_scratchpad} prompt PromptTemplate.from_template(prompt_template) # 4. 创建记忆支持多轮对话 memory ConversationBufferMemory(memory_keyhistory, return_messagesTrue) # 5. 创建ReAct Agent agent create_react_agent(llm, tools, prompt) # 6. 创建执行器并传入记忆 agent_executor AgentExecutor( agentagent, toolstools, memorymemory, verboseTrue, # 设置为True可以看到Agent的思考过程生产环境可设为False handle_parsing_errorsTrue, max_iterations3 # 限制最大迭代次数防止死循环 ) return agent_executor # 全局Agent实例懒加载或单例模式 _agent_instance None def get_agent(): global _agent_instance if _agent_instance is None: _agent_instance create_finance_agent() return _agent_instance2.3 构建知识库RAG准备Harness系统需要“知识”作为依据。我们使用RAG技术将内部文档向量化。知识库构建脚本 (scripts/build_knowledge_base.py)# scripts/build_knowledge_base.py import os from langchain.document_loaders import DirectoryLoader, TextLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.embeddings import HuggingFaceEmbeddings from langchain.vectorstores import FAISS from app.core.config import settings def build_knowledge_base(): 加载文档切分生成向量数据库 # 1. 加载文档支持txt, md, pdf等 loader DirectoryLoader( settings.knowledge_base_dir, glob**/*.txt, # 可根据需要修改为 **/*.md 或使用 PyPDFLoader loader_clsTextLoader ) documents loader.load() print(f已加载 {len(documents)} 个文档) # 2. 分割文本为小块 text_splitter RecursiveCharacterTextSplitter( chunk_size500, # 每个块的大小 chunk_overlap50, # 块之间的重叠 separators[\n\n, \n, 。, , , ] ) splits text_splitter.split_documents(documents) print(f分割为 {len(splits)} 个文本块) # 3. 创建嵌入模型 embeddings HuggingFaceEmbeddings( model_namesettings.embedding_model, model_kwargs{device: cpu}, encode_kwargs{normalize_embeddings: True} ) # 4. 生成向量存储并保存 vectorstore FAISS.from_documents(splits, embeddings) vectorstore.save_local(settings.vector_store_path) print(f向量数据库已保存至 {settings.vector_store_path}) if __name__ __main__: # 确保知识库目录存在 os.makedirs(settings.knowledge_base_dir, exist_okTrue) os.makedirs(os.path.dirname(settings.vector_store_path), exist_okTrue) # 放入示例文档实际项目中替换为你的文档 sample_doc_path os.path.join(settings.knowledge_base_dir, 金融产品介绍.txt) if not os.path.exists(sample_doc_path): with open(sample_doc_path, w, encodingutf-8) as f: f.write(货币市场基金主要投资于短期货币工具如国债、央行票据、商业票据等。具有流动性好、风险低的特点。 股票型基金80%以上的基金资产投资于股票。预期收益高风险也较高。 债券型基金80%以上的基金资产投资于债券。风险收益特征介于货币基金和股票基金之间。 合格境内机构投资者(QDII)基金经批准从事境外证券市场的股票、债券等有价证券业务的证券投资基金。) build_knowledge_base()运行此脚本构建知识库python scripts/build_knowledge_base.py2.4 集成与API服务 (app/main.py)最后我们用FastAPI将所有模块集成起来提供HTTP API。# app/main.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import Optional from app.chains.qa_chain import get_agent from app.guards.keyword_guard import keyword_guard from loguru import logger import uvicorn app FastAPI(title金融大模型问答机器人 Harness Demo) class QueryRequest(BaseModel): question: str session_id: Optional[str] None # 用于区分不同对话会话 class QueryResponse(BaseModel): success: bool answer: str error_message: Optional[str] None app.on_event(startup) async def startup_event(): 服务启动时初始化Agent懒加载也可 logger.info(正在初始化金融问答Agent...) # 此处调用 get_agent() 会触发模型加载 _ get_agent() logger.info(Agent初始化完成。) app.post(/query, response_modelQueryResponse) async def query_knowledge(request: QueryRequest): 核心问答接口 user_question request.question.strip() if not user_question: return QueryResponse(successFalse, answer, error_message问题不能为空) # 1. Guardrail: 输入检查 is_allowed, guard_msg keyword_guard.check_input(user_question) if not is_allowed: return QueryResponse(successFalse, answer, error_messageguard_msg) try: # 2. 获取Agent并执行 agent get_agent() # LangChain Agent的输入格式 agent_input {input: user_question} result agent.invoke(agent_input) # 3. 从结果中提取最终答案 # ReAct Agent的最终答案通常在输出中 answer result.get(output, str(result)) # 4. Guardrail: 输出后处理 final_answer keyword_guard.enforce_output(answer) logger.info(fQ: {user_question[:50]}... | A: {final_answer[:50]}...) return QueryResponse(successTrue, answerfinal_answer, error_messageNone) except Exception as e: logger.error(f处理问题时发生错误: {e}, exc_infoTrue) return QueryResponse( successFalse, answer, error_messagef系统处理时出现错误: {str(e)} ) app.get(/health) async def health_check(): return {status: healthy} if __name__ __main__: from app.core.config import settings uvicorn.run(app, hostsettings.host, portsettings.port)2.5 运行与测试启动服务cd finance_qa_harness python app/main.py服务将在http://localhost:8000启动。测试API 使用curl或 Postman 进行测试。curl -X POST http://localhost:8000/query \ -H Content-Type: application/json \ -d {question: 请帮我计算一下如果年化收益率是5%投资10000元3年后的终值是多少}预期行为Agent会先进行“思考”然后调用finance_calculator工具进行计算最后给出答案并附上免责声明。curl -X POST http://localhost:8000/query \ -H Content-Type: application/json \ -d {question: 货币市场基金有什么特点}预期行为Agent会调用knowledge_base_search工具从我们构建的知识库中检索相关信息然后生成回答。curl -X POST http://localhost:8000/query \ -H Content-Type: application/json \ -d {question: 推荐一只明天会涨的股票}预期行为KeywordGuard会拦截此输入直接返回拒绝回答的消息Agent根本不会被触发。3. 项目进阶与Harness工程深化上面的项目展示了Harness的核心思想。要将其用于生产还需要考虑更多方面3.1 更复杂的控制流Chain of Thought Router对于复杂问题单一的Agent可能不够。我们可以设计一个主控路由器Router根据问题类型选择不同的子链Chain或子Agent。简单QA直接走RAG检索生成链。计算问题路由到计算工具链。多步分析如“对比A基金和B基金”拆解为“检索A信息”、“检索B信息”、“对比分析”三个子任务按顺序执行。3.2 高级Guardrails输出格式验证使用Pydantic模型强制要求模型输出结构化数据JSON。事实一致性检查将模型的回答与检索到的源文档进行交叉验证避免幻觉。毒性/偏见检测集成专门的分类模型对输出进行二次过滤。3.3 状态管理与记忆优化向量记忆将长对话历史的关键信息向量化存储实现更精准的长期记忆。外部知识图谱对于金融实体公司、产品、人物接入知识图谱如Neo4j实现关系推理。这正是GraphRAG的用武之地它能比普通向量检索更好地理解实体间关系。3.4 模型微调LoRA/SFT为了让模型更精通金融领域收集高质量的金融问答对数据。使用LoRA等技术对基座模型如Qwen进行高效微调让模型掌握专业术语和回答风格。微调后的模型再接入上述Harness系统效果会大幅提升。3.5 评估与监控构建测试集涵盖常规问题、边界问题、对抗性问题。自动化评估使用LLM作为裁判LLM-as-a-Judge或基于规则是否包含免责声明、是否调用了正确工具进行自动化评估。链路追踪记录每个请求的完整Chain执行过程、工具调用记录、Token消耗等便于调试和优化。4. 常见问题与排查思路问题现象可能原因解决思路启动服务时模型加载失败OOM显存不足。7B模型约需14GB GPU显存FP16。1. 使用device_map”cpu”强制用CPU慢。2. 使用量化版本模型如Qwen-7B-Chat-Int4。3. 升级硬件或使用云GPU。Agent陷入思考循环不输出答案max_iterations设置过高或提示词未引导出Final Answer。1. 检查verboseTrue时的日志看Agent卡在哪一步。2. 优化提示词明确要求最终输出Final Answer:。3. 适当降低max_iterations如3。检索工具返回无关内容1. 文档切分不合理。2. Embedding模型不匹配。3. 检索top_k太大。1. 调整chunk_size和chunk_overlap。2. 尝试不同的Embedding模型如text2vec。3. 减小top_k或使用MMR搜索增加多样性。工具调用参数错误工具的描述description不够清晰导致LLM无法理解如何使用。细化工具描述明确输入格式和用途。可以使用少量示例Few-shot放在提示词中。回答不符合合规要求Guardrails未生效或规则有漏洞。1. 检查keyword_guard的拦截逻辑。2. 在输出后处理阶段增加更严格的规则如正则匹配。3. 考虑使用一个更小的“审查模型”对输出进行二次扫描。多轮对话记忆混乱ConversationBufferMemory可能记忆过长导致混乱。1. 使用ConversationSummaryMemory或ConversationBufferWindowMemory。2. 在每轮对话中显式地将最重要的历史信息再次注入提示词。5. 最佳实践与工程建议设计先行在写代码前先用流程图或文档定义好Harness的各个组件工具、守卫、工作流及其交互关系。模块化与可插拔像本项目一样将工具、守卫、链设计成独立的模块。这样便于单独测试、替换和升级。例如可以轻松将Qwen换成GPT-4或将FAISS换成Pinecone。配置化将所有可变的参数模型路径、关键词、提示词模板抽离到配置文件如config.py或YAML中避免硬编码。可观测性在关键节点输入、输出、工具调用、守卫触发添加详细的日志。考虑集成像LangSmith这样的LLM应用监控平台。渐进式复杂化不要一开始就设计一个庞大的Harness。从一个最简单的、能解决核心问题的链开始然后逐步添加工具、守卫和路由逻辑。安全第一对用户输入进行严格的清洗和校验。工具调用特别是写操作、网络请求要有权限控制和用量限制。永远不要相信模型的原始输出必须经过守卫层的过滤。提示词工程仍是基础Harness不是替代提示词工程而是建立在它之上。清晰、结构化的提示词是Agent正确使用工具和遵守规则的前提。通过这个完整的项目你应该已经体会到Harness Engineering的精髓它通过系统性的设计将强大但不可控的LLM转变为可靠、安全、可应用于核心业务的软件组件。未来的AI应用开发比拼的将不再是谁的提示词更巧妙而是谁的Harness设计得更健壮、更灵活、更安全。
Harness Engineering：构建可控AI智能体的系统工程实践
你好我是专注于AI大模型应用开发的工程师。在构建和部署AI应用尤其是基于大语言模型LLM的智能体Agent时你是否遇到过这样的困境模型本身能力很强但一接入实际业务就变得不稳定、不可控回答天马行空甚至执行危险操作代码调试像“开盲盒”每次迭代都伴随着巨大的不确定性。这正是传统“提示词工程”的瓶颈。我们需要的不是更聪明的模型而是能让模型在复杂系统中“可靠工作”的工程方法。本文将为你系统性地拆解一个正在重塑AI应用开发的新范式——Harness Engineering驾驭工程。我将从核心概念讲起通过一个完整的“金融大模型问答机器人”项目实战手把手带你从零构建一个具备业务逻辑约束、安全可控的AI应用。无论你是刚接触AI应用开发的新手还是正在为Agent落地头疼的资深开发者都能从中获得一套可复用的工程框架和代码。1. Harness Engineering从“提示词”到“系统工程”的范式革命在深入代码之前我们必须先理解“Harness”究竟是什么。它不是一个具体的框架或工具而是一种工程思想和架构模式。1.1 什么是Harness你可以把Harness理解为AI智能体的“缰绳”和“驾驶舱”。它的核心目标不是增强模型的内在能力而是从外部对模型的行为进行约束、引导和编排确保其在预设的轨道内安全、可靠、可控地完成任务。传统提示词工程类似于给一个非常聪明但缺乏纪律的员工大模型一份工作说明书Prompt期望他理解并执行。结果高度依赖说明书的撰写水平且员工仍有自由发挥的空间可能导致意外。Harness Engineering不仅提供说明书还为这位员工配备了一套完整的“工作台”Harness。这个工作台规定了操作流程Orchestration提供了专用工具Tools设置了安全围栏Guardrails并实时监控工作状态Monitoring。员工必须在工作台的规则下使用工具完成任务。1.2 为什么需要Harness—— 智能体落地的核心挑战不可控性LLM的生成具有随机性可能产生不符合业务逻辑、事实错误或有害的内容。缺乏状态与记忆简单的对话难以处理多轮、复杂的业务流程无法维持上下文状态。工具使用混乱如何让Agent在合适的时机以正确的参数调用外部工具API、数据库、函数安全与合规风险在金融、医疗等领域AI的每一句话、每一个决策都必须可审计、可追溯、符合规范。调试与评估困难黑盒模型使得问题定位和效果评估成本极高。Harness工程正是为了解决这些问题而生它强调系统设计优先将LLM视为一个需要被“驾驭”的核心组件而非全部。1.3 Harness的核心组件一个典型的Harness系统通常包含以下层次控制层Controller定义工作流Workflow决定任务的分解、步骤顺序和Agent的调度。例如先检索知识再分析最后生成报告。约束层Guardrails制定安全策略和业务规则。例如禁止回答股票预测类问题所有数字必须格式化引用必须标明来源。工具层Tools为Agent提供获取信息、执行操作的能力。例如查询数据库的Tool、调用计算API的Tool、搜索内部文档的Tool。状态管理层State Management维护对话历史、中间结果和业务上下文确保多轮交互的连贯性。评估与监控层Evaluation Monitoring对Agent的输入、输出、中间过程进行记录、分析和评估持续优化系统。接下来我们将把这些概念付诸实践构建一个实战项目。2. 项目实战金融大模型问答机器人项目目标构建一个服务于内部员工的金融知识问答助手。它不仅能基于内部知识库产品手册、合规文件、研报回答问题还能在回答时自动遵循金融行业的合规话术并对查询进行路由例如转人工、调用计算工具。技术栈选型LLMQwen-7B-Chat (开源可本地部署兼顾效果与成本)应用框架LangChain (Agent和Chain编排的核心)知识库与检索LangChain (TextSplitter, Embeddings), FAISS / Chroma (向量数据库)后端APIFastAPI (轻量高效)高级能力RAG (检索增强生成), Function Calling (工具调用)微调与优化LoRA (高效微调使模型更懂金融术语)2.1 环境准备与项目初始化我们使用Python 3.9进行开发。首先创建项目结构并安装依赖。# 创建项目目录 mkdir finance_qa_harness cd finance_qa_harness mkdir -p app/{core, tools, guards, chains, models} app/data app/static # 创建虚拟环境可选但推荐 python -m venv venv source venv/bin/activate # Linux/Mac # venv\Scripts\activate # Windows # 创建 requirements.txt 并安装核心依赖requirements.txt文件内容# 核心框架 langchain0.1.0 langchain-community0.0.10 langchain-core0.1.0 # LLM 与 Embedding transformers4.35.0 torch sentence-transformers modelscope # 用于下载通义千问模型 # 向量数据库 faiss-cpu # 或 faiss-gpu (如果有CUDA) # chromadb # Web框架与工具 fastapi uvicorn[standard] pydantic python-multipart # 工具与工具调用 requests pandas numpy # 其他 python-dotenv loguru安装依赖pip install -r requirements.txt2.2 核心模块设计与实现我们的Harness系统由多个模块组成下面逐一实现。2.2.1 配置与模型加载模块 (app/core/config.py)集中管理配置项如模型路径、API密钥如果用云端模型、工具配置等。# app/core/config.py import os from typing import Optional from pydantic_settings import BaseSettings from loguru import logger class Settings(BaseSettings): # 模型配置 model_name: str Qwen/Qwen-7B-Chat # Hugging Face 或 ModelScope 模型ID model_cache_dir: Optional[str] ./models embedding_model: str BAAI/bge-small-zh-v1.5 # 中文Embedding模型 # 向量数据库配置 vector_store_path: str ./data/faiss_index knowledge_base_dir: str ./data/knowledge # 服务器配置 host: str 0.0.0.0 port: int 8000 # 安全/合规关键词示例 forbidden_keywords: list [推荐股票, 保证收益, 稳赚不赔, 内幕消息] mandatory_disclaimer: str 【免责声明】本回答基于公开信息生成不构成任何投资建议。市场有风险投资需谨慎。 class Config: env_file .env settings Settings() # 初始化日志 logger.add(logs/finance_qa.log, rotation10 MB, levelINFO)2.2.2 工具层实现 (app/tools/)工具是Agent的手和脚。我们创建几个金融领域常用的工具。1. 金融计算器工具 (app/tools/calculator.py)# app/tools/calculator.py from langchain.tools import BaseTool from pydantic import BaseModel, Field from typing import Type import math class CalculatorInput(BaseModel): expression: str Field(description一个合法的数学表达式例如(12)*3/4) class FinanceCalculatorTool(BaseTool): name finance_calculator description 用于执行金融相关的数学计算如年化收益率、现值、终值等。输入应为数学表达式。 args_schema: Type[BaseModel] CalculatorInput def _run(self, expression: str) - str: 执行计算 try: # 警告直接eval在生产环境是危险的此处仅作演示。 # 真实场景应使用更安全的评估库如 ast.literal_eval或解析器。 # 这里可以扩展为专门的金融公式解析器如 numpy_financial result eval(expression, {__builtins__: {}}, {math: math}) return f计算结果: {result} except Exception as e: return f计算错误: {e} async def _arun(self, expression: str): raise NotImplementedError(异步调用暂不支持) # 示例可以扩展更多工具如 # - StockPriceTool: 查询实时股价需接入API # - NewsSearchTool: 搜索金融新闻 # - ReportQueryTool: 查询内部研报数据库2. 知识库检索工具 (app/tools/retriever.py) 这是RAG检索增强生成的核心负责从向量数据库中查找相关信息。# app/tools/retriever.py from langchain.vectorstores import FAISS from langchain.embeddings import HuggingFaceEmbeddings from app.core.config import settings import os class KnowledgeRetriever: def __init__(self): # 加载嵌入模型 self.embeddings HuggingFaceEmbeddings( model_namesettings.embedding_model, model_kwargs{device: cpu}, # 可改为 cuda encode_kwargs{normalize_embeddings: True} ) self.vector_store None self._load_vector_store() def _load_vector_store(self): 加载已有的向量数据库 if os.path.exists(settings.vector_store_path): self.vector_store FAISS.load_local( settings.vector_store_path, self.embeddings, allow_dangerous_deserializationTrue # 注意安全提示 ) print(f向量数据库已从 {settings.vector_store_path} 加载) else: print(警告向量数据库不存在请先运行知识库构建脚本。) self.vector_store None def retrieve(self, query: str, top_k: int 3): 检索相关文档 if self.vector_store is None: return [] docs self.vector_store.similarity_search(query, ktop_k) return docs def get_relevant_context(self, query: str) - str: 将检索到的文档合并为上下文文本 docs self.retrieve(query) context \n\n.join([doc.page_content for doc in docs]) return context # 全局检索器实例 retriever KnowledgeRetriever()2.2.3 约束层Guardrails实现 (app/guards/)Guardrails是Harness的“安全围栏”用于在输入/输出层面进行过滤和修正。1. 关键词过滤守卫 (app/guards/keyword_guard.py)# app/guards/keyword_guard.py from app.core.config import settings from loguru import logger from typing import Tuple class KeywordGuard: 检查用户输入或模型输出中是否包含违禁关键词 def __init__(self): self.forbidden_keywords settings.forbidden_keywords self.disclaimer settings.mandatory_disclaimer def check_input(self, user_input: str) - Tuple[bool, str]: 检查用户输入返回 (是否通过, 消息) for keyword in self.forbidden_keywords: if keyword in user_input: logger.warning(f用户输入触发违禁词: {keyword}) return False, f您的问题涉及{keyword}根据合规要求我无法回答该类问题。请咨询持牌投资顾问。 return True, def enforce_output(self, agent_output: str) - str: 对模型输出进行后处理添加免责声明等 # 简单的后处理确保输出包含免责声明 if not agent_output.endswith(self.disclaimer): agent_output f\n\n{self.disclaimer} # 这里可以添加更多规则如格式化数字、替换敏感词等 return agent_output # 全局守卫实例 keyword_guard KeywordGuard()2.2.4 智能体Agent与链Chain编排 (app/chains/)这是Harness的“控制层”我们将使用LangChain来组装各个部件。1. 构建带有工具和上下文的Agent (app/chains/qa_chain.py)# app/chains/qa_chain.py from langchain.agents import AgentExecutor, create_react_agent from langchain.memory import ConversationBufferMemory from langchain.prompts import PromptTemplate from langchain.tools import Tool from langchain_community.llms.huggingface_pipeline import HuggingFacePipeline from transformers import AutoModelForCausalLM, AutoTokenizer, pipeline from app.core.config import settings from app.tools.calculator import FinanceCalculatorTool from app.tools.retriever import retriever import torch def create_finance_agent(): 创建金融问答Agent # 1. 加载本地Qwen模型 print(正在加载Qwen模型...) tokenizer AutoTokenizer.from_pretrained( settings.model_name, trust_remote_codeTrue, cache_dirsettings.model_cache_dir ) model AutoModelForCausalLM.from_pretrained( settings.model_name, torch_dtypetorch.float16, # 半精度节省显存 device_mapauto, # 自动分配设备 trust_remote_codeTrue, cache_dirsettings.model_cache_dir ) # 创建文本生成管道 pipe pipeline( text-generation, modelmodel, tokenizertokenizer, max_new_tokens512, temperature0.1, # 低温度使输出更确定 do_sampleTrue, ) llm HuggingFacePipeline(pipelinepipe) # 2. 定义工具列表 tools [ FinanceCalculatorTool(), Tool( nameknowledge_base_search, funclambda q: retriever.get_relevant_context(q), description在内部金融知识库中搜索相关信息。输入是一个查询问题。 ), # 可以在此添加更多工具... ] # 3. 设计Agent的提示词模板 (ReAct格式) prompt_template 你是一个专业的金融领域AI助手必须严格遵守以下规则 1. 回答必须基于提供的事实和知识。 2. 如果用户询问投资建议、个股推荐或市场预测必须拒绝并引导用户咨询持牌顾问。 3. 可以使用工具来获取信息或进行计算。 4. 所有数字结论必须说明计算过程或数据来源。 历史对话 {history} 当前问题{input} 你拥有以下工具 {tools} 请按以下格式思考 思考我需要先理解问题并决定是否需要使用工具。 行动要使用的工具名必须是[{tool_names}]中的一个。 行动输入工具的输入 观察工具返回的结果 ...这个思考-行动-观察循环可以重复多次 最终答案基于所有观察给出最终回答。 开始 {agent_scratchpad} prompt PromptTemplate.from_template(prompt_template) # 4. 创建记忆支持多轮对话 memory ConversationBufferMemory(memory_keyhistory, return_messagesTrue) # 5. 创建ReAct Agent agent create_react_agent(llm, tools, prompt) # 6. 创建执行器并传入记忆 agent_executor AgentExecutor( agentagent, toolstools, memorymemory, verboseTrue, # 设置为True可以看到Agent的思考过程生产环境可设为False handle_parsing_errorsTrue, max_iterations3 # 限制最大迭代次数防止死循环 ) return agent_executor # 全局Agent实例懒加载或单例模式 _agent_instance None def get_agent(): global _agent_instance if _agent_instance is None: _agent_instance create_finance_agent() return _agent_instance2.3 构建知识库RAG准备Harness系统需要“知识”作为依据。我们使用RAG技术将内部文档向量化。知识库构建脚本 (scripts/build_knowledge_base.py)# scripts/build_knowledge_base.py import os from langchain.document_loaders import DirectoryLoader, TextLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.embeddings import HuggingFaceEmbeddings from langchain.vectorstores import FAISS from app.core.config import settings def build_knowledge_base(): 加载文档切分生成向量数据库 # 1. 加载文档支持txt, md, pdf等 loader DirectoryLoader( settings.knowledge_base_dir, glob**/*.txt, # 可根据需要修改为 **/*.md 或使用 PyPDFLoader loader_clsTextLoader ) documents loader.load() print(f已加载 {len(documents)} 个文档) # 2. 分割文本为小块 text_splitter RecursiveCharacterTextSplitter( chunk_size500, # 每个块的大小 chunk_overlap50, # 块之间的重叠 separators[\n\n, \n, 。, , , ] ) splits text_splitter.split_documents(documents) print(f分割为 {len(splits)} 个文本块) # 3. 创建嵌入模型 embeddings HuggingFaceEmbeddings( model_namesettings.embedding_model, model_kwargs{device: cpu}, encode_kwargs{normalize_embeddings: True} ) # 4. 生成向量存储并保存 vectorstore FAISS.from_documents(splits, embeddings) vectorstore.save_local(settings.vector_store_path) print(f向量数据库已保存至 {settings.vector_store_path}) if __name__ __main__: # 确保知识库目录存在 os.makedirs(settings.knowledge_base_dir, exist_okTrue) os.makedirs(os.path.dirname(settings.vector_store_path), exist_okTrue) # 放入示例文档实际项目中替换为你的文档 sample_doc_path os.path.join(settings.knowledge_base_dir, 金融产品介绍.txt) if not os.path.exists(sample_doc_path): with open(sample_doc_path, w, encodingutf-8) as f: f.write(货币市场基金主要投资于短期货币工具如国债、央行票据、商业票据等。具有流动性好、风险低的特点。 股票型基金80%以上的基金资产投资于股票。预期收益高风险也较高。 债券型基金80%以上的基金资产投资于债券。风险收益特征介于货币基金和股票基金之间。 合格境内机构投资者(QDII)基金经批准从事境外证券市场的股票、债券等有价证券业务的证券投资基金。) build_knowledge_base()运行此脚本构建知识库python scripts/build_knowledge_base.py2.4 集成与API服务 (app/main.py)最后我们用FastAPI将所有模块集成起来提供HTTP API。# app/main.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import Optional from app.chains.qa_chain import get_agent from app.guards.keyword_guard import keyword_guard from loguru import logger import uvicorn app FastAPI(title金融大模型问答机器人 Harness Demo) class QueryRequest(BaseModel): question: str session_id: Optional[str] None # 用于区分不同对话会话 class QueryResponse(BaseModel): success: bool answer: str error_message: Optional[str] None app.on_event(startup) async def startup_event(): 服务启动时初始化Agent懒加载也可 logger.info(正在初始化金融问答Agent...) # 此处调用 get_agent() 会触发模型加载 _ get_agent() logger.info(Agent初始化完成。) app.post(/query, response_modelQueryResponse) async def query_knowledge(request: QueryRequest): 核心问答接口 user_question request.question.strip() if not user_question: return QueryResponse(successFalse, answer, error_message问题不能为空) # 1. Guardrail: 输入检查 is_allowed, guard_msg keyword_guard.check_input(user_question) if not is_allowed: return QueryResponse(successFalse, answer, error_messageguard_msg) try: # 2. 获取Agent并执行 agent get_agent() # LangChain Agent的输入格式 agent_input {input: user_question} result agent.invoke(agent_input) # 3. 从结果中提取最终答案 # ReAct Agent的最终答案通常在输出中 answer result.get(output, str(result)) # 4. Guardrail: 输出后处理 final_answer keyword_guard.enforce_output(answer) logger.info(fQ: {user_question[:50]}... | A: {final_answer[:50]}...) return QueryResponse(successTrue, answerfinal_answer, error_messageNone) except Exception as e: logger.error(f处理问题时发生错误: {e}, exc_infoTrue) return QueryResponse( successFalse, answer, error_messagef系统处理时出现错误: {str(e)} ) app.get(/health) async def health_check(): return {status: healthy} if __name__ __main__: from app.core.config import settings uvicorn.run(app, hostsettings.host, portsettings.port)2.5 运行与测试启动服务cd finance_qa_harness python app/main.py服务将在http://localhost:8000启动。测试API 使用curl或 Postman 进行测试。curl -X POST http://localhost:8000/query \ -H Content-Type: application/json \ -d {question: 请帮我计算一下如果年化收益率是5%投资10000元3年后的终值是多少}预期行为Agent会先进行“思考”然后调用finance_calculator工具进行计算最后给出答案并附上免责声明。curl -X POST http://localhost:8000/query \ -H Content-Type: application/json \ -d {question: 货币市场基金有什么特点}预期行为Agent会调用knowledge_base_search工具从我们构建的知识库中检索相关信息然后生成回答。curl -X POST http://localhost:8000/query \ -H Content-Type: application/json \ -d {question: 推荐一只明天会涨的股票}预期行为KeywordGuard会拦截此输入直接返回拒绝回答的消息Agent根本不会被触发。3. 项目进阶与Harness工程深化上面的项目展示了Harness的核心思想。要将其用于生产还需要考虑更多方面3.1 更复杂的控制流Chain of Thought Router对于复杂问题单一的Agent可能不够。我们可以设计一个主控路由器Router根据问题类型选择不同的子链Chain或子Agent。简单QA直接走RAG检索生成链。计算问题路由到计算工具链。多步分析如“对比A基金和B基金”拆解为“检索A信息”、“检索B信息”、“对比分析”三个子任务按顺序执行。3.2 高级Guardrails输出格式验证使用Pydantic模型强制要求模型输出结构化数据JSON。事实一致性检查将模型的回答与检索到的源文档进行交叉验证避免幻觉。毒性/偏见检测集成专门的分类模型对输出进行二次过滤。3.3 状态管理与记忆优化向量记忆将长对话历史的关键信息向量化存储实现更精准的长期记忆。外部知识图谱对于金融实体公司、产品、人物接入知识图谱如Neo4j实现关系推理。这正是GraphRAG的用武之地它能比普通向量检索更好地理解实体间关系。3.4 模型微调LoRA/SFT为了让模型更精通金融领域收集高质量的金融问答对数据。使用LoRA等技术对基座模型如Qwen进行高效微调让模型掌握专业术语和回答风格。微调后的模型再接入上述Harness系统效果会大幅提升。3.5 评估与监控构建测试集涵盖常规问题、边界问题、对抗性问题。自动化评估使用LLM作为裁判LLM-as-a-Judge或基于规则是否包含免责声明、是否调用了正确工具进行自动化评估。链路追踪记录每个请求的完整Chain执行过程、工具调用记录、Token消耗等便于调试和优化。4. 常见问题与排查思路问题现象可能原因解决思路启动服务时模型加载失败OOM显存不足。7B模型约需14GB GPU显存FP16。1. 使用device_map”cpu”强制用CPU慢。2. 使用量化版本模型如Qwen-7B-Chat-Int4。3. 升级硬件或使用云GPU。Agent陷入思考循环不输出答案max_iterations设置过高或提示词未引导出Final Answer。1. 检查verboseTrue时的日志看Agent卡在哪一步。2. 优化提示词明确要求最终输出Final Answer:。3. 适当降低max_iterations如3。检索工具返回无关内容1. 文档切分不合理。2. Embedding模型不匹配。3. 检索top_k太大。1. 调整chunk_size和chunk_overlap。2. 尝试不同的Embedding模型如text2vec。3. 减小top_k或使用MMR搜索增加多样性。工具调用参数错误工具的描述description不够清晰导致LLM无法理解如何使用。细化工具描述明确输入格式和用途。可以使用少量示例Few-shot放在提示词中。回答不符合合规要求Guardrails未生效或规则有漏洞。1. 检查keyword_guard的拦截逻辑。2. 在输出后处理阶段增加更严格的规则如正则匹配。3. 考虑使用一个更小的“审查模型”对输出进行二次扫描。多轮对话记忆混乱ConversationBufferMemory可能记忆过长导致混乱。1. 使用ConversationSummaryMemory或ConversationBufferWindowMemory。2. 在每轮对话中显式地将最重要的历史信息再次注入提示词。5. 最佳实践与工程建议设计先行在写代码前先用流程图或文档定义好Harness的各个组件工具、守卫、工作流及其交互关系。模块化与可插拔像本项目一样将工具、守卫、链设计成独立的模块。这样便于单独测试、替换和升级。例如可以轻松将Qwen换成GPT-4或将FAISS换成Pinecone。配置化将所有可变的参数模型路径、关键词、提示词模板抽离到配置文件如config.py或YAML中避免硬编码。可观测性在关键节点输入、输出、工具调用、守卫触发添加详细的日志。考虑集成像LangSmith这样的LLM应用监控平台。渐进式复杂化不要一开始就设计一个庞大的Harness。从一个最简单的、能解决核心问题的链开始然后逐步添加工具、守卫和路由逻辑。安全第一对用户输入进行严格的清洗和校验。工具调用特别是写操作、网络请求要有权限控制和用量限制。永远不要相信模型的原始输出必须经过守卫层的过滤。提示词工程仍是基础Harness不是替代提示词工程而是建立在它之上。清晰、结构化的提示词是Agent正确使用工具和遵守规则的前提。通过这个完整的项目你应该已经体会到Harness Engineering的精髓它通过系统性的设计将强大但不可控的LLM转变为可靠、安全、可应用于核心业务的软件组件。未来的AI应用开发比拼的将不再是谁的提示词更巧妙而是谁的Harness设计得更健壮、更灵活、更安全。
