如果你正在学习AI大模型应用开发或者想从零开始构建一个AI Agent项目那么“Harness”这个词可能已经在你眼前出现了无数次。但你是否真的理解它到底是什么为什么它突然变得如此重要更重要的是当你面对一个“金融大模型问答机器人”这样的项目需求时Harness能帮你解决哪些具体问题很多人误以为Harness只是一个简单的“包装器”或“脚手架”把大模型API调用一下就算完事。这种理解会让你在实际项目中处处碰壁。真正的Harness Engineering解决的是从“一个能跑通的Demo”到“一个稳定、可靠、可维护的生产级AI应用”之间的巨大鸿沟。它涉及能力分层、模块边界、核心抽象、扩展机制甚至团队协作的权限模型。本文将从零开始手把手带你搞懂Harness的核心原理并通过一个完整的“金融大模型问答机器人”项目实战让你亲身体验如何用Harness工程化思维将一个AI想法落地为可交付的项目。无论你是零基础转行AI的开发者还是有一定经验但被AI项目混乱的代码结构所困扰这篇文章都将为你提供一条清晰的路径。1. Harness Engineering为什么它比“调API”重要100倍在AI大模型应用开发中最危险的错觉就是认为“调用API完成开发”。你写了几行代码调通了OpenAI的接口看到了返回结果感觉大功告成。但当你试图把这个“Demo”嵌入到真实的金融业务系统时问题接踵而至问题一提示词Prompt管理混乱。业务规则一变提示词就要改。你是把几百行的提示词模板硬编码在代码里还是写在配置文件里不同场景客服、研报解读、风险预警的提示词如何复用和组合问题二模型切换成本高。今天用GPT-4明天因为成本或响应速度想换为Claude或国产大模型。你需要重写多少代码如何保证不同模型输出格式的统一问题三缺乏业务逻辑与AI能力的隔离。你的核心业务逻辑如查询账户余额、计算投资组合风险和调用大模型的代码纠缠在一起。一旦需要升级或替换AI部分整个系统都得动。问题四可观测性Observability几乎为零。大模型为什么给出了某个回答它的思考过程Chain-of-Thought是什么每次调用的耗时、Token消耗、成功率是多少没有这些数据你根本无法进行性能优化和问题排查。问题五测试与评估Evaluation无从下手。如何自动化测试AI输出的准确性和稳定性难道每次都要人工看吗Harness Engineering驾驭工程的核心价值正是为了解决上述问题。它不是某个特定的框架如LangChain而是一套工程方法论和最佳实践的集合旨在为AI能力构建一个标准化的“接入层”和“控制层”。你可以把Harness想象成汽车的方向盘、油门和刹车系统Harness。发动机大模型本身威力巨大但如果没有这套驾驭系统汽车根本无法安全、可控地行驶。Harness就是让你能安全、高效、可控地“驾驭”大模型能力的那套工程体系。在接下来的内容中我们将围绕一个金融大模型问答机器人项目拆解Harness工程的关键组成部分并给出可运行的代码。2. 项目定义我们要构建一个什么样的系统在开始编码之前必须明确目标。我们的项目案例描述如下项目名称金融大模型问答机器人核心职责作为AI大模型应用开发工程师你需要设计并实现一个机器人能够理解用户关于金融领域的自然语言提问例如“帮我分析一下腾讯控股最近一年的股价走势”或“什么是ETF”并给出准确、可靠的回答。项目设计目标准确性对于事实类问题如定义、规则回答必须精确。安全性绝不能给出投资建议或做出财务预测合规要求。可扩展性易于接入新的数据源如实时行情API、财经新闻和新的AI模型。可维护性代码结构清晰提示词、模型配置、业务逻辑分离。3. Harness工程核心四层架构一个典型的Harness驱动AI应用可以抽象为以下四层。理解这个分层是做好一切设计的基础。层级职责对应项目中的组件关键问题1. 应用层 (Application Layer)面向用户的接口和核心业务流程。Web API (FastAPI/SpringBoot)、任务调度器。如何组织业务流水线如何将用户请求转化为AI可处理的任务2. 编排层 (Orchestration Layer)协调多个步骤或工具决定工作流。Harness 核心 定义Plan、Skill、Agent的执行逻辑。是先查数据库还是先问大模型如何根据答案决定下一步动作3. 能力层 (Capability Layer)提供具体的原子能力如调用模型、查询数据库、执行计算。Model Skill,Data Query Skill,Calculator Skill。如何统一不同能力模型、工具的调用接口如何管理它们的配置4. 连接层 (Connection Layer)管理与外部服务的连接和认证。大模型API密钥管理、数据库连接池、第三方财经API客户端。密钥如何安全存储连接如何复用和监控在我们的金融机器人中用户问题“分析腾讯股价”会这样流动应用层(接收HTTP请求) -编排层(制定计划先查询股价数据再让模型分析) -能力层(执行Data Skill查数据Model Skill发分析请求) -连接层(调用真实API)。4. 环境准备与项目初始化我们选择Python作为实现语言因为它在大模型生态中工具最丰富。项目将使用poetry进行依赖管理也可以用piprequirements.txt。4.1 基础环境Python版本: 3.9 或 3.103.11请留意某些包的兼容性操作系统: macOS / Linux / Windows (WSL2推荐)IDE: VSCode 或 PyCharm4.2 创建项目并安装核心依赖# 1. 创建项目目录 mkdir finance-ai-agent cd finance-ai-agent # 2. 初始化 poetry 项目 (如果没有poetry请先安装: pip install poetry) poetry init -n # 交互式创建或直接使用下面的 pyproject.toml # 3. 编辑 pyproject.toml添加依赖以下是pyproject.toml文件的内容示例# pyproject.toml [tool.poetry] name finance-ai-agent version 0.1.0 description A harness-engineered financial QA AI agent. authors [Your Name youexample.com] [tool.poetry.dependencies] python ^3.9 openai ^1.0.0 # 使用OpenAI官方新版SDK langchain ^0.1.0 # 可选这里我们主要展示自有Harness设计但可借鉴其思想 pydantic ^2.0.0 # 用于数据验证和设置管理 fastapi ^0.104.0 # 构建API接口 uvicorn ^0.24.0 # ASGI服务器 httpx ^0.25.0 # 异步HTTP客户端 python-dotenv ^1.0.0 # 管理环境变量 sqlalchemy ^2.0.0 # ORM (示例用) pandas ^2.0.0 # 数据处理 (示例用) [tool.poetry.group.dev.dependencies] pytest ^7.4.0 black ^23.0.0 isort ^5.12.0 [build-system] requires [poetry-core] build-backend poetry.core.masonry.api# 4. 安装依赖 poetry install # 5. 激活虚拟环境 poetry shell4.3 配置环境变量创建.env文件来存储敏感信息切记不要提交到版本控制系统。# .env OPENAI_API_KEYsk-your-openai-api-key-here # 未来可以扩展其他模型 ANTHROPIC_API_KEYyour-claude-key DATABASE_URLsqlite:///./finance.db # 示例用SQLite FINANCE_DATA_API_KEYyour-dummy-finance-api-key5. 核心模块设计与实现构建我们的Harness我们将自底向上实现Harness的各层。首先从最稳定的连接层和能力层开始。5.1 连接层统一的外部服务客户端创建core/clients.py:# core/clients.py import os from typing import Optional from openai import AsyncOpenAI import httpx from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from pydantic_settings import BaseSettings class Settings(BaseSettings): openai_api_key: str database_url: str sqlite:///./finance.db finance_api_base: str https://api.example-finance.com/v1 # 示例地址 class Config: env_file .env settings Settings() # 1. 大模型客户端 (以OpenAI为例) class OpenAIClient: _instance: Optional[AsyncOpenAI] None classmethod def get_client(cls) - AsyncOpenAI: if cls._instance is None: cls._instance AsyncOpenAI(api_keysettings.openai_api_key) return cls._instance # 2. 财经数据API客户端 (抽象类) class FinanceDataClient: def __init__(self): self.base_url settings.finance_api_base self.api_key os.getenv(FINANCE_DATA_API_KEY) self._async_client httpx.AsyncClient( base_urlself.base_url, headers{Authorization: fBearer {self.api_key}}, timeout30.0 ) async def get_stock_price(self, symbol: str, period: str 1y) - dict: 获取股票价格数据模拟 # 实际项目中这里会调用真实API如Yahoo Finance, Alpha Vantage等 # 此处返回模拟数据 return { symbol: symbol, period: period, data: [{date: 2023-01-01, close: 350.0}, {date: 2023-12-01, close: 400.0}], source: simulated } async def close(self): await self._async_client.aclose() # 3. 数据库会话工厂 engine create_engine(settings.database_url, connect_args{check_same_thread: False} if sqlite in settings.database_url else {}) SessionLocal sessionmaker(autocommitFalse, autoflushFalse, bindengine)5.2 能力层定义可复用的SkillSkill是Harness中的原子能力单元。创建skills/base_skill.py和具体Skill。# skills/base_skill.py from abc import ABC, abstractmethod from typing import Any, Dict from pydantic import BaseModel class SkillInput(BaseModel): Skill的输入参数基类 pass class SkillOutput(BaseModel): Skill的输出结果基类 success: bool data: Any error: Optional[str] None class BaseSkill(ABC): 所有Skill的抽象基类 name: str description: str abstractmethod async def execute(self, input_data: SkillInput) - SkillOutput: 执行技能的核心方法 pass现在实现两个具体的Skill调用大模型的ModelSkill和查询金融数据的DataQuerySkill。# skills/model_skill.py from skills.base_skill import BaseSkill, SkillInput, SkillOutput from core.clients import OpenAIClient from typing import Optional from pydantic import Field class ModelSkillInput(SkillInput): prompt: str model: str gpt-3.5-turbo temperature: float 0.7 max_tokens: Optional[int] None class ModelSkillOutput(SkillOutput): completion: Optional[str] None class ModelSkill(BaseSkill): name model_completion description 调用大语言模型生成文本 async def execute(self, input_data: ModelSkillInput) - ModelSkillOutput: client OpenAIClient.get_client() try: response await client.chat.completions.create( modelinput_data.model, messages[{role: user, content: input_data.prompt}], temperatureinput_data.temperature, max_tokensinput_data.max_tokens ) completion response.choices[0].message.content return ModelSkillOutput( successTrue, data{raw_response: response}, completioncompletion ) except Exception as e: return ModelSkillOutput( successFalse, dataNone, errorfModel调用失败: {str(e)} )# skills/data_query_skill.py from skills.base_skill import BaseSkill, SkillInput, SkillOutput from core.clients import FinanceDataClient from pydantic import Field class DataQuerySkillInput(SkillInput): query_type: str Field(..., description查询类型如 stock_price, news) parameters: Dict[str, Any] Field(default_factorydict) class DataQuerySkill(BaseSkill): name financial_data_query description 查询金融数据股价、新闻等 def __init__(self): self.client FinanceDataClient() async def execute(self, input_data: DataQuerySkillInput) - SkillOutput: try: if input_data.query_type stock_price: symbol input_data.parameters.get(symbol) period input_data.parameters.get(period, 1y) data await self.client.get_stock_price(symbol, period) return SkillOutput(successTrue, datadata) else: return SkillOutput( successFalse, dataNone, errorf不支持的查询类型: {input_data.query_type} ) except Exception as e: return SkillOutput(successFalse, dataNone, errorstr(e)) async def cleanup(self): 清理资源如关闭HTTP连接 await self.client.close()5.3 编排层制定与执行计划Plan编排层是Harness的大脑。它根据用户请求决定调用哪些Skill以及调用的顺序。我们实现一个简单的Planner和PlanExecutor。# orchestration/planner.py from typing import List, Dict, Any from pydantic import BaseModel from skills.base_skill import BaseSkill class PlanStep(BaseModel): skill_name: str skill_input: Dict[str, Any] depends_on: List[int] [] # 依赖的步骤索引 class Plan(BaseModel): steps: List[PlanStep] final_output_key: str # 最终结果从哪个步骤获取 class Planner: 根据用户意图制定执行计划 def __init__(self, available_skills: Dict[str, BaseSkill]): self.skills available_skills async def create_plan(self, user_query: str) - Plan: 核心规划逻辑分析用户问题生成执行步骤 # 这里可以非常简单也可以非常复杂甚至可以引入一个小模型来做规划 # 示例简单规则匹配 query_lower user_query.lower() if any(word in query_lower for word in [股价, stock, price, 走势]): # 识别出股票代码 (这里用简单正则实际应用需要更复杂的NLP) import re symbol_match re.search(r[A-Z]{1,5}, user_query.upper()) # 简单匹配美股代码 symbol symbol_match.group(0) if symbol_match else AAPL # 默认苹果 return Plan( steps[ PlanStep( skill_namefinancial_data_query, skill_input{query_type: stock_price, parameters: {symbol: symbol, period: 1y}} ), PlanStep( skill_namemodel_completion, skill_input{ prompt: f用户询问了{symbol}的股价走势。这里有一些历史数据{{stock_data}}。请用中文总结一下趋势但切记不要给出任何投资建议。, model: gpt-3.5-turbo }, depends_on[0] # 依赖于第0步的结果 ) ], final_output_key1 # 最终输出是第1步模型分析的结果 ) elif 是什么 in query_lower or 定义 in query_lower: # 纯概念性问题直接问模型 return Plan( steps[ PlanStep( skill_namemodel_completion, skill_input{ prompt: f用户问{user_query}。请用中文清晰、准确地解释这个概念如果涉及金融请确保解释符合常规金融定义。, model: gpt-3.5-turbo, temperature: 0.3 # 事实性问题降低随机性 } ) ], final_output_key0 ) else: # 默认兜底计划 return Plan( steps[ PlanStep( skill_namemodel_completion, skill_input{ prompt: f用户说{user_query}。请用中文友好地回应并说明你是一个金融信息问答助手无法提供个人投资建议。, model: gpt-3.5-turbo } ) ], final_output_key0 )# orchestration/executor.py from typing import Dict, Any from .planner import Plan, PlanStep from skills.base_skill import BaseSkill class PlanExecutor: 执行计划处理步骤间的依赖 def __init__(self, skill_registry: Dict[str, BaseSkill]): self.skill_registry skill_registry self.step_results: Dict[int, Any] {} async def execute_plan(self, plan: Plan) - Dict[str, Any]: 按依赖顺序执行计划中的所有步骤 # 拓扑排序执行简化版按顺序执行依赖通过占位符传递 for i, step in enumerate(plan.steps): # 准备输入替换依赖占位符 resolved_input self._resolve_input(step.skill_input, step.depends_on) # 获取对应的Skill skill self.skill_registry.get(step.skill_name) if not skill: raise ValueError(f未注册的Skill: {step.skill_name}) # 执行Skill output await skill.execute(resolved_input) if not output.success: # 错误处理可以重试、跳过或终止计划 raise Exception(f步骤 {i} ({step.skill_name}) 执行失败: {output.error}) # 存储结果 self.step_results[i] output.data # 特殊处理如果Skill输出有completion字段也存下来方便后续使用 if hasattr(output, completion) and output.completion: self.step_results[f{i}_completion] output.completion # 返回最终结果 final_result self.step_results.get(int(plan.final_output_key)) final_completion self.step_results.get(f{plan.final_output_key}_completion) return { final_data: final_result, final_completion: final_completion, all_step_results: self.step_results } def _resolve_input(self, skill_input: Dict[str, Any], depends_on: List[int]) - Dict[str, Any]: 解析输入中的占位符例如 {{stock_data}} import json resolved skill_input.copy() # 将整个字典转为字符串方便替换 input_str json.dumps(resolved) for dep_index in depends_on: dep_result self.step_results.get(dep_index) if dep_result: # 简单替换占位符实际项目需要更健壮的模板引擎 placeholder f{{{{step_{dep_index}}}}} # 这里我们简单地将依赖步骤的数据作为字符串插入。 # 更复杂的实现可以支持结构化数据的注入。 input_str input_str.replace(placeholder, json.dumps(dep_result)) return json.loads(input_str)5.4 应用层构建API服务最后我们用FastAPI将上述所有层整合起来暴露一个Web API。# main.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from contextlib import asynccontextmanager from core.clients import SessionLocal from skills.model_skill import ModelSkill from skills.data_query_skill import DataQuerySkill from orchestration.planner import Planner from orchestration.executor import PlanExecutor # 定义请求/响应模型 class QueryRequest(BaseModel): question: str user_id: str anonymous # 可用于审计和个性化 class QueryResponse(BaseModel): success: bool answer: str plan_id: str None # 可用于追踪 debug_info: dict {} # 开发阶段可返回调试信息 # 全局资源管理 skill_registry {} planner None executor None asynccontextmanager async def lifespan(app: FastAPI): # 启动逻辑 print(启动AI Agent服务...) # 1. 注册所有Skill global skill_registry, planner, executor skill_registry[model_completion] ModelSkill() skill_registry[financial_data_query] DataQuerySkill() # 2. 初始化编排器 planner Planner(skill_registry) executor PlanExecutor(skill_registry) yield # 关闭逻辑 print(关闭AI Agent服务...) # 清理资源如关闭数据库连接、HTTP客户端 data_skill skill_registry.get(financial_data_query) if data_skill: await data_skill.cleanup() app FastAPI(title金融AI问答机器人, lifespanlifespan) app.post(/query, response_modelQueryResponse) async def handle_query(request: QueryRequest): 处理用户查询的核心端点 try: # 1. 制定计划 plan await planner.create_plan(request.question) # 2. 执行计划 result await executor.execute_plan(plan) # 3. 构建响应 answer result.get(final_completion, 抱歉我暂时无法处理这个问题。) return QueryResponse( successTrue, answeranswer, debug_info{plan_steps: len(plan.steps)} # 生产环境应关闭 ) except Exception as e: # 记录日志 print(f处理查询时出错: {e}) raise HTTPException(status_code500, detailf内部服务器错误: {str(e)}) app.get(/health) async def health_check(): return {status: healthy} if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)6. 运行与测试6.1 启动服务确保在项目根目录下且虚拟环境已激活.env文件中的OPENAI_API_KEY已正确设置。python main.py服务将在http://localhost:8000启动。6.2 测试API使用curl或任何API测试工具如Postman进行测试。# 测试健康检查 curl http://localhost:8000/health # 测试概念性问题 curl -X POST http://localhost:8000/query \ -H Content-Type: application/json \ -d {question: 什么是ETF} # 测试股价分析模拟数据 curl -X POST http://localhost:8000/query \ -H Content-Type: application/json \ -d {question: AAPL的股价最近一年走势如何}6.3 预期输出对于“什么是ETF”的查询你应该会收到一个结构化的JSON响应其中answer字段包含大模型生成的解释文本并且会强调其作为金融信息助手不提供投资建议的立场。7. 常见问题与排查思路问题现象可能原因排查方式解决方案服务启动失败提示ModuleNotFoundError依赖未安装或虚拟环境未激活。检查poetry install是否成功确认终端处于poetry shell激活的虚拟环境中。重新运行poetry install并激活环境。API调用返回401或Invalid API KeyOpenAI API密钥错误或未设置。检查.env文件中的OPENAI_API_KEY确保其正确且未被空格包围。在OpenAI官网重新生成API Key并更新.env文件。查询股价时返回模拟数据而非真实数据。FinanceDataClient中的get_stock_price方法是模拟实现。查看skills/data_query_skill.py中的方法实现。接入真实的金融数据API如Yahoo Finance, Alpha Vantage, 聚宽等并替换模拟方法。计划执行失败提示未注册的Skill。skill_registry中Skill的名称与PlanStep中的skill_name不匹配。检查main.py中skill_registry的注册键名和planner.py中PlanStep的生成是否一致。确保Skill类中的name属性与注册、调用时使用的字符串完全一致。大模型响应慢或超时。网络问题或模型负载高提示词过于复杂。检查网络连接在ModelSkill中增加超时设置优化提示词。使用httpx客户端设置超时考虑使用更快的模型如gpt-3.5-turbo简化提示词。提示词注入攻击风险。用户输入被直接拼接进提示词可能引导模型执行恶意指令。审查planner.py中create_plan方法里提示词的构建逻辑。对用户输入进行严格的清洗和转义使用系统提示词System Message来设定模型行为边界。8. 最佳实践与工程建议通过上面的项目我们已经实现了一个Harness工程的基本框架。要让其达到生产级别还需要考虑以下方面配置中心化将模型参数、API端点、超时时间等全部移出代码放入配置文件如YAML或配置服务如Apollo。pydantic-settings是一个很好的选择。可观测性Observability日志为每个Skill的执行、每个Plan的生成与执行记录结构化日志如使用structlog。指标Metrics记录每个请求的延迟、Token消耗、成功率可使用Prometheus。追踪Tracing使用OpenTelemetry追踪一个用户请求在所有微服务和Skill间的完整路径。测试策略单元测试测试每个Skill的execute方法。集成测试测试Planner和Executor的协作。端到端测试模拟用户请求测试整个API流程。大模型输出评估这是难点。可以定义评估标准相关性、安全性、事实准确性并编写评估脚本在数据集上运行。提示词管理将提示词模板从代码中剥离存入数据库或文件。可以设计版本管理、A/B测试和效果分析功能。模型路由与降级注册多个模型Skill如GPT-4, Claude, 文心一言。Planner可以根据问题类型、成本预算或当前负载动态选择模型。当主模型不可用时自动降级到备用模型。Agent状态与记忆当前的Agent是无状态的。对于多轮对话需要引入记忆机制如将历史对话摘要后放入上下文这需要在Planner和Skill输入中考虑会话状态。安全与合规输入输出过滤对用户输入和模型输出进行内容安全过滤。审计日志记录所有用户查询和AI响应满足合规要求。权限控制在应用层实现基于用户或角色的权限控制决定其可以访问哪些Skill或数据。9. 总结从Demo到工程的跨越至此我们已经完成了一个具备Harness工程思想的金融问答机器人从零到一的搭建。回顾整个过程Harness带来的最大改变不是功能而是结构。过去混乱的Demo一个app.py文件里混杂了API路由、Prompt字符串、OpenAI调用、数据处理逻辑。改一处而动全身无法测试难以扩展。现在Harness工程化我们有了清晰的分层连接、能力、编排、应用有了可复用、可测试的Skill有了可解释、可调度的Plan。要增加一个新功能比如查询财经新闻你只需要在连接层增加一个新闻API客户端。在能力层创建一个NewsQuerySkill。在编排层的Planner中增加对新闻类问题的识别和计划生成规则。应用层和执行器几乎无需改动。这个项目只是一个起点。你可以在此基础上深入探索更复杂的Planner例如基于大模型本身来动态规划步骤Meta-Prompting。集成向量数据库让机器人具备检索增强生成RAG能力回答基于私有知识库的问题。实现Skill的自动发现和注册机制。为Plan添加可视化工具方便调试复杂的AI工作流。Harness Engineering不是银弹但它为AI应用开发提供了应对复杂性的“工程地图”。当你下次再启动一个AI项目时不妨先问自己这个项目的Harness应该怎么设计
如果你正在学习AI大模型应用开发或者想从零开始构建一个AI Agent项目那么“Harness”这个词可能已经在你眼前出现了无数次。但你是否真的理解它到底是什么为什么它突然变得如此重要更重要的是当你面对一个“金融大模型问答机器人”这样的项目需求时Harness能帮你解决哪些具体问题很多人误以为Harness只是一个简单的“包装器”或“脚手架”把大模型API调用一下就算完事。这种理解会让你在实际项目中处处碰壁。真正的Harness Engineering解决的是从“一个能跑通的Demo”到“一个稳定、可靠、可维护的生产级AI应用”之间的巨大鸿沟。它涉及能力分层、模块边界、核心抽象、扩展机制甚至团队协作的权限模型。本文将从零开始手把手带你搞懂Harness的核心原理并通过一个完整的“金融大模型问答机器人”项目实战让你亲身体验如何用Harness工程化思维将一个AI想法落地为可交付的项目。无论你是零基础转行AI的开发者还是有一定经验但被AI项目混乱的代码结构所困扰这篇文章都将为你提供一条清晰的路径。1. Harness Engineering为什么它比“调API”重要100倍在AI大模型应用开发中最危险的错觉就是认为“调用API完成开发”。你写了几行代码调通了OpenAI的接口看到了返回结果感觉大功告成。但当你试图把这个“Demo”嵌入到真实的金融业务系统时问题接踵而至问题一提示词Prompt管理混乱。业务规则一变提示词就要改。你是把几百行的提示词模板硬编码在代码里还是写在配置文件里不同场景客服、研报解读、风险预警的提示词如何复用和组合问题二模型切换成本高。今天用GPT-4明天因为成本或响应速度想换为Claude或国产大模型。你需要重写多少代码如何保证不同模型输出格式的统一问题三缺乏业务逻辑与AI能力的隔离。你的核心业务逻辑如查询账户余额、计算投资组合风险和调用大模型的代码纠缠在一起。一旦需要升级或替换AI部分整个系统都得动。问题四可观测性Observability几乎为零。大模型为什么给出了某个回答它的思考过程Chain-of-Thought是什么每次调用的耗时、Token消耗、成功率是多少没有这些数据你根本无法进行性能优化和问题排查。问题五测试与评估Evaluation无从下手。如何自动化测试AI输出的准确性和稳定性难道每次都要人工看吗Harness Engineering驾驭工程的核心价值正是为了解决上述问题。它不是某个特定的框架如LangChain而是一套工程方法论和最佳实践的集合旨在为AI能力构建一个标准化的“接入层”和“控制层”。你可以把Harness想象成汽车的方向盘、油门和刹车系统Harness。发动机大模型本身威力巨大但如果没有这套驾驭系统汽车根本无法安全、可控地行驶。Harness就是让你能安全、高效、可控地“驾驭”大模型能力的那套工程体系。在接下来的内容中我们将围绕一个金融大模型问答机器人项目拆解Harness工程的关键组成部分并给出可运行的代码。2. 项目定义我们要构建一个什么样的系统在开始编码之前必须明确目标。我们的项目案例描述如下项目名称金融大模型问答机器人核心职责作为AI大模型应用开发工程师你需要设计并实现一个机器人能够理解用户关于金融领域的自然语言提问例如“帮我分析一下腾讯控股最近一年的股价走势”或“什么是ETF”并给出准确、可靠的回答。项目设计目标准确性对于事实类问题如定义、规则回答必须精确。安全性绝不能给出投资建议或做出财务预测合规要求。可扩展性易于接入新的数据源如实时行情API、财经新闻和新的AI模型。可维护性代码结构清晰提示词、模型配置、业务逻辑分离。3. Harness工程核心四层架构一个典型的Harness驱动AI应用可以抽象为以下四层。理解这个分层是做好一切设计的基础。层级职责对应项目中的组件关键问题1. 应用层 (Application Layer)面向用户的接口和核心业务流程。Web API (FastAPI/SpringBoot)、任务调度器。如何组织业务流水线如何将用户请求转化为AI可处理的任务2. 编排层 (Orchestration Layer)协调多个步骤或工具决定工作流。Harness 核心 定义Plan、Skill、Agent的执行逻辑。是先查数据库还是先问大模型如何根据答案决定下一步动作3. 能力层 (Capability Layer)提供具体的原子能力如调用模型、查询数据库、执行计算。Model Skill,Data Query Skill,Calculator Skill。如何统一不同能力模型、工具的调用接口如何管理它们的配置4. 连接层 (Connection Layer)管理与外部服务的连接和认证。大模型API密钥管理、数据库连接池、第三方财经API客户端。密钥如何安全存储连接如何复用和监控在我们的金融机器人中用户问题“分析腾讯股价”会这样流动应用层(接收HTTP请求) -编排层(制定计划先查询股价数据再让模型分析) -能力层(执行Data Skill查数据Model Skill发分析请求) -连接层(调用真实API)。4. 环境准备与项目初始化我们选择Python作为实现语言因为它在大模型生态中工具最丰富。项目将使用poetry进行依赖管理也可以用piprequirements.txt。4.1 基础环境Python版本: 3.9 或 3.103.11请留意某些包的兼容性操作系统: macOS / Linux / Windows (WSL2推荐)IDE: VSCode 或 PyCharm4.2 创建项目并安装核心依赖# 1. 创建项目目录 mkdir finance-ai-agent cd finance-ai-agent # 2. 初始化 poetry 项目 (如果没有poetry请先安装: pip install poetry) poetry init -n # 交互式创建或直接使用下面的 pyproject.toml # 3. 编辑 pyproject.toml添加依赖以下是pyproject.toml文件的内容示例# pyproject.toml [tool.poetry] name finance-ai-agent version 0.1.0 description A harness-engineered financial QA AI agent. authors [Your Name youexample.com] [tool.poetry.dependencies] python ^3.9 openai ^1.0.0 # 使用OpenAI官方新版SDK langchain ^0.1.0 # 可选这里我们主要展示自有Harness设计但可借鉴其思想 pydantic ^2.0.0 # 用于数据验证和设置管理 fastapi ^0.104.0 # 构建API接口 uvicorn ^0.24.0 # ASGI服务器 httpx ^0.25.0 # 异步HTTP客户端 python-dotenv ^1.0.0 # 管理环境变量 sqlalchemy ^2.0.0 # ORM (示例用) pandas ^2.0.0 # 数据处理 (示例用) [tool.poetry.group.dev.dependencies] pytest ^7.4.0 black ^23.0.0 isort ^5.12.0 [build-system] requires [poetry-core] build-backend poetry.core.masonry.api# 4. 安装依赖 poetry install # 5. 激活虚拟环境 poetry shell4.3 配置环境变量创建.env文件来存储敏感信息切记不要提交到版本控制系统。# .env OPENAI_API_KEYsk-your-openai-api-key-here # 未来可以扩展其他模型 ANTHROPIC_API_KEYyour-claude-key DATABASE_URLsqlite:///./finance.db # 示例用SQLite FINANCE_DATA_API_KEYyour-dummy-finance-api-key5. 核心模块设计与实现构建我们的Harness我们将自底向上实现Harness的各层。首先从最稳定的连接层和能力层开始。5.1 连接层统一的外部服务客户端创建core/clients.py:# core/clients.py import os from typing import Optional from openai import AsyncOpenAI import httpx from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from pydantic_settings import BaseSettings class Settings(BaseSettings): openai_api_key: str database_url: str sqlite:///./finance.db finance_api_base: str https://api.example-finance.com/v1 # 示例地址 class Config: env_file .env settings Settings() # 1. 大模型客户端 (以OpenAI为例) class OpenAIClient: _instance: Optional[AsyncOpenAI] None classmethod def get_client(cls) - AsyncOpenAI: if cls._instance is None: cls._instance AsyncOpenAI(api_keysettings.openai_api_key) return cls._instance # 2. 财经数据API客户端 (抽象类) class FinanceDataClient: def __init__(self): self.base_url settings.finance_api_base self.api_key os.getenv(FINANCE_DATA_API_KEY) self._async_client httpx.AsyncClient( base_urlself.base_url, headers{Authorization: fBearer {self.api_key}}, timeout30.0 ) async def get_stock_price(self, symbol: str, period: str 1y) - dict: 获取股票价格数据模拟 # 实际项目中这里会调用真实API如Yahoo Finance, Alpha Vantage等 # 此处返回模拟数据 return { symbol: symbol, period: period, data: [{date: 2023-01-01, close: 350.0}, {date: 2023-12-01, close: 400.0}], source: simulated } async def close(self): await self._async_client.aclose() # 3. 数据库会话工厂 engine create_engine(settings.database_url, connect_args{check_same_thread: False} if sqlite in settings.database_url else {}) SessionLocal sessionmaker(autocommitFalse, autoflushFalse, bindengine)5.2 能力层定义可复用的SkillSkill是Harness中的原子能力单元。创建skills/base_skill.py和具体Skill。# skills/base_skill.py from abc import ABC, abstractmethod from typing import Any, Dict from pydantic import BaseModel class SkillInput(BaseModel): Skill的输入参数基类 pass class SkillOutput(BaseModel): Skill的输出结果基类 success: bool data: Any error: Optional[str] None class BaseSkill(ABC): 所有Skill的抽象基类 name: str description: str abstractmethod async def execute(self, input_data: SkillInput) - SkillOutput: 执行技能的核心方法 pass现在实现两个具体的Skill调用大模型的ModelSkill和查询金融数据的DataQuerySkill。# skills/model_skill.py from skills.base_skill import BaseSkill, SkillInput, SkillOutput from core.clients import OpenAIClient from typing import Optional from pydantic import Field class ModelSkillInput(SkillInput): prompt: str model: str gpt-3.5-turbo temperature: float 0.7 max_tokens: Optional[int] None class ModelSkillOutput(SkillOutput): completion: Optional[str] None class ModelSkill(BaseSkill): name model_completion description 调用大语言模型生成文本 async def execute(self, input_data: ModelSkillInput) - ModelSkillOutput: client OpenAIClient.get_client() try: response await client.chat.completions.create( modelinput_data.model, messages[{role: user, content: input_data.prompt}], temperatureinput_data.temperature, max_tokensinput_data.max_tokens ) completion response.choices[0].message.content return ModelSkillOutput( successTrue, data{raw_response: response}, completioncompletion ) except Exception as e: return ModelSkillOutput( successFalse, dataNone, errorfModel调用失败: {str(e)} )# skills/data_query_skill.py from skills.base_skill import BaseSkill, SkillInput, SkillOutput from core.clients import FinanceDataClient from pydantic import Field class DataQuerySkillInput(SkillInput): query_type: str Field(..., description查询类型如 stock_price, news) parameters: Dict[str, Any] Field(default_factorydict) class DataQuerySkill(BaseSkill): name financial_data_query description 查询金融数据股价、新闻等 def __init__(self): self.client FinanceDataClient() async def execute(self, input_data: DataQuerySkillInput) - SkillOutput: try: if input_data.query_type stock_price: symbol input_data.parameters.get(symbol) period input_data.parameters.get(period, 1y) data await self.client.get_stock_price(symbol, period) return SkillOutput(successTrue, datadata) else: return SkillOutput( successFalse, dataNone, errorf不支持的查询类型: {input_data.query_type} ) except Exception as e: return SkillOutput(successFalse, dataNone, errorstr(e)) async def cleanup(self): 清理资源如关闭HTTP连接 await self.client.close()5.3 编排层制定与执行计划Plan编排层是Harness的大脑。它根据用户请求决定调用哪些Skill以及调用的顺序。我们实现一个简单的Planner和PlanExecutor。# orchestration/planner.py from typing import List, Dict, Any from pydantic import BaseModel from skills.base_skill import BaseSkill class PlanStep(BaseModel): skill_name: str skill_input: Dict[str, Any] depends_on: List[int] [] # 依赖的步骤索引 class Plan(BaseModel): steps: List[PlanStep] final_output_key: str # 最终结果从哪个步骤获取 class Planner: 根据用户意图制定执行计划 def __init__(self, available_skills: Dict[str, BaseSkill]): self.skills available_skills async def create_plan(self, user_query: str) - Plan: 核心规划逻辑分析用户问题生成执行步骤 # 这里可以非常简单也可以非常复杂甚至可以引入一个小模型来做规划 # 示例简单规则匹配 query_lower user_query.lower() if any(word in query_lower for word in [股价, stock, price, 走势]): # 识别出股票代码 (这里用简单正则实际应用需要更复杂的NLP) import re symbol_match re.search(r[A-Z]{1,5}, user_query.upper()) # 简单匹配美股代码 symbol symbol_match.group(0) if symbol_match else AAPL # 默认苹果 return Plan( steps[ PlanStep( skill_namefinancial_data_query, skill_input{query_type: stock_price, parameters: {symbol: symbol, period: 1y}} ), PlanStep( skill_namemodel_completion, skill_input{ prompt: f用户询问了{symbol}的股价走势。这里有一些历史数据{{stock_data}}。请用中文总结一下趋势但切记不要给出任何投资建议。, model: gpt-3.5-turbo }, depends_on[0] # 依赖于第0步的结果 ) ], final_output_key1 # 最终输出是第1步模型分析的结果 ) elif 是什么 in query_lower or 定义 in query_lower: # 纯概念性问题直接问模型 return Plan( steps[ PlanStep( skill_namemodel_completion, skill_input{ prompt: f用户问{user_query}。请用中文清晰、准确地解释这个概念如果涉及金融请确保解释符合常规金融定义。, model: gpt-3.5-turbo, temperature: 0.3 # 事实性问题降低随机性 } ) ], final_output_key0 ) else: # 默认兜底计划 return Plan( steps[ PlanStep( skill_namemodel_completion, skill_input{ prompt: f用户说{user_query}。请用中文友好地回应并说明你是一个金融信息问答助手无法提供个人投资建议。, model: gpt-3.5-turbo } ) ], final_output_key0 )# orchestration/executor.py from typing import Dict, Any from .planner import Plan, PlanStep from skills.base_skill import BaseSkill class PlanExecutor: 执行计划处理步骤间的依赖 def __init__(self, skill_registry: Dict[str, BaseSkill]): self.skill_registry skill_registry self.step_results: Dict[int, Any] {} async def execute_plan(self, plan: Plan) - Dict[str, Any]: 按依赖顺序执行计划中的所有步骤 # 拓扑排序执行简化版按顺序执行依赖通过占位符传递 for i, step in enumerate(plan.steps): # 准备输入替换依赖占位符 resolved_input self._resolve_input(step.skill_input, step.depends_on) # 获取对应的Skill skill self.skill_registry.get(step.skill_name) if not skill: raise ValueError(f未注册的Skill: {step.skill_name}) # 执行Skill output await skill.execute(resolved_input) if not output.success: # 错误处理可以重试、跳过或终止计划 raise Exception(f步骤 {i} ({step.skill_name}) 执行失败: {output.error}) # 存储结果 self.step_results[i] output.data # 特殊处理如果Skill输出有completion字段也存下来方便后续使用 if hasattr(output, completion) and output.completion: self.step_results[f{i}_completion] output.completion # 返回最终结果 final_result self.step_results.get(int(plan.final_output_key)) final_completion self.step_results.get(f{plan.final_output_key}_completion) return { final_data: final_result, final_completion: final_completion, all_step_results: self.step_results } def _resolve_input(self, skill_input: Dict[str, Any], depends_on: List[int]) - Dict[str, Any]: 解析输入中的占位符例如 {{stock_data}} import json resolved skill_input.copy() # 将整个字典转为字符串方便替换 input_str json.dumps(resolved) for dep_index in depends_on: dep_result self.step_results.get(dep_index) if dep_result: # 简单替换占位符实际项目需要更健壮的模板引擎 placeholder f{{{{step_{dep_index}}}}} # 这里我们简单地将依赖步骤的数据作为字符串插入。 # 更复杂的实现可以支持结构化数据的注入。 input_str input_str.replace(placeholder, json.dumps(dep_result)) return json.loads(input_str)5.4 应用层构建API服务最后我们用FastAPI将上述所有层整合起来暴露一个Web API。# main.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from contextlib import asynccontextmanager from core.clients import SessionLocal from skills.model_skill import ModelSkill from skills.data_query_skill import DataQuerySkill from orchestration.planner import Planner from orchestration.executor import PlanExecutor # 定义请求/响应模型 class QueryRequest(BaseModel): question: str user_id: str anonymous # 可用于审计和个性化 class QueryResponse(BaseModel): success: bool answer: str plan_id: str None # 可用于追踪 debug_info: dict {} # 开发阶段可返回调试信息 # 全局资源管理 skill_registry {} planner None executor None asynccontextmanager async def lifespan(app: FastAPI): # 启动逻辑 print(启动AI Agent服务...) # 1. 注册所有Skill global skill_registry, planner, executor skill_registry[model_completion] ModelSkill() skill_registry[financial_data_query] DataQuerySkill() # 2. 初始化编排器 planner Planner(skill_registry) executor PlanExecutor(skill_registry) yield # 关闭逻辑 print(关闭AI Agent服务...) # 清理资源如关闭数据库连接、HTTP客户端 data_skill skill_registry.get(financial_data_query) if data_skill: await data_skill.cleanup() app FastAPI(title金融AI问答机器人, lifespanlifespan) app.post(/query, response_modelQueryResponse) async def handle_query(request: QueryRequest): 处理用户查询的核心端点 try: # 1. 制定计划 plan await planner.create_plan(request.question) # 2. 执行计划 result await executor.execute_plan(plan) # 3. 构建响应 answer result.get(final_completion, 抱歉我暂时无法处理这个问题。) return QueryResponse( successTrue, answeranswer, debug_info{plan_steps: len(plan.steps)} # 生产环境应关闭 ) except Exception as e: # 记录日志 print(f处理查询时出错: {e}) raise HTTPException(status_code500, detailf内部服务器错误: {str(e)}) app.get(/health) async def health_check(): return {status: healthy} if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)6. 运行与测试6.1 启动服务确保在项目根目录下且虚拟环境已激活.env文件中的OPENAI_API_KEY已正确设置。python main.py服务将在http://localhost:8000启动。6.2 测试API使用curl或任何API测试工具如Postman进行测试。# 测试健康检查 curl http://localhost:8000/health # 测试概念性问题 curl -X POST http://localhost:8000/query \ -H Content-Type: application/json \ -d {question: 什么是ETF} # 测试股价分析模拟数据 curl -X POST http://localhost:8000/query \ -H Content-Type: application/json \ -d {question: AAPL的股价最近一年走势如何}6.3 预期输出对于“什么是ETF”的查询你应该会收到一个结构化的JSON响应其中answer字段包含大模型生成的解释文本并且会强调其作为金融信息助手不提供投资建议的立场。7. 常见问题与排查思路问题现象可能原因排查方式解决方案服务启动失败提示ModuleNotFoundError依赖未安装或虚拟环境未激活。检查poetry install是否成功确认终端处于poetry shell激活的虚拟环境中。重新运行poetry install并激活环境。API调用返回401或Invalid API KeyOpenAI API密钥错误或未设置。检查.env文件中的OPENAI_API_KEY确保其正确且未被空格包围。在OpenAI官网重新生成API Key并更新.env文件。查询股价时返回模拟数据而非真实数据。FinanceDataClient中的get_stock_price方法是模拟实现。查看skills/data_query_skill.py中的方法实现。接入真实的金融数据API如Yahoo Finance, Alpha Vantage, 聚宽等并替换模拟方法。计划执行失败提示未注册的Skill。skill_registry中Skill的名称与PlanStep中的skill_name不匹配。检查main.py中skill_registry的注册键名和planner.py中PlanStep的生成是否一致。确保Skill类中的name属性与注册、调用时使用的字符串完全一致。大模型响应慢或超时。网络问题或模型负载高提示词过于复杂。检查网络连接在ModelSkill中增加超时设置优化提示词。使用httpx客户端设置超时考虑使用更快的模型如gpt-3.5-turbo简化提示词。提示词注入攻击风险。用户输入被直接拼接进提示词可能引导模型执行恶意指令。审查planner.py中create_plan方法里提示词的构建逻辑。对用户输入进行严格的清洗和转义使用系统提示词System Message来设定模型行为边界。8. 最佳实践与工程建议通过上面的项目我们已经实现了一个Harness工程的基本框架。要让其达到生产级别还需要考虑以下方面配置中心化将模型参数、API端点、超时时间等全部移出代码放入配置文件如YAML或配置服务如Apollo。pydantic-settings是一个很好的选择。可观测性Observability日志为每个Skill的执行、每个Plan的生成与执行记录结构化日志如使用structlog。指标Metrics记录每个请求的延迟、Token消耗、成功率可使用Prometheus。追踪Tracing使用OpenTelemetry追踪一个用户请求在所有微服务和Skill间的完整路径。测试策略单元测试测试每个Skill的execute方法。集成测试测试Planner和Executor的协作。端到端测试模拟用户请求测试整个API流程。大模型输出评估这是难点。可以定义评估标准相关性、安全性、事实准确性并编写评估脚本在数据集上运行。提示词管理将提示词模板从代码中剥离存入数据库或文件。可以设计版本管理、A/B测试和效果分析功能。模型路由与降级注册多个模型Skill如GPT-4, Claude, 文心一言。Planner可以根据问题类型、成本预算或当前负载动态选择模型。当主模型不可用时自动降级到备用模型。Agent状态与记忆当前的Agent是无状态的。对于多轮对话需要引入记忆机制如将历史对话摘要后放入上下文这需要在Planner和Skill输入中考虑会话状态。安全与合规输入输出过滤对用户输入和模型输出进行内容安全过滤。审计日志记录所有用户查询和AI响应满足合规要求。权限控制在应用层实现基于用户或角色的权限控制决定其可以访问哪些Skill或数据。9. 总结从Demo到工程的跨越至此我们已经完成了一个具备Harness工程思想的金融问答机器人从零到一的搭建。回顾整个过程Harness带来的最大改变不是功能而是结构。过去混乱的Demo一个app.py文件里混杂了API路由、Prompt字符串、OpenAI调用、数据处理逻辑。改一处而动全身无法测试难以扩展。现在Harness工程化我们有了清晰的分层连接、能力、编排、应用有了可复用、可测试的Skill有了可解释、可调度的Plan。要增加一个新功能比如查询财经新闻你只需要在连接层增加一个新闻API客户端。在能力层创建一个NewsQuerySkill。在编排层的Planner中增加对新闻类问题的识别和计划生成规则。应用层和执行器几乎无需改动。这个项目只是一个起点。你可以在此基础上深入探索更复杂的Planner例如基于大模型本身来动态规划步骤Meta-Prompting。集成向量数据库让机器人具备检索增强生成RAG能力回答基于私有知识库的问题。实现Skill的自动发现和注册机制。为Plan添加可视化工具方便调试复杂的AI工作流。Harness Engineering不是银弹但它为AI应用开发提供了应对复杂性的“工程地图”。当你下次再启动一个AI项目时不妨先问自己这个项目的Harness应该怎么设计
双梁正问题 【 torch 实战】（Python代码实现）)
