构建大模型驾驭工程：从RAG到智能体，打造稳定可控的金融问答系统

最近在尝试将大模型集成到实际业务系统时发现一个普遍痛点模型本身能力很强但如何让它稳定、可靠、可控地工作却成了比调优模型本身更棘手的工程难题。网上资料要么是零散的Prompt技巧要么是复杂的论文解读缺乏一套从设计思想到项目落地的完整工程化方案。本文将以“Harness Engineering”驾驭工程为核心系统性地拆解如何为大模型应用构建一套健壮、可扩展的工程框架并附上一个完整的“金融大模型问答机器人”实战项目包含从零到一的代码、配置与避坑指南。无论你是刚接触AI应用开发的新手还是正在为模型稳定性发愁的工程师都能从中获得可直接复用的解决方案。1. 背景与核心概念为什么需要“驾驭”大模型在传统的软件开发中我们编写确定性的代码输入和输出之间的关系是明确的。然而大模型Large Language Model, LLM是一种概率模型其输出具有不确定性和开放性。直接将原始模型接入生产环境就像试图驾驭一匹未经驯服的野马——力量强大但方向不可控容易出现“幻觉”生成虚假信息、输出不稳定、无法处理复杂逻辑链等问题。Harness Engineering驾驭工程正是为了解决这一问题而生的工程范式。它的核心思想不是等待一个“完美”的模型而是通过系统性的工程化设计为模型构建一个可靠的工作环境与执行框架从而约束、引导、增强模型的能力使其能够稳定、安全、高效地完成特定任务。我们可以将其类比为汽车制造大模型LLM如同强大的发动机提供核心动力。Harness Engineering如同整车的底盘、传动系统、控制系统方向盘、刹车、ECU。它决定了动力如何被有效、安全地传递和控制最终让汽车能平稳行驶而非原地空转或失控。从“让模型写代码”到“设计让模型可靠工作的系统”这是开发范式的根本转变。一个优秀的Harness系统通常具备以下核心抽象能力分层将任务拆解为规划、执行、验证等不同层次模型专注于其擅长的部分如规划或决策其他部分由确定性代码或工具完成。模块边界清晰定义Agent、工具Tools、记忆Memory、评估Evaluation等模块的职责与接口。核心抽象设计统一的流程控制器Orchestrator、状态管理器和工具调用接口。扩展机制支持灵活地添加新的工具、知识源或验证规则。权限与安全模型控制模型可以访问的数据和可以执行的操作。接下来我们将从零开始构建一个体现Harness Engineering思想的实战项目。2. 环境准备与版本说明本项目将构建一个金融领域智能问答机器人。它不仅能回答通用金融知识还能基于提供的私有知识库如公司财报、内部产品文档进行精准回答并可以调用工具查询实时金融数据。技术栈选型说明LLM大语言模型Qwen-7B-Chat。选择开源模型便于本地部署和微调Qwen在中文理解和推理上表现优异。应用框架LangChain。它提供了构建基于LLM应用的丰富组件Models, Prompts, Chains, Agents, Tools是实现Harness思想的绝佳脚手架。向量数据库与检索Chroma LangChain Embeddings。用于存储和检索私有知识实现RAG检索增强生成能力。后端APIFastAPI。轻量、高性能适合快速构建AI服务接口。微调与优化LoRA, SFT。用于领域适配。工具调用自定义Tool用于查询外部API如模拟的股价接口。开发环境操作系统Ubuntu 20.04 / macOS 12 / Windows 10 (WSL2推荐)Python版本3.9 或 3.10CUDA11.8如需GPU推理可选包管理Conda pip版本依赖 (requirements.txt)# 核心框架 langchain0.1.0 langchain-community0.0.10 langchain-chroma0.1.0 # 模型与嵌入 transformers4.35.0 torch2.0.0 sentence-transformers2.2.2 qwen2-llm0.1.0 # 或通过 modelscope 安装 # 向量数据库 chromadb0.4.22 # Web框架与工具 fastapi0.104.0 uvicorn0.24.0 pydantic2.0.0 requests2.31.0 # 工具链 python-dotenv1.0.0 # 管理环境变量项目结构预览financial_qa_harness/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 应用入口 │ ├── harness/ # 核心驾驭工程模块 │ │ ├── __init__.py │ │ ├── orchestrator.py # 流程编排器核心 │ │ ├── agents/ # 智能体定义 │ │ │ ├── __init__.py │ │ │ ├── planner_agent.py │ │ │ └── query_agent.py │ │ ├── tools/ # 工具集 │ │ │ ├── __init__.py │ │ │ ├── knowledge_tool.py │ │ │ └── market_tool.py │ │ ├── memory/ # 记忆管理 │ │ │ ├── __init__.py │ │ │ └── conversation_memory.py │ │ └── evaluators/ # 评估与验证 │ │ ├── __init__.py │ │ └── safety_checker.py │ ├── models/ # 数据模型 │ │ ├── __init__.py │ │ └── schemas.py │ └── knowledge/ # 知识库相关 │ ├── __init__.py │ ├── vector_store.py # 向量库封装 │ └── docs/ # 存放知识文档 ├── config/ │ └── settings.py # 配置文件 ├── scripts/ │ ├── init_knowledge_base.py # 初始化知识库脚本 │ └── fine_tune.py # 微调脚本可选 ├── tests/ # 测试目录 ├── .env.example # 环境变量示例 ├── requirements.txt └── README.md3. 核心模块拆解Harness 系统的工程化设计在编写代码前我们必须理解系统如何被“驾驭”。我们的金融问答机器人将被设计成一个多智能体协作系统由编排器Orchestrator统一调度。3.1 编排器Orchestrator系统的大脑编排器是Harness系统的核心控制器它不直接生成答案而是负责任务分解、路由和流程控制。其工作流程如下接收用户查询。意图识别与分类判断问题属于通用知识、私有知识查询还是需要调用工具如查股价。任务规划将复杂问题拆解为子任务序列例如先检索知识再总结最后检查安全性。调度执行根据规划调用相应的智能体Agent或工具Tool执行子任务。结果合成与验证汇总各子任务结果进行逻辑一致性或安全性检查最终生成回复。这种设计将“思考”和“执行”分离提高了系统的可控性和可解释性。3.2 智能体Agent专业的执行者智能体是负责执行具体类型任务的模块。我们设计两类规划智能体Planner Agent擅长分析问题、制定步骤。它由一个大模型驱动接收用户问题输出一个JSON格式的任务计划。查询智能体Query Agent擅长执行信息检索与合成。它内部集成了RAG检索链专门处理需要从知识库中找答案的问题。3.3 工具Tools模型的手和脚工具是模型与外部世界交互的接口。我们将实现KnowledgeSearchTool封装向量数据库检索操作。GetStockPriceTool模拟调用外部金融API获取实时股价。 工具的使用由智能体根据编排器的规划来调用所有工具调用都有明确的输入输出格式和错误处理。3.4 记忆Memory与评估Evaluator记忆维护对话历史使机器人具备多轮对话能力。我们使用ConversationBufferWindowMemory来保存最近几轮的对话。评估器在最终输出前对内容进行安全检查过滤敏感或不恰当内容。4. 完整实战构建金融问答机器人Harness系统4.1 第一步初始化项目与环境创建项目目录并安装依赖。# 创建项目目录 mkdir financial_qa_harness cd financial_qa_harness # 创建虚拟环境以conda为例 conda create -n finance-qa python3.10 -y conda activate finance-qa # 安装依赖 pip install -r requirements.txt # 创建项目结构参考上面的目录树 mkdir -p app/harness/{agents,tools,memory,evaluators} app/models app/knowledge/docs config scripts tests设置环境变量文件.env# .env MODEL_PATHQwen/Qwen-7B-Chat # 可以是本地路径或HuggingFace模型ID EMBEDDING_MODELsentence-transformers/paraphrase-multilingual-MiniLM-L12-v2 CHROMA_DB_PATH./data/chroma_db API_HOST0.0.0.0 API_PORT80004.2 第二步构建知识库RAG基础首先准备一些金融知识文档TXT或PDF格式放入app/knowledge/docs/。然后编写向量库初始化脚本。文件app/knowledge/vector_store.pyfrom langchain_community.vectorstores import Chroma from langchain_community.embeddings import HuggingFaceEmbeddings from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_community.document_loaders import DirectoryLoader, TextLoader import os class FinancialKnowledgeStore: def __init__(self, persist_directory: str, embedding_model_name: str): self.persist_directory persist_directory self.embedding_model HuggingFaceEmbeddings(model_nameembedding_model_name) self.vector_store None def init_from_documents(self, docs_dir: str): 从文档目录初始化向量数据库 # 1. 加载文档 loader DirectoryLoader(docs_dir, glob**/*.txt, loader_clsTextLoader) documents loader.load() if not documents: print(未找到文档将创建空向量库。) # 创建空的向量库 self.vector_store Chroma( collection_namefinancial_knowledge, embedding_functionself.embedding_model, persist_directoryself.persist_directory ) return # 2. 分割文本 text_splitter RecursiveCharacterTextSplitter( chunk_size500, chunk_overlap50, separators[\n\n, \n, 。, , , ] ) splits text_splitter.split_documents(documents) print(f文档分割为 {len(splits)} 个片段。) # 3. 创建并持久化向量存储 self.vector_store Chroma.from_documents( documentssplits, embeddingself.embedding_model, persist_directoryself.persist_directory, collection_namefinancial_knowledge ) print(f知识库已初始化并保存至 {self.persist_directory}) def get_retriever(self, search_kwargs: dict {k: 3}): 获取检索器 if self.vector_store is None: # 如果之前已持久化则加载现有库 self.vector_store Chroma( collection_namefinancial_knowledge, embedding_functionself.embedding_model, persist_directoryself.persist_directory ) return self.vector_store.as_retriever(search_kwargssearch_kwargs) # 使用示例 if __name__ __main__: import sys sys.path.append(..) from config.settings import settings knowledge_store FinancialKnowledgeStore( persist_directorysettings.CHROMA_DB_PATH, embedding_model_namesettings.EMBEDDING_MODEL ) knowledge_store.init_from_documents(./app/knowledge/docs/)运行初始化脚本python app/knowledge/vector_store.py。4.3 第三步实现工具Tools文件app/harness/tools/knowledge_tool.pyfrom langchain.tools import BaseTool from pydantic import BaseModel, Field from typing import Type, Optional from app.knowledge.vector_store import FinancialKnowledgeStore from config.settings import settings class KnowledgeSearchInput(BaseModel): query: str Field(description用于检索知识库的查询语句) class KnowledgeSearchTool(BaseTool): name financial_knowledge_search description 在金融知识库中搜索与问题相关的信息。当用户询问金融概念、产品详情、公司政策等静态知识时使用此工具。 args_schema: Type[BaseModel] KnowledgeSearchInput def __init__(self): super().__init__() self.knowledge_store FinancialKnowledgeStore( persist_directorysettings.CHROMA_DB_PATH, embedding_model_namesettings.EMBEDDING_MODEL ) self.retriever self.knowledge_store.get_retriever(search_kwargs{k: 3}) def _run(self, query: str) - str: 执行检索 try: docs self.retriever.get_relevant_documents(query) if not docs: return 在知识库中未找到相关信息。 # 拼接检索到的文档内容 context \n\n---\n\n.join([doc.page_content for doc in docs]) return f从知识库中检索到以下相关信息\n{context} except Exception as e: return f知识库检索过程中发生错误{str(e)}文件app/harness/tools/market_tool.pyfrom langchain.tools import BaseTool from pydantic import BaseModel, Field from typing import Type import requests import json class StockPriceInput(BaseModel): symbol: str Field(description股票代码例如AAPL, 000001.SZ) class GetStockPriceTool(BaseTool): name get_stock_price description 获取指定股票的实时或最新价格。输入应为股票代码。 args_schema: Type[BaseModel] StockPriceInput def _run(self, symbol: str) - str: 模拟调用股票API。实际应用中应替换为真实的API调用。 # 这里模拟一个API响应 mock_data { AAPL: {price: 172.35, change: 1.23, currency: USD}, 000001.SZ: {price: 15.42, change: -0.08, currency: CNY}, TSLA: {price: 175.79, change: -3.21, currency: USD}, } if symbol.upper() in mock_data: data mock_data[symbol.upper()] return json.dumps({ symbol: symbol, price: data[price], change: data[change], currency: data[currency], note: 此为模拟数据实际使用时请接入雅虎财经、Alpha Vantage等API。 }, ensure_asciiFalse) else: return json.dumps({error: f未找到股票代码 {symbol} 的模拟数据。}, ensure_asciiFalse)4.4 第四步实现智能体Agents文件app/harness/agents/planner_agent.pyfrom langchain.agents import AgentExecutor, create_structured_chat_agent from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain.schema import SystemMessage from langchain_community.chat_models import ChatQwen # 假设有适配Qwen的ChatModel from config.settings import settings import json class PlannerAgent: def __init__(self): # 初始化模型 self.llm ChatQwen( model_namesettings.MODEL_PATH, temperature0.1, # 低温度保证规划稳定性 ) self.prompt self._create_prompt() self.agent_executor None def _create_prompt(self): 创建规划智能体的提示词 system_message SystemMessage(content你是一个任务规划专家。你的工作是将用户的金融问题分解成一个清晰的、可执行的JSON格式计划。 计划类型包括 1. knowledge_query: 需要从金融知识库中检索信息。 2. market_data: 需要查询实时市场数据如股价。 3. general_qa: 通用金融问答无需额外工具。 4. multi_step: 复杂问题需要组合多个步骤。 请严格按照以下JSON格式输出不要添加任何额外解释 { plan_type: knowledge_query | market_data | general_qa | multi_step, steps: [ {step_type: knowledge_search, query: 具体查询语句}, {step_type: market_lookup, symbol: 股票代码}, ... // 其他步骤 ], final_synthesis_instruction: 如何将各步骤结果合成最终答案的简要说明 } 如果问题很简单steps可以为空列表。) prompt ChatPromptTemplate.from_messages([ system_message, MessagesPlaceholder(variable_namechat_history, optionalTrue), (human, {input}), MessagesPlaceholder(variable_nameagent_scratchpad), ]) return prompt def plan(self, user_input: str, chat_history: list None) - dict: 执行规划返回计划字典 if self.agent_executor is None: # 规划智能体不需要外部工具 from langchain.agents import AgentType self.agent_executor AgentExecutor.from_agent_and_tools( agentcreate_structured_chat_agent( llmself.llm, tools[], # 规划器不直接使用工具 promptself.prompt ), tools[], verbosesettings.DEBUG, handle_parsing_errorsTrue, ) try: result self.agent_executor.invoke({ input: user_input, chat_history: chat_history or [] }) plan_str result[output] # 解析JSON输出 return json.loads(plan_str) except json.JSONDecodeError: print(f规划器输出无法解析为JSON: {plan_str}) # 降级处理返回一个默认的通用问答计划 return {plan_type: general_qa, steps: [], final_synthesis_instruction: 直接回答用户问题。} except Exception as e: print(f规划执行出错: {e}) return {plan_type: general_qa, steps: [], final_synthesis_instruction: 直接回答用户问题。}文件app/harness/agents/query_agent.pyfrom langchain.chains import RetrievalQA from langchain.prompts import PromptTemplate from app.knowledge.vector_store import FinancialKnowledgeStore from config.settings import settings class QueryAgent: def __init__(self): self.knowledge_store FinancialKnowledgeStore( persist_directorysettings.CHROMA_DB_PATH, embedding_model_namesettings.EMBEDDING_MODEL ) self.retriever self.knowledge_store.get_retriever() # 初始化一个用于RAG的LLM可以与规划器使用同一个但温度可以不同 from langchain_community.chat_models import ChatQwen self.llm ChatQwen(model_namesettings.MODEL_PATH, temperature0.2) # 定义RAG提示词模板 self.qa_prompt PromptTemplate( input_variables[context, question], template你是一个专业的金融顾问请严格根据以下提供的上下文信息来回答问题。如果上下文信息不足以回答问题请如实说明你不知道不要编造信息。 上下文 {context} 问题{question} 请提供专业、准确、简洁的回答 ) self.qa_chain RetrievalQA.from_chain_type( llmself.llm, chain_typestuff, retrieverself.retriever, chain_type_kwargs{prompt: self.qa_prompt}, return_source_documentsTrue ) def answer_with_knowledge(self, question: str) - dict: 基于知识库回答问题 result self.qa_chain.invoke({query: question}) return { answer: result[result], source_documents: [doc.page_content[:200] ... for doc in result.get(source_documents, [])] # 截取部分内容 }4.5 第五步实现核心编排器Orchestrator文件app/harness/orchestrator.pyfrom typing import Dict, Any, List from app.harness.agents.planner_agent import PlannerAgent from app.harness.agents.query_agent import QueryAgent from app.harness.tools.knowledge_tool import KnowledgeSearchTool from app.harness.tools.market_tool import GetStockPriceTool from app.harness.memory.conversation_memory import ConversationMemoryManager from app.harness.evaluators.safety_checker import SafetyChecker from config.settings import settings import json class Orchestrator: 核心编排器负责协调所有组件完成用户查询。 def __init__(self): self.planner PlannerAgent() self.query_agent QueryAgent() self.knowledge_tool KnowledgeSearchTool() self.market_tool GetStockPriceTool() self.memory_manager ConversationMemoryManager() self.safety_checker SafetyChecker() # 可以初始化一个通用的LLM用于最终合成或简单问答 from langchain_community.chat_models import ChatQwen self.general_llm ChatQwen(model_namesettings.MODEL_PATH, temperature0.7) def process_query(self, user_input: str, session_id: str default) - Dict[str, Any]: 处理用户查询的主流程 # 1. 获取对话历史 chat_history self.memory_manager.get_history(session_id) # 2. 规划阶段决定如何解决问题 plan self.planner.plan(user_input, chat_history) print(f[Orchestrator] 生成计划: {json.dumps(plan, indent2, ensure_asciiFalse)}) # 3. 执行阶段根据计划调用相应模块 intermediate_results [] if plan[plan_type] general_qa: # 通用问答直接调用LLM response self.general_llm.invoke(chat_history [(human, user_input)]) final_answer response.content elif plan[plan_type] knowledge_query: # 知识库问答 result self.query_agent.answer_with_knowledge(user_input) final_answer result[answer] intermediate_results.append({type: knowledge, sources: result[source_documents]}) elif plan[plan_type] market_data: # 市场数据查询需要从问题中提取股票代码这里简化处理 # 实际应用中可以用一个NLU模块来提取实体 symbol self._extract_stock_symbol(user_input) if symbol: tool_result self.market_tool.run(symbol) intermediate_results.append({type: market_data, raw_result: tool_result}) # 让LLM将工具结果转化为自然语言回答 prompt f用户询问股票 {symbol} 的价格。工具返回的数据是{tool_result}。请生成一段友好的中文回复给用户。 response self.general_llm.invoke([(human, prompt)]) final_answer response.content else: final_answer 抱歉我无法从您的问题中识别出有效的股票代码。请提供如AAPL或000001.SZ这样的代码。 elif plan[plan_type] multi_step: # 多步骤执行示例先查知识再查股价 final_answer 执行复杂计划... for step in plan.get(steps, []): if step[step_type] knowledge_search: tool_result self.knowledge_tool.run(step[query]) intermediate_results.append({type: knowledge, raw_result: tool_result}) elif step[step_type] market_lookup: tool_result self.market_tool.run(step[symbol]) intermediate_results.append({type: market_data, raw_result: tool_result}) # 根据计划中的指令使用LLM合成最终答案 synthesis_prompt f你收到了以下中间结果 {json.dumps(intermediate_results, ensure_asciiFalse, indent2)} 请根据这些信息并遵循这个指令来回答用户的问题“{user_input}” 指令{plan.get(final_synthesis_instruction, 综合所有信息进行回答。)} response self.general_llm.invoke([(human, synthesis_prompt)]) final_answer response.content else: final_answer 抱歉我暂时无法处理这类问题。 # 4. 安全与合规检查 is_safe, checked_answer self.safety_checker.check(final_answer) if not is_safe: checked_answer 您的问题或我的回答可能涉及不适宜的内容我已进行过滤。请尝试询问其他金融相关问题。 # 5. 更新对话记忆 self.memory_manager.add_interaction(session_id, user_input, checked_answer) # 6. 组装返回结果 return { answer: checked_answer, plan: plan, intermediate_results: intermediate_results, session_id: session_id } def _extract_stock_symbol(self, text: str) - str: 简单的股票代码提取函数示例实际应用需更复杂的NLP import re # 简单匹配大写字母代码或数字代码 patterns [ r\b([A-Z]{2,5})\b, # 如 AAPL, TSLA r\b(\d{6}\.[A-Z]{2})\b, # 如 000001.SZ ] for pattern in patterns: match re.search(pattern, text.upper()) if match: return match.group(1) return 4.6 第六步构建API服务与运行文件app/main.pyfrom fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import Optional from app.harness.orchestrator import Orchestrator import uvicorn from config.settings import settings app FastAPI(title金融大模型问答机器人 API, version1.0.0) orchestrator Orchestrator() # 全局初始化实际生产需考虑生命周期和并发 class QueryRequest(BaseModel): question: str session_id: Optional[str] default_user class QueryResponse(BaseModel): answer: str session_id: str plan_type: Optional[str] None success: bool True app.post(/api/query, response_modelQueryResponse) async def query_financial_bot(request: QueryRequest): 处理用户查询的主接口 try: result orchestrator.process_query(request.question, request.session_id) return QueryResponse( answerresult[answer], session_idresult[session_id], plan_typeresult[plan].get(plan_type), successTrue ) except Exception as e: raise HTTPException(status_code500, detailf服务器内部错误: {str(e)}) app.get(/health) async def health_check(): return {status: healthy, service: financial_qa_bot} if __name__ __main__: uvicorn.run( app.main:app, hostsettings.API_HOST, portsettings.API_PORT, reloadsettings.DEBUG )配置文件config/settings.pyimport os from pydantic_settings import BaseSettings from dotenv import load_dotenv load_dotenv() class Settings(BaseSettings): MODEL_PATH: str os.getenv(MODEL_PATH, Qwen/Qwen-7B-Chat) EMBEDDING_MODEL: str os.getenv(EMBEDDING_MODEL, sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2) CHROMA_DB_PATH: str os.getenv(CHROMA_DB_PATH, ./data/chroma_db) API_HOST: str os.getenv(API_HOST, 0.0.0.0) API_PORT: int int(os.getenv(API_PORT, 8000)) DEBUG: bool os.getenv(DEBUG, False).lower() true settings Settings()运行服务# 确保知识库已初始化 python scripts/init_knowledge_base.py # 启动FastAPI服务 python -m uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload测试APIcurl -X POST http://localhost:8000/api/query \ -H Content-Type: application/json \ -d {question: 什么是市盈率, session_id: test_user}5. 常见问题与排查思路在构建和运行此类Harness系统时你可能会遇到以下典型问题问题现象可能原因排查思路与解决方案模型加载失败或响应慢1. 模型路径错误。2. 显存不足。3. 网络问题从HF下载。1. 检查MODEL_PATH是本地路径还是HF ID确保路径正确。2. 使用nvidia-smi查看显存考虑使用量化模型如Qwen-7B-Chat-Int4或CPU推理。3. 设置环境变量HF_ENDPOINThttps://hf-mirror.com使用镜像。知识库检索不到相关内容1. 文档未成功加载或分割。2. 嵌入模型不匹配或效果差。3. 检索参数k太小。1. 检查docs/目录下是否有.txt文件运行初始化脚本看日志。2. 尝试更换嵌入模型如BAAI/bge-large-zh-v1.5。3. 在vector_store.py中调整search_kwargs{k: 5}。工具调用失败或格式错误1. Tool的args_schema定义与模型输出不匹配。2. 模型未能正确理解何时调用工具。1. 确保description字段清晰并启用Agent的verboseTrue模式观察模型思考过程。2. 优化规划器或Agent的提示词Prompt明确工具的使用条件和输入格式。多轮对话记忆丢失1.session_id未正确传递。2. 记忆存储未持久化服务重启后丢失。1. 确保前端或调用方每次请求传入相同的session_id。2. 将ConversationBufferWindowMemory替换为持久化后端如RedisChatMessageHistory。回答出现“幻觉”或无关内容1. RAG检索的相关性低。2. 合成答案的LLM温度过高。3. 未严格限制模型仅基于上下文回答。1. 优化检索器尝试不同嵌入模型、调整chunk大小、使用重排序器。2. 降低query_agent中LLM的temperature如设为0.1。3. 强化提示词模板加入“严格基于上下文”的指令。服务并发请求出错1. 全局共享的orchestrator或LLM实例非线程安全。2. GPU模型推理无法处理并发。1. 使用依赖注入如FastAPI的Depends为每个请求创建新的组件实例或使用线程锁。2. 考虑使用模型推理API服务如TGI vLLM或请求队列。6. 最佳实践与工程建议将Harness Engineering思想落地到生产环境除了核心功能还需关注以下工程化细节1. 配置化管理将所有模型路径、API密钥、超时参数、开关配置如是否启用安全审查集中到配置文件如settings.py或配置中心。使用pydantic-settings进行验证和环境变量注入。2. 可观测性与日志结构化日志使用structlog或jsonlogger记录关键事件请求入参、规划结果、工具调用、最终回答、耗时。链路追踪为每个用户请求生成唯一request_id贯穿所有组件规划、检索、工具调用、LLM生成便于问题追踪。监控指标记录请求量、响应延迟、Token消耗、各阶段耗时、缓存命中率、工具调用成功率等。3. 稳定性与容错组件超时与重试为LLM调用、外部工具API设置超时和重试机制。降级策略当知识库检索失败时降级为通用LLM回答并明确告知用户当规划器失败时使用默认的通用问答流程。输入输出验证对用户输入进行清洗和长度限制对模型输出进行格式验证防止后续流程解析失败。4. 性能优化缓存对频繁且结果不变的查询如某些知识检索、股票价格实施缓存Redis。模型量化与加速生产环境使用量化模型GPTQ, AWQ或推理加速框架vLLM, TensorRT-LLM。异步处理对于IO密集型操作如网络请求、磁盘读取使用异步编程async/await。5. 安全与合规输入过滤防止Prompt注入攻击对用户输入进行关键词过滤和意图合法性检查。输出审查必须集成内容安全过滤器如本项目的SafetyChecker防止生成有害、偏见或敏感信息。权限控制不同用户或租户可能访问不同的知识库或工具需要在编排器层设计权限路由。审计日志记录所有用户查询和系统回答满足合规审计要求。6. 迭代与评估评估流水线构建离线评估集定期测试系统在“事实准确性”、“回答相关性”、“安全性”等维度的表现。A/B测试对新版本的规划策略、提示词或模型进行A/B测试用数据驱动决策。反馈闭环提供用户反馈入口如“回答是否有用”将反馈数据用于优化检索和排序。7. 总结与扩展方向通过本项目我们实践了Harness Engineering的核心思想不是单纯依赖一个大模型而是通过精巧的工程化设计编排器、多智能体、工具、记忆、评估来“驾驭”它构建出一个稳定、可控、可扩展的AI应用系统。你已掌握的核心技能点Harness系统架构设计理解了编排器、智能体、工具的分层协作模式。RAG完整实现从文档加载、文本分割、向量化存储到检索增强生成的闭环。LangChain智能体与工具集成如何利用LangChain框架快速构建可工具调用的AI智能体。工程化项目结构配置管理、模块化设计、API服务封装。基础的安全与稳定性考量输入输出检查、错误处理、降级策略。下一步可以深入探索的方向更复杂的规划与推理集成ReAct、Chain of Thought等推理框架让智能体处理更复杂的多步骤问题。流式输出与前端集成将API改造为流式响应Server-Sent Events并构建一个交互式Web前端。领域微调SFT/LoRA使用金融领域的对话数据对Qwen模型进行微调提升其在专业领域的表现。图检索增强GraphRAG引入知识图谱让机器人不仅能检索片段还能理解实体间的复杂关系。多模态扩展接入视觉模型使其能解读财报中的图表信息。项目的完整代码已结构化呈现你可以从app/harness/orchestrator.py这个核心文件开始阅读逐步理解数据流与控制流。在实际部署时请务必仔细测试每个环节并从简单的场景开始逐步增加复杂度。记住一个好的Harness系统是在持续迭代和观察中打磨出来的。