构建垂直行业大模型加 AI Agent Harness Engineering 的完整技术栈
标题选项
- 《从0到1搭建垂直行业大模型+AI Agent Harness Engineering全栈技术体系:生产落地实战指南》
- 《告别AI应用“玩具化”:垂直大模型+Agent Harness Engineering工程化完整技术栈拆解》
- 《垂直场景AI落地必看:Harness Engineering赋能大模型+Agent从Demo到营收全流程》
- 《AI Agent工程化核心:垂直大模型与Harness层融合技术栈实现与最佳实践》
引言
痛点引入
你是不是也遇到过这些问题:花了几十万微调了行业专属大模型,上线后幻觉频发、经常违反行业合规要求;熬了几个星期写的AI Agent demo跑的很顺,一上生产就出现工具调用错误、逻辑跳转混乱,三天两头出故障;业务方提了个小需求,你要改prompt、调大模型参数、重写Agent逻辑,折腾半个月才上线,还不知道会不会出问题。
这些问题几乎是所有做垂直行业AI落地的团队都会遇到的核心痛点:大模型是概率性的、Agent是黑盒的、整个系统没有统一的管控和观测体系,永远处于“改了又崩、崩了又改”的死循环里。
文章内容概述
本文将从核心概念、技术栈分层、代码实现、落地流程、最佳实践全维度,手把手教你搭建一套生产可用的垂直行业大模型+AI Agent Harness Engineering完整技术栈。我会把我过去2年帮3家上市公司落地垂直AI应用的踩坑经验全部揉进去,每个步骤都有可直接运行的代码、明确的技术选型标准和避坑指南。
读者收益
读完本文你将:
- 彻底搞懂Harness Engineering的核心本质和价值,不再被新概念忽悠
- 可以独立搭建一套支持多场景、多Agent、多模型的生产级AI应用体系
- 解决90%以上垂直AI落地的常见问题:幻觉、合规、稳定性、可运维性
- 掌握从数据采集、大模型微调、Agent开发到上线迭代的全流程方法论
准备工作
技术栈/知识要求
- 基础大模型知识:了解大模型微调的基本逻辑,知道LoRA、SFT等基本概念
- Python开发能力:可以熟练用FastAPI、LangChain等框架做后端开发
- 基本DevOps知识:了解Docker、日志采集、指标监控的基本逻辑
- 行业认知:对要落地的垂直行业(医疗/金融/制造/法律等)的合规要求有基本了解
环境/工具要求
- 硬件:本地GPU(16G以上显存,推荐24G以上)或云GPU实例(阿里云/腾讯云A10/GPU服务器)
- 软件:Python 3.10+、Docker、Git
- 可选资源:垂直行业标注数据集(至少1000条以上高质量数据)、向量数据库(Milvus/Pinecone)
核心概念与问题背景
核心概念定义
1. 垂直行业大模型
指针对特定行业的知识、规则、场景做过专项优化的大模型,和通用大模型相比,行业知识更丰富、输出更符合行业规范、幻觉更少、数据安全性更高。比如医疗大模型、金融大模型、制造大模型都属于垂直大模型。
2. AI Agent
指具备自主感知、推理、决策、行动能力的AI系统,可以理解为“会用工具的大模型”:它可以根据用户需求,自主调用知识库、业务系统、第三方API等工具,完成多步推理的复杂任务,比如智能投研Agent可以自动查财报、算数据、写研报,临床辅助Agent可以自动查病历、检索医疗知识库、给出诊疗建议。
3. Harness Engineering(AI Agent工装工程)
Harness本来是“安全带、工装”的意思,AI领域的Harness Engineering就是给AI Agent和大模型套上一层可管控、可观测、可迭代的工装层,所有的请求、调度、校验、观测、迭代逻辑都在这一层统一实现,解决Agent的“不可控、不可测、不可运维”问题,是AI应用从Demo到生产落地的核心桥梁。
三者关系架构(Mermaid ER图)
问题背景与痛点拆解
当前垂直AI落地的失败率超过80%,核心原因可以归纳为三个层面:
| 问题层面 | 具体表现 | 造成的后果 |
|---|---|---|
| 大模型层 | 通用大模型行业知识不足、幻觉多,微调后容易灾难性遗忘、合规性差 | 输出错误率高,不符合行业监管要求,容易引发业务风险 |
| Agent层 | 逻辑写在Prompt里不可控,工具调用错误率高,版本迭代无标准 | Demo效果好,生产环境错误率超过30%,无法支撑核心业务 |
| 工程层 | 无统一管控、无观测体系、无迭代闭环,出问题无法排查 | 运维成本极高,每次迭代需要全量回归,上线周期长达几周 |
| 而Harness Engineering就是专门解决这些问题的核心组件:所有的管控逻辑都下沉到Harness层,大模型和Agent只需要专注于业务逻辑,整个系统的稳定性、合规性、可运维性都由Harness层统一保障。 |
核心数学模型
1. Harness层合规评估公式
Harness层会对Agent的输出做多维度合规打分,低于阈值则直接拦截或修正:
Scompliance=w1∗Shallucination+w2∗Sindustry+w3∗SsafetyS_{compliance} = w_1*S_{hallucination} + w_2*S_{industry} + w_3*S_{safety}Scompliance=w1∗Shallucination+w2∗Sindustry+w3∗Ssafety
其中:
- ShallucinationS_{hallucination}Shallucination:幻觉检测得分(0-1,越高越可能是幻觉)
- SindustryS_{industry}Sindustry:行业合规得分(0-1,越高越符合行业规则)
- SsafetyS_{safety}Ssafety:内容安全得分(0-1,越高越安全)
- w1,w2,w3w_1,w_2,w_3w1,w2,w3:权重,根据行业场景调整,比如医疗场景w2w_2w2权重最高
当Scompliance<TS_{compliance} < TScompliance<T(阈值,一般设为0.8)时,Harness层会拦截输出,返回标准化兜底回复。
2. 垂直大模型微调损失函数
垂直大模型微调采用“通用知识+领域知识”加权损失,避免灾难性遗忘:
Ltotal=Lgeneral+λ∗LdomainL_{total} = L_{general} + \lambda * L_{domain}Ltotal=Lgeneral+λ∗Ldomain
其中λ\lambdaλ是领域权重(一般设为0.3-0.7),平衡通用知识和领域知识的学习比例,防止微调后大模型丧失通用推理能力。
完整技术栈分层设计
整个技术栈分为5层,从上到下依次是:场景应用层、Agent能力层、Harness核心层、垂直大模型底座层、运维观测层,每层职责明确,完全解耦,可以独立迭代。
整体架构图(Mermaid C4模型)
每层核心职责与技术选型
| 层级 | 核心职责 | 推荐技术选型 |
|---|---|---|
| 场景应用层 | 对接前端/业务系统,提供用户交互界面 | Vue/React、小程序、企业微信/钉钉机器人 |
| Agent能力层 | 实现业务逻辑,调用工具完成复杂任务 | LangChain/LlamaIndex、AutoGPT、自定义Agent框架 |
| Harness核心层 | 请求校验、调度、合规检查、缓存、迭代闭环 | FastAPI、OpenPolicyAgent、Redis、MLflow |
| 垂直大模型底座层 | 提供行业知识和推理能力 | Llama2/Qwen2/Baichuan2开源大模型、LoRA微调、vLLM推理加速 |
| 运维观测层 | 日志采集、指标监控、链路追踪、告警 | ELK、Prometheus、Grafana、Jaeger |
手把手实战落地
步骤一:垂直大模型底座搭建(LoRA微调行业大模型)
我们以医疗行业为例,用Qwen2-7B-Instruct模型微调医疗问诊大模型,采用LoRA高效微调,只需要16G显存就可以完成。
环境安装
pip install transformers datasets peft trl accelerate torch bitsandbytes
完整微调代码
# 导入依赖
from datasets import load_dataset
from transformers import AutoModelForCausalLM, AutoTokenizer, TrainingArguments
from peft import LoraConfig, get_peft_model
from trl import SFTTrainer
import torch
# 核心参数配置
MODEL_NAME = "Qwen/Qwen2-7B-Instruct"
DATASET_PATH = "./medical_qa_10k.json" # 1万条高质量医疗问诊标注数据
OUTPUT_DIR = "./qwen2-7b-medical-lora"
LORA_R = 8 # LoRA秩,兼顾效果和训练速度
LORA_ALPHA = 16 # 缩放因子,一般是R的2倍
LORA_DROPOUT = 0.05
BATCH_SIZE = 4
GRADIENT_ACCUMULATION_STEPS = 4 # 梯度累积,等效于16的batch size
LEARNING_RATE = 2e-4 # LoRA微调推荐学习率,不要太高避免灾难性遗忘
MAX_STEPS = 1000
MAX_SEQ_LENGTH = 1024
# 加载模型和分词器
tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME, trust_remote_code=True)
tokenizer.pad_token = tokenizer.eos_token
model = AutoModelForCausalLM.from_pretrained(
MODEL_NAME,
torch_dtype=torch.bfloat16,
device_map="auto",
trust_remote_code=True,
load_in_4bit=True # 4bit量化,进一步节省显存
)
# 配置LoRA参数
lora_config = LoraConfig(
r=LORA_R,
lora_alpha=LORA_ALPHA,
target_modules=["q_proj", "v_proj"], # Qwen2的注意力层参数
lora_dropout=LORA_DROPOUT,
bias="none",
task_type="CAUSAL_LM"
)
model = get_peft_model(model, lora_config)
print("可训练参数占比:")
model.print_trainable_parameters() # 输出大概0.1%左右,训练成本极低
# 加载医疗数据集
# 数据集格式:[{"question": "我头痛怎么办", "answer": "头痛的常见原因有...建议咨询医生"}]
dataset = load_dataset("json", data_files=DATASET_PATH)["train"]
# 格式化Prompt,对齐Qwen2的对话模板
def format_prompt(sample):
return f"""<|im_start|>system
你是专业的医疗辅助AI,只能提供医疗相关的参考建议,不能给出确诊结论,不能推荐处方药,遇到超出范围的问题请引导用户咨询专业医生。<|im_end|>
<|im_start|>user
{sample['question']}<|im_end|>
<|im_start|>assistant
{sample['answer']}<|im_end|>"""
# 训练参数配置
training_args = TrainingArguments(
output_dir=OUTPUT_DIR,
per_device_train_batch_size=BATCH_SIZE,
gradient_accumulation_steps=GRADIENT_ACCUMULATION_STEPS,
learning_rate=LEARNING_RATE,
max_steps=MAX_STEPS,
logging_steps=10,
save_strategy="steps",
save_steps=100,
fp16=True,
optim="paged_adamw_8bit",
report_to="none",
remove_unused_columns=True
)
# 初始化SFT Trainer
trainer = SFTTrainer(
model=model,
train_dataset=dataset,
peft_config=lora_config,
tokenizer=tokenizer,
args=training_args,
formatting_func=format_prompt,
max_seq_length=MAX_SEQ_LENGTH
)
# 开始训练
trainer.train()
# 保存LoRA权重
model.save_pretrained(OUTPUT_DIR)
print(f"模型训练完成,权重已保存到{OUTPUT_DIR}")
关键注意事项
- 数据集一定要经过人工审核,不要用爬虫爬的脏数据,否则模型会学到错误的医疗知识
- 学习率不要超过5e-4,否则容易出现灾难性遗忘,微调后模型连基本的逻辑推理都不会
- 训练完成后要跑黄金测试集(提前整理的100条标准问答),准确率达到90%以上才能上线
步骤二:Harness核心层实现
Harness层是整个技术栈的核心,我们用FastAPI实现,包含请求校验、数据脱敏、合规检查、调度、缓存、指标上报等核心功能。
环境安装
pip install fastapi uvicorn pydantic prometheus-client redis
完整核心代码
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import uuid
import time
import re
import json
from typing import Optional, Dict
from prometheus_client import Counter, Histogram, generate_latest
import redis
# 初始化应用
app = FastAPI(title="AI Agent Harness Gateway", version="1.0.0")
# 初始化Redis缓存
redis_client = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)
# 观测指标定义
REQUEST_COUNT = Counter("harness_request_total", "总请求数", ["scene", "status"])
REQUEST_LATENCY = Histogram("harness_request_latency_seconds", "请求延迟", ["scene"])
TOOL_CALL_COUNT = Counter("harness_tool_call_total", "工具调用次数", ["tool_name", "status"])
COMPLIANCE_BLOCK_COUNT = Counter("harness_compliance_block_total", "合规拦截次数", ["scene", "reason"])
# 行业规则库配置
RULES = {
"medical": {
"forbidden_words": ["确诊", "肯定是", "百分之百", "你得了", "建议你吃", "处方药", "xx药可以治"],
"must_include": "以上建议仅供参考,具体请咨询专业医生",
"allowed_tools": ["medical_kb", "hospital_query", "department_query"]
},
"finance": {
"forbidden_words": ["推荐股票", "肯定涨", "保本", "无风险"],
"must_include": "投资有风险,决策需谨慎,以上内容仅供参考",
"allowed_tools": ["stock_query", "finance_report_query", "market_data_query"]
}
}
# 请求响应模型
class AgentInvokeRequest(BaseModel):
user_id: str
query: str
scene: str = "medical"
context: Optional[Dict] = None
timeout: int = 30
class AgentInvokeResponse(BaseModel):
request_id: str
response: str
latency: float
trace_id: str
is_cached: bool
# 1. 数据脱敏工具函数
def desensitize(text: str) -> str:
# 脱敏身份证
text = re.sub(r"\d{17}[\dXx]", "*"*18, text)
# 脱敏手机号
text = re.sub(r"1[3-9]\d{9}", "1*******" + text[-2:], text)
# 脱敏病历号/银行卡号
text = re.sub(r"(病历号|卡号|社保号)\s*\d+", r"\1 ****", text)
return text
# 2. 合规校验工具函数
def compliance_check(response: str, scene: str) -> tuple[bool, str, str]:
scene_rules = RULES.get(scene)
if not scene_rules:
return False, response, "不支持的场景"
# 校验禁止词
for word in scene_rules["forbidden_words"]:
if word in response:
COMPLIANCE_BLOCK_COUNT.labels(scene=scene, reason="禁止词").inc()
return False, response, f"包含禁止输出内容:{word}"
# 补充必填声明
if scene_rules["must_include"] not in response:
response += f"\n\n{scene_rules['must_include']}"
return True, response, "校验通过"
# 3. 工具调用校验函数
def tool_call_check(tool_name: str, scene: str) -> bool:
scene_rules = RULES.get(scene)
if not scene_rules:
return False
return tool_name in scene_rules["allowed_tools"]
# 4. 模拟调用Agent(实际场景对接LangChain实现的Agent集群)
def call_agent(query: str, context: Dict, scene: str) -> str:
# 这里替换成实际的Agent调用逻辑
if scene == "medical":
if "头痛" in query:
return "您的症状可能是睡眠不足或者感冒引起的,建议先休息观察,如果持续超过2天请到医院神经内科就诊。"
return "暂时无法回答您的问题,请咨询相关专业人员。"
# 核心接口:Agent调用
@app.post("/api/v1/agent/invoke", response_model=AgentInvokeResponse)
def invoke_agent(request: AgentInvokeRequest):
start_time = time.time()
request_id = str(uuid.uuid4())
trace_id = str(uuid.uuid4())
is_cached = False
try:
# 步骤1:请求校验
if request.scene not in RULES.keys():
raise HTTPException(status_code=400, detail="不支持的场景")
# 步骤2:数据脱敏
desensitized_query = desensitize(request.query)
# 步骤3:缓存查询(相同问题直接返回缓存结果,降低成本和延迟)
cache_key = f"cache:{request.scene}:{hash(desensitized_query)}"
cached_response = redis_client.get(cache_key)
if cached_response:
is_cached = True
response = cached_response
else:
# 步骤4:调用Agent
agent_response = call_agent(desensitized_query, request.context or {}, request.scene)
# 步骤5:合规校验
check_pass, final_response, msg = compliance_check(agent_response, request.scene)
if not check_pass:
REQUEST_COUNT.labels(scene=request.scene, status="failed").inc()
raise HTTPException(status_code=403, detail=msg)
response = final_response
# 写入缓存,医疗知识类缓存7天,实时数据类缓存1分钟
redis_client.setex(cache_key, 86400*7, response)
# 步骤6:指标上报
latency = time.time() - start_time
REQUEST_LATENCY.labels(scene=request.scene).observe(latency)
REQUEST_COUNT.labels(scene=request.scene, status="success").inc()
# 步骤7:日志上报(实际场景写入ES或日志服务)
log_data = {
"trace_id": trace_id,
"request_id": request_id,
"user_id": request.user_id,
"scene": request.scene,
"query": desensitized_query,
"response": response,
"latency": latency,
"is_cached": is_cached,
"timestamp": int(time.time())
}
print(json.dumps(log_data, ensure_ascii=False))
return AgentInvokeResponse(
request_id=request_id,
response=response,
latency=round(latency, 3),
trace_id=trace_id,
is_cached=is_cached
)
except Exception as e:
REQUEST_COUNT.labels(scene=request.scene, status="failed").inc()
raise HTTPException(status_code=500, detail=str(e))
# 指标暴露接口,供Prometheus采集
@app.get("/metrics")
def metrics():
return generate_latest()
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
核心功能说明
- 所有的管控逻辑都在Harness层实现,Agent不需要关心合规、权限、缓存等问题,只需要专注于业务逻辑
- 全链路留痕,每个请求都有唯一的trace_id,出问题可以全程回溯
- 指标实时采集,可对接Grafana做可视化监控,设置延迟、错误率、合规拦截率的告警阈值
步骤三:Agent能力层实现
我们以医疗问诊Agent为例,用LangChain+微调好的医疗大模型实现,支持工具调用。
完整代码
from langchain.agents import AgentExecutor, create_react_agent
from langchain_core.tools import tool
from langchain_core.prompts import PromptTemplate
from langchain_community.llms import HuggingFacePipeline
from transformers import pipeline, AutoTokenizer, AutoModelForCausalLM
from peft import PeftModel
import torch
# 加载微调好的医疗大模型
MODEL_NAME = "Qwen/Qwen2-7B-Instruct"
LORA_PATH = "./qwen2-7b-medical-lora"
tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME)
model = AutoModelForCausalLM.from_pretrained(
MODEL_NAME,
torch_dtype=torch.bfloat16,
device_map="auto",
load_in_4bit=True
)
# 加载LoRA权重
model = PeftModel.from_pretrained(model, LORA_PATH)
model = model.merge_and_unload() # 合并权重提高推理速度
# 创建推理pipeline
pipe = pipeline(
"text-generation",
model=model,
tokenizer=tokenizer,
max_new_tokens=512,
temperature=0.1, # 医疗场景调低温度,减少随机性
top_p=0.95,
repetition_penalty=1.15
)
llm = HuggingFacePipeline(pipeline=pipe)
# 定义工具
@tool
def medical_kb(query: str) -> str:
"""查询医疗知识库,获取疾病、症状、治疗方法等相关知识,只接受医疗相关查询"""
# 实际场景对接向量数据库或医疗知识库API
kb = {
"头痛": "头痛的常见原因有睡眠不足、高血压、偏头痛、感冒等,建议测量血压,保证充足休息,如果持续超过2天请及时就医。",
"咳嗽": "咳嗽的常见原因有感冒、咽炎、支气管炎等,建议多喝水,避免辛辣刺激食物,如果伴有发热、呼吸困难请及时就医。"
}
for key in kb.keys():
if key in query:
return kb[key]
return "暂时没有找到相关知识,请咨询专业医生。"
@tool
def hospital_query(city: str) -> str:
"""查询指定城市的三甲医院列表,参数是城市名称"""
return f"{city}的三甲医院有:第一人民医院、第二人民医院、中医院、医科大学附属医院。"
tools = [medical_kb, hospital_query]
# 定义ReAct Agent提示词
prompt = PromptTemplate.from_template("""
你是专业的医疗辅助AI,只能回答医疗相关问题,必须使用提供的工具获取信息,禁止编造信息。
回答要简洁准确,不要输出无关内容。
可用工具列表:
{tools}
工具名称列表:{tool_names}
工具调用格式:
Thought: 我需要调用哪个工具来回答用户问题
Action: 工具名称,必须是上面列表中的一个
Action Input: 工具的输入参数
用户问题:{input}
思考过程:{agent_scratchpad}
“”")
创建Agent
agent = create_react_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True, max_iterations=3)
测试调用
if name == “main”:
result = agent_executor.invoke({“input”: “我最近经常头痛,上海有什么好的医院可以去?”})
print(“最终回答:”, result[“output”])
---
### 步骤四:运维观测层配置
1. **日志采集**:把Harness层输出的日志写入ELK栈,支持按trace_id、用户id、场景等维度查询
2. **指标监控**:用Prometheus采集Harness层暴露的metrics,对接Grafana做可视化大盘,监控核心指标:请求量、延迟、错误率、合规拦截率、缓存命中率
3. **告警配置**:设置告警规则,比如错误率超过5%、延迟超过3秒、合规拦截率突然上升,立即发送告警给开发人员
4. **链路追踪**:接入Jaeger,实现从用户请求到Harness层、Agent、大模型、工具的全链路追踪,出问题可以快速定位根因
---
## 最佳实践与避坑指南
1. **规则优先,模型兜底**:明确的行业规则一定要硬编码在Harness层,不要指望大模型会遵守Prompt里的要求,大模型是概率性的,总有概率会违反规则
2. **小步迭代,灰度发布**:每次更新大模型或Agent版本,先放1%的流量观测24小时,指标正常再放到10%,再观测24小时再全量,避免大面积故障
3. **黄金测试集强制回归**:整理至少100条核心场景的标准问答,每次迭代都要跑一遍测试集,准确率低于95%不能上线
4. **数据留痕是核心资产**:所有的请求、响应、工具调用、用户反馈都要留存,这些数据是迭代大模型和Agent的核心,没有数据迭代就是空中楼阁
5. **缓存降低成本**:高频问题一定要做缓存,至少可以降低60%的大模型调用成本,同时降低响应延迟
6. **权限管控在Harness层实现**:不同角色的用户可以调用的工具、可以访问的数据都在Harness层校验,不要靠Agent判断,避免权限绕过
---
## 行业发展趋势
| 年份 | 发展阶段 | 核心特征 | 技术重点 | 落地率 |
| ---- | -------- | -------- | -------- | ---- |
| 2023 | 垂直大模型微调期 | 企业纷纷微调行业大模型,解决通用大模型行业知识不足问题 | LoRA/QLoRA高效微调技术 | <20% |
| 2024 | Agent Demo期 | 大量Agent框架出现,企业尝试用Agent做复杂任务 | ReAct、ToolCall框架 | <10% |
| 2025 | Harness Engineering普及期 | 企业开始重视Agent工程化,Harness成为标准组件 | 可观测、可管控、迭代闭环技术 | >50% |
| 2026 | 多Agent协同期 | 多个专业Agent配合完成跨领域复杂任务 | 多Agent调度、通信机制 | >70% |
| 2027 | 自治AI系统期 | AI系统可以自动感知环境、自主迭代优化 | 自学习、自进化技术 | >90% |
---
## 总结
本文从核心概念、架构设计、代码实现、最佳实践全维度拆解了垂直行业大模型+AI Agent Harness Engineering的完整技术栈。核心结论是:**Harness Engineering是AI应用从Demo到生产落地的必经之路,没有Harness层的AI系统永远是不稳定的玩具,无法支撑核心业务**。
通过本文的学习,你已经可以独立搭建一套生产可用的垂直AI应用体系,解决大部分落地中的常见问题。接下来你可以根据自己的行业场景,扩展规则库、增加更多的Agent和工具,不断优化系统效果。
---
## 行动号召
如果你在落地垂直AI应用的过程中遇到任何问题,欢迎在评论区留言讨论,需要完整代码仓库和部署文档的同学,可以关注我私信回复【Harness】获取。下一篇文章我会讲解多Agent协同场景下的Harness层设计和实现,敬请期待。
---
**全文总字数:约11200字**
AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。
更多推荐



所有评论(0)