基于Harness框架构建工业级多模态RAG Agent实战指南

最近在推进企业级大模型应用落地时很多团队都卡在了从“玩具Demo”到“工业级系统”的鸿沟上。一个简单的RAG问答原型可能几天就能搭出来但一旦涉及多模态数据、复杂业务流程、稳定生产部署和成本控制各种问题就接踵而至。本文旨在分享一套基于Harness框架构建工业级多模态RAG Agent的完整实战方案。我们将告别简单的单文本问答深入一个包含图像、PDF、表格数据的真实业务场景从系统架构设计、核心模块开发、到使用Harness进行工程化封装与部署手把手带你走完一个企业级项目的全流程。无论你是想入门Agent开发的程序员还是正在为项目落地发愁的架构师这套从架构到落地的实战经验或许能帮你少走90%的弯路。1. 背景与核心概念为什么需要工业级Agent在讨论具体技术之前我们首先要厘清几个关键概念并理解从“玩具”到“工业”的挑战所在。1.1 Agent、RAG与多模态技术演进与融合AI Agent智能体 可以简单理解为一个能感知环境、进行决策并执行动作以达成目标的AI程序。它不仅仅是调用一次大模型API而是具备规划Planning、工具使用Tool Use、记忆Memory等核心能力。在企业场景中Agent可以是一个自动处理工单的客服、一个分析报表的数据助手或一个管理研发流程的协调员。RAG检索增强生成 为了解决大模型知识陈旧、幻觉和私有数据查询问题而诞生的技术。其核心是“检索Retrieve相关文档片段 -增强Augment用户提问 -生成Generate最终答案”。它是提升大模型回答准确性和可追溯性的基石。多模态RAG 传统的RAG通常只处理文本。而企业数据是多样化的——产品设计图图像、财务年报PDF、销售数据表Excel、会议录音音频。多模态RAG要求系统能理解、检索和关联这些不同类型的数据并基于此生成回答。例如问“上一季度某产品的市场反馈如何”Agent需要能同时检索文本形式的调研报告和图像形式的产品评测截图。Harness 这里特指DeepSeek-Harness根据网络热词一个旨在简化大模型应用开发与部署的工程化框架。它并非某个具体的Agent实现而是提供了一套标准化的“缰绳Harness”用来管理和协调多个AI模型、工具链和业务流程让开发者能更专注于业务逻辑而非基础设施的搭建。你可以把它想象成AI时代的“Spring Framework”负责依赖注入、流程编排和生命周期管理。1.2 从玩具Demo到工业级系统的挑战搭建一个简单的单文本RAG聊天机器人玩具Demo可能只需要LangChain Chroma OpenAI API但将其升级为工业级系统你需要面对数据复杂性 多源、多格式、非结构化数据如何统一处理与索引流程复杂性 任务是否需要拆解是否需要调用外部API或数据库失败如何重试或降级稳定性与可观测性 如何监控每个环节检索、生成、工具调用的耗时、成功率和资源消耗如何做日志追踪和调试成本与性能 如何优化检索精度与召回率以降低大模型调用成本如何缓存中间结果部署与运维 如何将整个复杂的Agent系统打包、部署、扩缩容并集成到现有企业IT系统中Harness框架的价值正在于此它提供了一套范式来解决这些工程化问题让我们能构建出健壮、可维护、可观测的Agent应用。2. 环境准备与项目架构设计在写第一行代码之前明确我们的目标和整体设计至关重要。2.1 项目目标与场景定义我们将构建一个“企业智能知识库助手”Agent它需要处理以下场景用户提问“请总结上周产品评审会关于‘智能客服模块’的结论并找出相关的UI设计稿。”Agent需要理解问题拆解出“会议纪要”文本和“UI设计稿”图像两种需求。从知识库中检索上周的会议纪要PDF文档。从知识库中检索所有标签为“智能客服”的UI设计图PNG/JPG。综合文本和图像信息生成一份包含关键结论和设计稿摘要的报告。2.2 技术栈与版本说明Python: 3.9大模型服务:文本模型: OpenAI GPT-4o / Anthropic Claude 3.5 Sonnet / 或本地部署的 DeepSeek-V3。本文示例使用OpenAI API。多模态模型/Embedding模型: OpenAItext-embedding-3-small(用于文本)CLIP(用于图像向量化可选开源方案)。向量数据库:Qdrant(推荐性能好支持多向量) 或Pinecone(云服务)。本文使用Qdrant。开发框架:Harness (DeepSeek-Harness): 用于Agent流程编排与工程化管理。LangChain: 作为底层工具链的一部分用于文档加载、文本分割等。其他工具:PyPDF2/pdfplumber: PDF解析。Pillow/OpenCV: 图像处理。pandas: 表格数据处理。重要提示 框架和库版本迭代迅速以下代码示例以核心逻辑和Harness的使用范式为主具体版本请根据官方文档安装。2.3 系统架构设计一个工业级多模态RAG Agent的典型架构如下我们将在Harness框架内实现它[用户接口] | v [Harness Agent 入口] | (解析意图规划任务) v [任务执行引擎] (Harness Orchestrator) | |-----------------------| | | v v [文本处理管道] [图像处理管道] | | v v [文本加载/分割] [图像加载/特征提取] | | v v [文本向量化] [图像向量化] | | v v [向量数据库 (Qdrant)] - 统一索引 (支持多模态) | | |----------------------| (多路检索) v [检索结果融合与重排序] | v [提示词工程与上下文构建] | v [大模型调用 (生成答案)] | v [结果后处理与格式化] | v [返回给用户]Harness的角色 图中的Harness Agent 入口和任务执行引擎是Harness框架的核心。它定义了整个Agent的工作流Workflow每个步骤Step可以是工具调用、模型推理或条件判断。Harness负责管理这些步骤的状态、传递数据、处理异常并提供统一的日志和监控接口。3. 核心模块开发构建多模态RAG引擎在引入Harness编排之前我们先搭建好底层的核心能力模块。3.1 多模态数据加载与处理首先创建处理不同数据类型的工具函数。# file: core/data_processor.py import os from typing import List, Dict, Any, Tuple from PIL import Image import PyPDF2 import pdfplumber import pandas as pd import json class MultiModalDataProcessor: 多模态数据处理器 staticmethod def process_text_file(file_path: str) - List[Dict[str, Any]]: 处理纯文本文件 chunks [] with open(file_path, r, encodingutf-8) as f: content f.read() # 简单的按段落分割生产环境可用更复杂的语义分割 paragraphs [p for p in content.split(\n\n) if p.strip()] for i, para in enumerate(paragraphs): chunks.append({ content: para, metadata: { source: file_path, type: text, chunk_id: i, char_count: len(para) } }) return chunks staticmethod def process_pdf_file(file_path: str) - List[Dict[str, Any]]: 处理PDF文件提取文本和元数据 chunks [] try: with pdfplumber.open(file_path) as pdf: for page_num, page in enumerate(pdf.pages): text page.extract_text() if text: # 按页面分割也可按章节分割 chunks.append({ content: text, metadata: { source: file_path, type: pdf, page: page_num 1, char_count: len(text) } }) except Exception as e: print(f处理PDF {file_path} 时出错: {e}) return chunks staticmethod def process_image_file(file_path: str) - Dict[str, Any]: 处理图像文件提取基础信息和后续用于向量化的特征 try: img Image.open(file_path) # 提取基础元数据 info { source: file_path, type: image, format: img.format, size: img.size, mode: img.mode, } # 这里可以集成CLIP等模型的预处理后续在向量化步骤进行 # 目前只返回元数据和PIL对象或路径 return { content: file_path, # 存储路径实际向量化时再读取 metadata: info } except Exception as e: print(f处理图像 {file_path} 时出错: {e}) return None staticmethod def process_csv_file(file_path: str) - List[Dict[str, Any]]: 处理CSV文件将每行或相关行组作为chunk chunks [] try: df pd.read_csv(file_path) # 将DataFrame转换为易读的文本格式例如每行一个chunk for idx, row in df.iterrows(): row_text , .join([f{col}: {val} for col, val in row.items()]) chunks.append({ content: fRow {idx}: {row_text}, metadata: { source: file_path, type: csv, row_index: idx, columns: list(df.columns) } }) except Exception as e: print(f处理CSV {file_path} 时出错: {e}) return chunks3.2 多模态向量化与索引接下来创建向量化客户端用于将文本和图像转换为向量并存入Qdrant。# file: core/vector_client.py import openai from qdrant_client import QdrantClient from qdrant_client.http import models from typing import List, Dict, Any import numpy as np # 假设使用CLIP进行图像向量化需安装clip和torch # import torch # import clip class MultiModalVectorClient: 多模态向量化与检索客户端 def __init__(self, qdrant_host: str localhost, qdrant_port: int 6333, openai_api_key: str None, collection_name: str multimodal_kb): self.qdrant_client QdrantClient(hostqdrant_host, portqdrant_port) self.openai_api_key openai_api_key openai.api_key openai_api_key self.collection_name collection_name self.text_embedding_model text-embedding-3-small # 初始化CLIP模型示例实际需加载模型 # self.device cuda if torch.cuda.is_available() else cpu # self.clip_model, self.clip_preprocess clip.load(ViT-B/32, deviceself.device) self._ensure_collection() def _ensure_collection(self): 确保Qdrant集合存在并配置多向量支持如果需要 # 检查集合是否存在 collections self.qdrant_client.get_collections().collections collection_names [c.name for c in collections] if self.collection_name not in collection_names: # 创建支持多向量的集合。这里我们为文本和图像分别创建向量字段。 # 实际中可以创建一个包含text_vector和image_vector的集合。 # 为简化本例使用一个向量字段但用metadata区分类型。 self.qdrant_client.create_collection( collection_nameself.collection_name, vectors_configmodels.VectorParams( size1536, # OpenAI text-embedding-3-small 的维度 distancemodels.Distance.COSINE ) ) print(f集合 {self.collection_name} 创建成功。) def embed_text(self, text: str) - List[float]: 使用OpenAI Embedding API生成文本向量 response openai.embeddings.create( modelself.text_embedding_model, inputtext ) return response.data[0].embedding # def embed_image(self, image_path: str) - List[float]: # 使用CLIP生成图像向量示例函数 # image self.clip_preprocess(Image.open(image_path)).unsqueeze(0).to(self.device) # with torch.no_grad(): # image_features self.clip_model.encode_image(image) # return image_features.cpu().numpy().squeeze().tolist() def add_documents(self, documents: List[Dict[str, Any]]): 将处理后的文档文本/图像添加到向量数据库 points [] for doc in documents: payload doc[metadata] content doc[content] if payload[type] text or payload[type] pdf or payload[type] csv: # 文本类数据 vector self.embed_text(content) elif payload[type] image: # 图像类数据此处简化实际调用embed_image # vector self.embed_image(content) # 为示例我们暂时用文本描述生成向量生产环境需替换 # 例如可以先用BLIP等模型为图像生成描述文本再embed description f图像文件: {content} vector self.embed_text(description) payload[image_description] description else: continue point_id hash(f{payload[source]}_{payload.get(chunk_id, 0)}) points.append( models.PointStruct( idpoint_id, vectorvector, payloadpayload ) ) if points: self.qdrant_client.upsert( collection_nameself.collection_name, pointspoints ) print(f成功插入 {len(points)} 个点。) def search(self, query: str, filter_type: str None, limit: int 5) - List[Dict]: 混合检索根据查询文本检索相关片段可过滤类型 query_vector self.embed_text(query) # 构建过滤条件 filter_condition None if filter_type: filter_condition models.Filter( must[ models.FieldCondition( keytype, matchmodels.MatchValue(valuefilter_type) ) ] ) search_result self.qdrant_client.search( collection_nameself.collection_name, query_vectorquery_vector, query_filterfilter_condition, limitlimit ) results [] for hit in search_result: results.append({ content: hit.payload.get(content, ), # 注意存储时content可能在payload中 metadata: {k: v for k, v in hit.payload.items() if k ! content}, score: hit.score }) return results3.3 知识库初始化脚本创建一个脚本用于将示例数据灌入向量数据库。# file: scripts/init_knowledge_base.py import sys import os sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from core.data_processor import MultiModalDataProcessor from core.vector_client import MultiModalVectorClient from config.settings import QDRANT_HOST, QDRANT_PORT, OPENAI_API_KEY def init_kb(data_dir: str): 初始化知识库 processor MultiModalDataProcessor() vector_client MultiModalVectorClient( qdrant_hostQDRANT_HOST, qdrant_portQDRANT_PORT, openai_api_keyOPENAI_API_KEY ) all_documents [] for root, dirs, files in os.walk(data_dir): for file in files: file_path os.path.join(root, file) ext os.path.splitext(file)[1].lower() if ext .txt: chunks processor.process_text_file(file_path) all_documents.extend(chunks) elif ext .pdf: chunks processor.process_pdf_file(file_path) all_documents.extend(chunks) elif ext in [.png, .jpg, .jpeg]: img_doc processor.process_image_file(file_path) if img_doc: all_documents.append(img_doc) elif ext .csv: chunks processor.process_csv_file(file_path) all_documents.extend(chunks) else: print(f跳过不支持的文件类型: {file_path}) print(f共处理 {len(all_documents)} 个文档片段/文件。) # 批量添加到向量数据库 vector_client.add_documents(all_documents) print(知识库初始化完成) if __name__ __main__: # 假设你的数据放在 ./data 目录下 data_directory ./data init_kb(data_directory)4. 使用Harness构建工业级Agent现在我们引入Harness框架将上述模块组装成一个可管理、可观测的智能体。4.1 定义Harness Agent与工具Harness的核心是定义Agent和Tool。我们先定义几个核心工具。# file: agent/tools.py from harness.decision import Tool from pydantic import Field from typing import Type, Any import json from core.vector_client import MultiModalVectorClient from config.settings import QDRANT_HOST, QDRANT_PORT, OPENAI_API_KEY class SearchKnowledgeBaseTool(Tool): 检索知识库工具 name: str search_knowledge_base description: str 根据用户问题从多模态知识库中检索最相关的文本和图像信息。 query: str Field(..., description用户的查询问题) filter_type: str Field(None, description可选过滤类型如 text, pdf, image) def run(self, query: str, filter_type: str None) - str: vector_client MultiModalVectorClient(QDRANT_HOST, QDRANT_PORT, OPENAI_API_KEY) results vector_client.search(query, filter_typefilter_type, limit5) # 将结果格式化为清晰的文本供LLM阅读 formatted_results [] for res in results: source res[metadata].get(source, Unknown) content_preview res[content][:200] ... if len(res[content]) 200 else res[content] formatted_results.append(f- 来源: {source}\n 内容预览: {content_preview}\n 相关性分数: {res[score]:.3f}) return f检索到 {len(results)} 条相关信息\n \n---\n.join(formatted_results) class GenerateReportTool(Tool): 生成报告工具调用大模型 name: str generate_report description: str 根据检索到的信息生成一份结构化的总结报告。 retrieved_info: str Field(..., description从知识库检索到的信息) user_question: str Field(..., description用户的原始问题) def run(self, retrieved_info: str, user_question: str) - str: # 这里模拟调用大模型API。实际应集成OpenAI/Claude等。 # 为简化示例我们返回一个模拟响应。 prompt f 用户问题{user_question} 检索到的相关信息 {retrieved_info} 请基于以上信息生成一份简洁、专业的回答报告。报告应涵盖关键结论并提及相关图像或文档的来源。 # 模拟LLM调用 print(f[模拟LLM调用] Prompt: {prompt[:100]}...) # 实际调用代码示例 (OpenAI): # import openai # response openai.chat.completions.create( # modelgpt-4, # messages[{role: user, content: prompt}] # ) # report response.choices[0].message.content simulated_report f **关于“{user_question}”的报告** **核心结论** 1. 根据会议纪要上周产品评审会确定了智能客服模块的V1.2版本需求重点优化了多轮对话意图识别准确率。 2. 相关的UI设计稿共3份已定位主要调整了对话气泡的交互流程和视觉反馈。 **详细信息** {retrieved_info[:500]}... **建议** 建议UI团队根据评审会结论在下一轮设计迭代中融入提到的反馈点。 return simulated_report4.2 定义Harness Agent与工作流现在我们创建一个Agent它能够理解用户意图并自动规划和使用上述工具。# file: agent/multimodal_rag_agent.py from harness.agent import Agent from harness.decision import ReAct from agent.tools import SearchKnowledgeBaseTool, GenerateReportTool from config.settings import OPENAI_API_KEY, LLM_MODEL class MultimodalRAGAgent(Agent): 多模态RAG智能体 def __init__(self): # 初始化决策引擎例如ReAct推理行动 llm_config { api_key: OPENAI_API_KEY, model: LLM_MODEL, # 例如 gpt-4 } # Harness 的 ReAct 决策引擎会帮我们管理工具调用和LLM思考的循环 reasoning_engine ReAct( llm_configllm_config, tools[SearchKnowledgeBaseTool(), GenerateReportTool()] # 注册工具 ) super().__init__( reasoning_enginereasoning_engine, nameMultimodal_KB_Assistant, description一个能处理多模态企业知识库查询的智能助手。 ) def run(self, query: str) - str: 运行Agent处理用户查询 # Harness Agent 的 run 方法会触发决策引擎 # 决策引擎会根据query自动决定调用哪些工具、以什么顺序调用 final_result self.reasoning_engine.run(query) return final_result # 一个简单的、不依赖Harness高级编排的启动脚本 if __name__ __main__: agent MultimodalRAGAgent() user_question 请总结上周产品评审会关于‘智能客服模块’的结论并找出相关的UI设计稿。 print(f用户提问: {user_question}) print(\n *50 \n) answer agent.run(user_question) print(fAgent回答:\n{answer})4.3 通过Harness定义复杂工作流进阶对于更复杂的场景如先检索文本再根据文本结果中的关键词检索图像我们可以利用Harness的Workflow或Pipeline功能进行显式编排。# file: workflows/multimodal_qa_workflow.yaml (Harness工作流配置示例) # 注此为概念性YAML配置具体语法请参考Harness官方文档 name: multimodal_qa_workflow description: 多模态问答标准工作流 steps: - name: parse_intent type: llm config: prompt: | 分析用户问题{{input.question}} 判断需要检索的信息类型文本、图像、两者。 输出JSON格式{needs_text: true/false, needs_image: true/false, keywords: []} - name: retrieve_text type: tool tool: search_knowledge_base condition: {{steps.parse_intent.output.needs_text}} inputs: query: {{input.question}} {{#each steps.parse_intent.output.keywords}} {{this}} {{/each}} filter_type: text - name: retrieve_image type: tool tool: search_knowledge_base condition: {{steps.parse_intent.output.needs_image}} inputs: query: {{input.question}} {{#each steps.parse_intent.output.keywords}} {{this}} {{/each}} filter_type: image - name: synthesize_context type: code # 一个自定义函数用于融合文本和图像的检索结果 config: code: | def run(text_results, image_results): context 文本信息\n text_results \n\n图像信息\n image_results return {fused_context: context} inputs: text_results: {{steps.retrieve_text.output}} image_results: {{steps.retrieve_image.output}} - name: generate_final_answer type: llm config: prompt: | 基于以下综合信息回答用户问题{{input.question}} 信息 {{steps.synthesize_context.output.fused_context}} 请生成专业、准确的回答。通过YAML定义工作流我们可以将业务逻辑可视化、可配置化并且Harness引擎会负责执行、监控和记录每个步骤极大地提升了系统的可维护性和可观测性。5. 部署、监控与最佳实践5.1 项目结构与部署一个工业级项目的标准结构可能如下enterprise_rag_agent/ ├── config/ │ ├── __init__.py │ └── settings.py # 所有配置项API密钥、数据库地址等 ├── core/ # 核心业务逻辑 │ ├── __init__.py │ ├── data_processor.py # 多模态数据处理 │ └── vector_client.py # 向量化与检索 ├── agent/ # Agent相关定义 │ ├── __init__.py │ ├── tools.py # Harness Tool 定义 │ └── multimodal_rag_agent.py # 主Agent ├── workflows/ # Harness 工作流定义文件 │ └── multimodal_qa_workflow.yaml ├── scripts/ # 辅助脚本 │ └── init_knowledge_base.py ├── data/ # 原始数据 ├── tests/ # 单元测试 ├── Dockerfile # 容器化部署 ├── requirements.txt # Python依赖 ├── docker-compose.yml # 定义Qdrant等服务 └── README.md使用Docker Compose部署# file: docker-compose.yml version: 3.8 services: qdrant: image: qdrant/qdrant:latest ports: - 6333:6333 - 6334:6334 volumes: - ./qdrant_storage:/qdrant/storage restart: unless-stopped rag-agent-service: build: . ports: - 8000:8000 depends_on: - qdrant environment: - QDRANT_HOSTqdrant - QDRANT_PORT6333 - OPENAI_API_KEY${OPENAI_API_KEY} volumes: - ./data:/app/data command: uvicorn app.main:app --host 0.0.0.0 --port 8000 # 假设有一个FastAPI入口5.2 可观测性与日志Harness框架通常内置或可集成监控。此外你需要自己记录关键指标应用日志 使用structlog或loguru结构化记录每个请求的ID、用户查询、调用的工具、耗时、检索结果数量、Token使用量、最终答案。性能指标 使用Prometheus客户端暴露指标如rag_retrieval_duration_seconds、llm_invocation_total、tool_call_success_rate并用Grafana展示。链路追踪 对于复杂工作流使用OpenTelemetry进行分布式追踪看清一个请求流经的所有服务向量DB、Embedding API、LLM API。5.3 工业级最佳实践配置中心化 所有密钥、端点、模型名称必须通过环境变量或配置中心如Apollo管理严禁硬编码。优雅降级与重试 对LLM API、向量数据库的调用必须添加重试机制如tenacity库和超时设置。当核心服务失败时应有降级策略如返回缓存结果或友好提示。缓存策略 对频繁的相似查询结果进行缓存如使用Redis可以显著降低成本和延迟。权限与审计 企业级应用必须考虑数据权限。在检索前根据用户角色过滤可访问的数据源。记录所有查询和生成记录用于审计。评估与迭代 建立评估体系定期用测试集评估Agent的准确率、相关性和安全性。根据评估结果迭代优化提示词、检索策略和模型。成本控制 监控Token消耗设置预算和告警。考虑对内部知识使用更小的专用模型仅在需要复杂推理时调用大模型。6. 常见问题与排查思路问题现象可能原因排查思路与解决方案Agent回答“未找到相关信息”1. 知识库未成功灌入数据。2. 检索query与文档embedding不匹配。3. 向量数据库连接失败。1. 检查init_knowledge_base.py脚本日志确认数据已插入。2. 检查查询文本的预处理是否与入库时一致如大小写、标点。3. 使用Qdrant Dashboard或客户端检查集合内是否有数据。处理图像时效果差1. 使用文本描述代替图像向量丢失视觉信息。2. CLIP等图像模型未正确加载或版本不匹配。1.核心实现真正的多模态Embedding。使用CLIP、BLIP等模型生成图像向量并与文本向量在同一个Qdrant集合的不同字段存储进行多向量检索。2. 确保图像预处理缩放、归一化符合模型要求。Harness Agent不调用工具1. Tool的description描述不清LLM无法理解其用途。2. LLM配置API Key, Base URL错误。3. ReAct决策循环超时或出错。1. 优化Tool的name和description确保清晰、无歧义。2. 检查Harness Agent的LLM配置测试简单的LLM调用是否成功。3. 打开Harness的详细日志查看决策过程中的思考链Chain-of-Thought。系统响应慢1. Embedding API调用延迟高。2. 检索时未使用索引或过滤条件不当。3. LLM生成速度慢。1. 考虑使用本地Embedding模型如BGE、text2vec。2. 在Qdrant中为常用过滤字段如type,source创建Payload索引。3. 对大模型生成设置合理的max_tokens和temperature考虑使用流式响应。部署后服务不稳定1. 内存或CPU资源不足。2. 依赖服务Qdrant, OpenAI网络波动。3. 代码未处理所有异常。1. 使用Docker资源限制监控容器资源使用率。2. 为所有外部调用添加重试、熔断机制如使用backoff,circuitbreaker库。3. 完善全局异常处理返回友好的错误信息避免服务崩溃。构建工业级Agent是一个系统工程它远不止于拼接几个API。本文通过一个多模态RAG Agent的实战案例展示了如何从核心模块开发起步再利用Harness这样的工程化框架进行编排和整合最终关注部署、监控和最佳实践。这条路没有捷径但遵循清晰的架构和工程化思想能让你避开大多数坑真正将大模型能力稳定、可靠地融入企业业务流程。下一步你可以尝试集成更复杂的工具如数据库查询、API调用设计更精细的工作流并建立完整的评估与运维体系让你的Agent从“能用”变得“好用”且“耐用”。