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？—— 智能体落地的核心挑战

1. 不可控性：LLM的生成具有随机性，可能产生不符合业务逻辑、事实错误或有害的内容。
2. 缺乏状态与记忆：简单的对话难以处理多轮、复杂的业务流程，无法维持上下文状态。
3. 工具使用混乱：如何让Agent在合适的时机，以正确的参数调用外部工具（API、数据库、函数）？
4. 安全与合规风险：在金融、医疗等领域，AI的每一句话、每一个决策都必须可审计、可追溯、符合规范。
5. 调试与评估困难：黑盒模型使得问题定位和效果评估成本极高。

Harness工程正是为了解决这些问题而生，它强调系统设计优先，将LLM视为一个需要被“驾驭”的核心组件，而非全部。

1.3 Harness的核心组件

一个典型的Harness系统通常包含以下层次：

• 控制层（Controller）：定义工作流（Workflow），决定任务的分解、步骤顺序和Agent的调度。例如，先检索知识，再分析，最后生成报告。
• 约束层（Guardrails）：制定安全策略和业务规则。例如，禁止回答股票预测类问题，所有数字必须格式化，引用必须标明来源。
• 工具层（Tools）：为Agent提供获取信息、执行操作的能力。例如，查询数据库的Tool、调用计算API的Tool、搜索内部文档的Tool。
• 状态管理层（State Management）：维护对话历史、中间结果和业务上下文，确保多轮交互的连贯性。
• 评估与监控层（Evaluation & Monitoring）：对Agent的输入、输出、中间过程进行记录、分析和评估，持续优化系统。

接下来，我们将把这些概念付诸实践，构建一个实战项目。

2. 项目实战：金融大模型问答机器人

项目目标：构建一个服务于内部员工的金融知识问答助手。它不仅能基于内部知识库（产品手册、合规文件、研报）回答问题，还能在回答时自动遵循金融行业的合规话术，并对查询进行路由（例如，转人工、调用计算工具）。

技术栈选型：

• LLM：Qwen-7B-Chat (开源，可本地部署，兼顾效果与成本)
• 应用框架：LangChain (Agent和Chain编排的核心)
• 知识库与检索：LangChain (TextSplitter, Embeddings), FAISS / Chroma (向量数据库)
• 后端API：FastAPI (轻量高效)
• 高级能力：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文件内容：

# 核心框架 langchain==0.1.0 langchain-community==0.0.10 langchain-core==0.1.0 # LLM 与 Embedding transformers>=4.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.txt

2.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", rotation="10 MB", level="INFO")

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="一个合法的数学表达式，例如：'(1+2)*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_name=settings.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_deserialization=True # 注意安全提示 ) 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, k=top_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_code=True, cache_dir=settings.model_cache_dir ) model = AutoModelForCausalLM.from_pretrained( settings.model_name, torch_dtype=torch.float16, # 半精度节省显存 device_map="auto", # 自动分配设备 trust_remote_code=True, cache_dir=settings.model_cache_dir ) # 创建文本生成管道 pipe = pipeline( "text-generation", model=model, tokenizer=tokenizer, max_new_tokens=512, temperature=0.1, # 低温度使输出更确定 do_sample=True, ) llm = HuggingFacePipeline(pipeline=pipe) # 2. 定义工具列表 tools = [ FinanceCalculatorTool(), Tool( name="knowledge_base_search", func=lambda 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_key="history", return_messages=True) # 5. 创建ReAct Agent agent = create_react_agent(llm, tools, prompt) # 6. 创建执行器，并传入记忆 agent_executor = AgentExecutor( agent=agent, tools=tools, memory=memory, verbose=True, # 设置为True可以看到Agent的思考过程，生产环境可设为False handle_parsing_errors=True, max_iterations=3 # 限制最大迭代次数，防止死循环 ) 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_instance

2.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_cls=TextLoader ) documents = loader.load() print(f"已加载 {len(documents)} 个文档") # 2. 分割文本为小块 text_splitter = RecursiveCharacterTextSplitter( chunk_size=500, # 每个块的大小 chunk_overlap=50, # 块之间的重叠 separators=["\n\n", "\n", "。", "，", " ", ""] ) splits = text_splitter.split_documents(documents) print(f"分割为 {len(splits)} 个文本块") # 3. 创建嵌入模型 embeddings = HuggingFaceEmbeddings( model_name=settings.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_ok=True) os.makedirs(os.path.dirname(settings.vector_store_path), exist_ok=True) # 放入示例文档（实际项目中替换为你的文档） 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', encoding='utf-8') as f: f.write("""货币市场基金：主要投资于短期货币工具，如国债、央行票据、商业票据等。具有流动性好、风险低的特点。 股票型基金：80%以上的基金资产投资于股票。预期收益高，风险也较高。 债券型基金：80%以上的基金资产投资于债券。风险收益特征介于货币基金和股票基金之间。 合格境内机构投资者(QDII)基金：经批准从事境外证券市场的股票、债券等有价证券业务的证券投资基金。""") build_knowledge_base()

运行此脚本构建知识库：

python scripts/build_knowledge_base.py

2.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_model=QueryResponse) async def query_knowledge(request: QueryRequest): """核心问答接口""" user_question = request.question.strip() if not user_question: return QueryResponse(success=False, answer="", error_message="问题不能为空") # 1. Guardrail: 输入检查 is_allowed, guard_msg = keyword_guard.check_input(user_question) if not is_allowed: return QueryResponse(success=False, answer="", error_message=guard_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(f"Q: {user_question[:50]}... | A: {final_answer[:50]}...") return QueryResponse(success=True, answer=final_answer, error_message=None) except Exception as e: logger.error(f"处理问题时发生错误: {e}", exc_info=True) return QueryResponse( success=False, answer="", error_message=f"系统处理时出现错误: {str(e)}" ) @app.get("/health") async def health_check(): return {"status": "healthy"} if __name__ == "__main__": from app.core.config import settings uvicorn.run(app, host=settings.host, port=settings.port)

2.5 运行与测试

1. 启动服务：

   cd finance_qa_harness python app/main.py

   服务将在http://localhost:8000启动。

2. 测试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）

为了让模型更精通金融领域：

1. 收集高质量的金融问答对数据。
2. 使用LoRA等技术对基座模型（如Qwen）进行高效微调，让模型掌握专业术语和回答风格。
3. 微调后的模型再接入上述Harness系统，效果会大幅提升。

3.5 评估与监控

• 构建测试集：涵盖常规问题、边界问题、对抗性问题。
• 自动化评估：使用LLM作为裁判（LLM-as-a-Judge），或基于规则（是否包含免责声明、是否调用了正确工具）进行自动化评估。
• 链路追踪：记录每个请求的完整Chain执行过程、工具调用记录、Token消耗等，便于调试和优化。

4. 常见问题与排查思路

5. 最佳实践与工程建议

1. 设计先行：在写代码前，先用流程图或文档定义好Harness的各个组件（工具、守卫、工作流）及其交互关系。
2. 模块化与可插拔：像本项目一样，将工具、守卫、链设计成独立的模块。这样便于单独测试、替换和升级。例如，可以轻松将Qwen换成GPT-4，或将FAISS换成Pinecone。
3. 配置化：将所有可变的参数（模型路径、关键词、提示词模板）抽离到配置文件（如config.py或YAML）中，避免硬编码。
4. 可观测性：在关键节点（输入、输出、工具调用、守卫触发）添加详细的日志。考虑集成像LangSmith这样的LLM应用监控平台。
5. 渐进式复杂化：不要一开始就设计一个庞大的Harness。从一个最简单的、能解决核心问题的链开始，然后逐步添加工具、守卫和路由逻辑。
6. 安全第一：
   • 对用户输入进行严格的清洗和校验。
   • 工具调用（特别是写操作、网络请求）要有权限控制和用量限制。
   • 永远不要相信模型的原始输出，必须经过守卫层的过滤。
7. 提示词工程仍是基础：Harness不是替代提示词工程，而是建立在它之上。清晰、结构化的提示词是Agent正确使用工具和遵守规则的前提。

通过这个完整的项目，你应该已经体会到Harness Engineering的精髓：它通过系统性的设计，将强大但不可控的LLM，转变为可靠、安全、可应用于核心业务的软件组件。未来的AI应用开发，比拼的将不再是谁的提示词更巧妙，而是谁的Harness设计得更健壮、更灵活、更安全。