人机协作中 AI Agent Harness Engineering 的角色定位

杨正康396

322人浏览 · 2026-05-22 02:00:50

杨正康396 · 2026-05-22 02:00:50 发布

人机协作新纪元：AI Agent Harness Engineering 的角色定位与实践指南

副标题：从概念到落地，构建高效人机协同系统的核心工程方法

第一部分：引言与基础 (Introduction & Foundation)

1. 摘要/引言 (Abstract / Introduction)

问题陈述

随着大语言模型（LLM）和AI Agent技术的爆发式发展，AI正在从“工具”向“协作伙伴”演进。无论是编程助手（如GitHub Copilot X）、智能客服，还是医疗诊断辅助系统，AI Agent都在尝试与人类深度协作，完成复杂任务。然而，当前的实践面临诸多痛点：

不可控性：Agent容易产生“幻觉”（Hallucination），编造虚假信息；
交互生硬：人类与Agent的沟通成本高，缺乏自然、直观的协作体验；
性能不稳定：Agent在不同场景下的表现差异大，难以保证可靠性；
集成困难：Agent难以无缝融入现有业务系统，缺乏系统化的工程方法。

这些问题的核心在于：我们缺乏一套专门用于“驾驭”（Harness）AI Agent的工程体系——而这正是本文要探讨的AI Agent Harness Engineering。

核心方案

本文提出：AI Agent Harness Engineering是一门独立的工程学科，它以“实现高效、安全、可控的人机协作”为目标，涵盖Agent架构设计、交互体验设计、安全合规、性能优化、监控维护等全生命周期的工程方法。

我们将从理论到实践，层层递进：

明确Harness Engineering的核心概念与角色定位；
对比它与传统软件工程、ML Engineering、Prompt Engineering的关系；
通过一个完整的“智能文档协作助手”项目，展示如何在实际中应用Harness Engineering；
总结最佳实践与未来趋势。

主要成果/价值

读完本文，你将：

理解Harness Engineering在人机协作中的核心价值；
掌握Harness Engineering的核心概念、方法与工具链；
能够从零开始构建一个可落地的人机协作Agent系统；
避开Agent开发中的常见“坑”，遵循行业最佳实践。

文章导览

本文分为四个部分：

第一部分：引言与基础，介绍问题背景、目标读者与文章结构；
第二部分：核心内容，深入讲解Harness Engineering的概念、理论、环境准备与项目实现；
第三部分：验证与扩展，展示项目结果、最佳实践、常见问题与未来展望；
第四部分：总结与附录，回顾核心要点，提供参考资料与补充资源。

2. 目标读者与前置知识 (Target Audience & Prerequisites)

目标读者

本文适合以下人群：

软件工程师：希望在项目中集成AI Agent，构建人机协作系统；
AI/ML工程师：关注Agent的工程化落地与性能优化；
产品经理：希望理解Agent产品的设计逻辑与工程边界；
技术研究者：对人机协作与Agent工程化方向感兴趣。

前置知识

阅读本文需要具备：

基本的Python编程能力；
对大语言模型（LLM）、Prompt Engineering有初步了解；
基本的软件工程概念（如API设计、数据库使用）；
（可选）对LangChain等Agent开发框架有基础认知。

3. 文章目录 (Table of Contents)

第一部分：引言与基础
1. 摘要/引言
2. 目标读者与前置知识
3. 文章目录
第二部分：核心内容
4. 问题背景与动机
5. 核心概念与理论基础
6. 环境准备
7. 分步实现：智能文档协作助手
8. 关键代码解析与深度剖析
第三部分：验证与扩展
9. 结果展示与验证
10. 性能优化与最佳实践
11. 常见问题与解决方案
12. 未来展望与扩展方向
第四部分：总结与附录
13. 总结
14. 参考资料
15. 附录

第二部分：核心内容 (Core Content)

4. 问题背景与动机 (Problem Background & Motivation)

4.1 人机协作的时代已经到来

我们正在经历一场“人机协作”的革命：

编程领域：GitHub Copilot X让AI成为程序员的“结对编程伙伴”，自动生成代码、解释逻辑、修复bug；
客服领域：AI客服与人工客服协作，AI处理简单问题，人工处理复杂投诉，提升效率；
医疗领域：AI辅助医生分析影像、查阅文献，减少误诊率；
办公领域：AI助手帮助整理文档、撰写邮件、安排会议，释放人类的创造力。

根据Gartner的预测，到2025年，超过60%的企业将使用AI Agent来辅助员工完成日常工作。人机协作不再是“未来”，而是“现在”。

4.2 现有方案的局限性

然而，当前的Agent开发大多处于“手工作坊”阶段，缺乏系统化的工程方法，主要问题包括：

4.2.1 把Agent当“API调用工具”，缺乏体系化设计

很多团队只是简单地调用LLM API，写几行Prompt，就认为完成了Agent开发。但实际上：

Agent需要感知环境（如读取文档、访问数据库）；
Agent需要决策行动（如调用工具、生成内容）；
Agent需要保持上下文（如多轮对话、状态跟踪）。
简单的API调用无法满足这些需求，必须有专门的架构设计。

4.2.2 忽视人类体验，交互生硬

很多Agent产品的交互设计“以AI为中心”，而不是“以人类为中心”：

用户不知道如何与Agent沟通（比如“我该说什么才能让它听懂？”）；
Agent的回复过于技术化，不够自然；
缺乏反馈机制，用户无法纠正Agent的错误。
这些问题导致用户体验差，产品难以推广。

4.2.3 安全与合规风险被忽视

Agent的安全问题非常严峻：

幻觉问题：Agent编造虚假信息，可能导致业务损失（比如客服给用户错误的退款政策）；
隐私泄露：Agent可能无意中泄露用户的隐私数据（比如在回答中提到用户的病历信息）；
有害内容：Agent可能生成仇恨言论、暴力内容，违反法律法规。
很多团队在开发时没有考虑这些风险，直到出了问题才补救。

4.2.4 缺乏监控与迭代机制

Agent的性能不是“一劳永逸”的：

随着时间推移，用户的需求会变化，Agent的表现会下降；
缺乏监控，无法及时发现Agent的问题（比如响应时间过长、用户满意度下降）；
缺乏反馈收集，无法用用户的意见来优化Agent。
这些问题导致Agent的生命周期很短，难以持续产生价值。

4.3 为什么我们需要Harness Engineering？

“驾驭”（Harness）这个词的本意是“给马套上缰绳，让它按照人类的意愿前进”。AI Agent就像一匹“千里马”，能力很强，但如果没有缰绳，就会失控。Harness Engineering就是那根“缰绳”——它的作用是：

让Agent可控：确保Agent的行为符合人类的期望；
让协作高效：降低人类与Agent的沟通成本；
让系统可靠：保证Agent的性能稳定，安全合规；
让迭代持续：通过监控与反馈，持续优化Agent。

简单来说：LLM给了Agent“能力”，而Harness Engineering给了Agent“方向”。

5. 核心概念与理论基础 (Core Concepts & Theoretical Foundation)

5.1 核心概念定义

5.1.1 AI Agent

AI Agent是一个具有**感知（Perception）、决策（Decision Making）、行动（Action）**能力的智能体，它可以与环境（包括人类）交互，完成特定任务。

一个完整的Agent通常包含以下组件：

感知模块：获取环境信息（如读取文件、接收用户输入、访问数据库）；
推理/决策模块：根据感知到的信息，决定下一步做什么（如调用工具、生成内容）；
行动模块：执行决策（如调用API、发送消息、修改文件）；
记忆模块：保存历史信息（如对话上下文、任务状态）。

我们可以用一个简单的公式来描述Agent：
$A g e n t = P er ce pt i o n + R e a so nin g + A c t i o n + M e m or y$

例如，GitHub Copilot X就是一个Agent：

感知：读取当前代码文件、编辑器上下文、Git历史；
推理：根据上下文，决定生成什么代码；
行动：在编辑器中插入代码、解释代码、修复bug；
记忆：保存之前的交互历史，理解用户的编码习惯。

5.1.2 Harness Engineering

Harness Engineering是一门专门研究如何设计、开发、部署、监控和优化AI Agent，使其能够高效、安全、可控地与人类协作的工程学科。

它的核心目标是：最大化人机协作的总效用，同时最小化风险。

5.1.3 人机协作

人机协作是指人类与Agent共同完成任务，两者发挥各自的优势：

人类的优势：创造力、批判性思维、情感理解、道德判断；
Agent的优势：处理大量数据、快速推理、24小时工作、不会疲劳。

好的人机协作不是“Agent代替人类”，而是“Agent增强人类”——让人类专注于更有价值的工作，把重复性、机械性的工作交给Agent。

5.2 Harness Engineering的核心要素组成

Harness Engineering涵盖Agent全生命周期的工程活动，核心要素包括：

核心要素	描述
Agent架构设计	选择合适的Agent架构（如ReAct、Reflexion、Plan-and-Execute），定义组件边界。
交互体验设计	设计人类与Agent的交互方式（如对话、可视化、语音），确保自然、直观。
安全与合规管理	防止幻觉、隐私泄露、有害内容，确保符合法律法规。
性能优化	提高Agent的响应速度、准确率、可靠性，优化资源使用。
监控与维护	实时监控Agent的性能，收集反馈，及时修复问题，迭代优化。
系统集成	将Agent无缝融入现有业务系统（如CRM、ERP、IDE），提供标准API。

5.3 概念之间的关系：对比与架构图

5.3.1 相关概念对比：Harness Engineering vs 其他学科

为了更清晰地理解Harness Engineering的角色，我们将它与传统软件工程、ML Engineering、Prompt Engineering进行对比：

维度	传统软件工程	ML Engineering	Prompt Engineering	Harness Engineering
核心目标	构建可靠的软件系统	训练和部署ML模型	优化LLM的Prompt	驾驭Agent实现高效人机协作
主要任务	需求分析、编码、测试	数据处理、模型训练、部署	Prompt设计、调试、优化	Agent设计、交互设计、安全、监控、优化
技术栈	Java/Python/Go等	TensorFlow/PyTorch等	LLM API、Prompt工具	LangChain、AutoGPT、FastAPI等
关注重点	功能正确性、性能	模型准确率、泛化能力	Prompt效果、LLM响应	人机协作效率、Agent可控性、安全性
输出	软件应用	ML模型、API	优化后的Prompt	可协作的Agent系统、监控平台
与人类的关系	人类使用软件	人类标注数据、使用模型	人类设计Prompt	人类与Agent协作

从对比中可以看出：Harness Engineering不是对其他学科的替代，而是以Agent为核心的跨学科融合——它吸收了传统软件工程的工程化思维、ML Engineering的模型优化方法、Prompt Engineering的LLM交互技巧，形成了一套独立的工程体系。

5.3.2 概念联系的ER实体关系图

我们用ER图来展示Harness Engineering系统、AI Agent、人类用户、交互数据之间的关系：

5.3.3 交互关系图：人机协作的数据流

我们用序列图来展示人类用户、Harness Engineering系统、AI Agent、数据层之间的交互流程：

5.4 数学模型：人机协作的效用与Agent决策

5.4.1 人机协作的效用模型

我们用一个数学模型来描述人机协作的总效用。假设：

$H (t)$ ：人类在时间 $t$ 的贡献（如创造力、决策质量）；
$A (t)$ ：Agent在时间 $t$ 的贡献（如数据处理速度、准确率）；
$I (t)$ ：交互质量（如沟通成本、理解程度）；
$\alpha, \beta, \gamma$ ：权重系数（ $\gamma$ 表示协同效应的强度）。

那么，人机协作的总效用 $U (t)$ 可以表示为：
$\alpha H(t) + \beta A(t) + \gamma \cdot I(t) \cdot H(t) \cdot A(t)$

其中，第三项 $\gamma \cdot I(t) \cdot H(t) \cdot A(t)$ 是协同效应——当交互质量 $I (t)$ 很高时，人类和Agent的贡献会相互放大，产生“1+1>2”的效果。

Harness Engineering的核心目标就是最大化长期总效用：
$\max_{HE} \int_{0}^{T} U(t) dt$

同时，需要满足一系列约束条件：

安全约束： $\in SafeSet$ （Agent的行为必须在安全范围内）；
性能约束： $\leq R_{max}$ （响应时间不超过阈值）；
满意度约束： $\geq SAT_{min}$ （用户满意度不低于最低要求）。

5.4.2 Agent决策的POMDP模型

Agent的决策过程通常可以用**部分可观测马尔可夫决策过程（POMDP）**来描述，因为Agent往往只能感知到环境的部分信息。

POMDP的核心要素包括：

状态空间 $\mathcal{S}$ ：环境的所有可能状态（如用户的意图、任务的进度）；
观测空间 $\mathcal{O}$ ：Agent能够感知到的信息（如用户的输入、系统的反馈）；
动作空间 $\mathcal{A}$ ：Agent可以执行的动作（如生成回复、调用工具、结束对话）；
转移函数 $P (s^{'} ∣ s, a)$ ：在状态 $s$ 执行动作 $a$ 后，转移到状态 $s^{'}$ 的概率；
观测函数 $P (o ∣ s^{'}, a)$ ：在状态 $s^{'}$ 执行动作 $a$ 后，观测到 $o$ 的概率；
奖励函数 $R (s, a, s^{'})$ ：在状态 $s$ 执行动作 $a$ 转移到 $s^{'}$ 后，获得的奖励。

Harness Engineering的一个关键任务就是设计合适的奖励函数 $R$ ，让Agent的决策符合人类的期望。例如，在客服场景中：

如果Agent正确回答了用户的问题，给予高奖励；
如果Agent编造了信息，给予高惩罚；
如果Agent主动将复杂问题转交给人工，给予中等奖励。

通过优化奖励函数，我们可以“引导”Agent做出正确的决策。

5.5 Harness Engineering的工作流程

我们用流程图来展示Harness Engineering的完整工作流程：

这个流程是循环迭代的——Agent的优化不是一蹴而就的，需要根据监控数据和用户反馈持续改进。

6. 环境准备 (Environment Setup)

在进入实践之前，我们需要准备好开发环境。我们将使用Python作为主要编程语言，结合LangChain（Agent开发框架）、FastAPI（后端）、Streamlit（前端）、ChromaDB（向量数据库）来构建项目。

6.1 软件与库清单

软件/库	版本要求	用途
Python	≥3.10	编程语言
LangChain	0.1.10	Agent开发框架，提供组件化的Agent构建能力
LangChain-OpenAI	0.0.8	集成OpenAI的LLM与嵌入模型
LangChain-Chroma	0.1.2	集成ChromaDB向量数据库
FastAPI	0.110.0	后端API框架，高性能、易使用
Uvicorn	0.27.1	ASGI服务器，用于运行FastAPI
Streamlit	1.32.0	前端框架，快速构建数据应用与聊天界面
PyPDF	4.1.0	解析PDF文档
python-docx	1.1.0	解析Word文档
python-multipart	0.0.9	处理文件上传
python-dotenv	1.0.1	管理环境变量

6.2 安装步骤

6.2.1 创建虚拟环境

首先，我们创建一个Python虚拟环境，避免依赖冲突：

# Windows
python -m venv venv
venv\Scripts\activate

# Linux/Mac
python3 -m venv venv
source venv/bin/activate

6.2.2 安装依赖

创建requirements.txt文件，内容如下：

langchain==0.1.10
langchain-openai==0.0.8
langchain-chroma==0.1.2
fastapi==0.110.0
uvicorn==0.27.1
streamlit==1.32.0
pypdf==4.1.0
python-docx==1.1.0
python-multipart==0.0.9
python-dotenv==1.0.1

然后安装依赖：

pip install -r requirements.txt

6.2.3 配置环境变量

创建.env文件，配置必要的环境变量：

# OpenAI API密钥（需要在OpenAI官网申请）
OPENAI_API_KEY=your_openai_api_key_here

# ChromaDB向量数据库的存储路径
CHROMA_DB_PATH=./chroma_db

# 后端API的基础URL
API_BASE_URL=http://localhost:8000

注意：你需要替换your_openai_api_key_here为你自己的OpenAI API密钥（如果没有，可以在OpenAI官网申请）。如果你不想使用OpenAI，也可以替换为其他LLM（如Anthropic Claude、Llama 2），只需要修改对应的LangChain集成即可。

6.3 项目目录结构

我们将项目组织为以下结构：

ai-agent-harness-demo/
├── .env                      # 环境变量
├── requirements.txt          # 依赖清单
├── main.py                   # 后端API入口
├── streamlit_app.py          # 前端界面入口
├── agent/                    # Agent层模块
│   ├── document_processor.py # 文档处理模块
│   ├── rag_retriever.py      # RAG检索模块
│   ├── dialog_manager.py     # 对话管理模块
│   ├── response_generator.py # 响应生成模块
│   └── feedback_handler.py   # 反馈处理模块
└── database/                 # 数据库模块
    └── sqlite_db.py          # SQLite数据库操作

7. 分步实现：智能文档协作助手 (Step-by-Step Implementation)

为了让大家更好地理解Harness Engineering的实践，我们将从零开始构建一个智能文档协作助手项目。

7.1 项目介绍

这个项目的目标是：帮助用户上传和解析文档（PDF、Word、Markdown），与Agent进行多轮对话，询问文档相关的问题，获取内容生成建议（如摘要、续写），同时提供反馈机制与监控面板，持续优化Agent的性能。

7.2 系统功能设计

我们将系统分为以下核心功能模块：

功能模块	描述
文档管理	支持上传PDF/Word/Markdown文档，解析内容，分块存储到向量数据库。
智能问答	根据用户的问题，从文档中检索相关信息，生成准确的回答，支持多轮对话。
内容生成建议	生成文档摘要、续写段落、优化内容，帮助用户提高写作效率。
协作交互	用户可以对Agent的响应进行评分与评论，提供反馈。
性能监控	显示Agent的平均响应时间、用户满意度、总交互数等指标。

7.3 系统架构设计

我们采用分层架构，将系统分为表现层、API层、Agent层、数据层、监控层：

7.4 系统接口设计

我们设计以下核心RESTful API接口：

7.4.1 文档上传接口

路径：POST /api/documents
请求参数：
- file：上传的文件（multipart/form-data）；
- user_id：用户ID（query参数）。
响应：
- 成功（201 Created）：{"document_id": "xxx", "message": "Document uploaded successfully"}；
- 失败（400/500）：返回错误信息。

7.4.2 文档列表接口

路径：GET /api/documents
请求参数：
- user_id：用户ID（query参数）。
响应：
- 成功（200 OK）：{"documents": [{"document_id": "xxx", "name": "xxx", "upload_time": "xxx"}]}。

7.4.3 聊天接口

路径：POST /api/chat

请求参数：

{
  "user_id": "user_123",
  "session_id": "session_456（可选）",
  "message": "这篇文档的主要内容是什么？"
}

响应：

{
  "session_id": "session_456",
  "interaction_id": "inter_789",
  "response": "这篇文档主要介绍了...",
  "sources": [{"document_id": "xxx", "filename": "xxx", "content": "xxx"}]
}

7.4.4 反馈接口

路径：POST /api/feedback

请求参数：

{
  "user_id": "user_123",
  "session_id": "session_456",
  "interaction_id": "inter_789",
  "rating": 5,
  "comment": "回答非常准确！"
}

响应：
- 成功（201 Created）：{"message": "Feedback submitted successfully"}。

7.4.5 监控指标接口

路径：GET /api/metrics
请求参数：
- start_time：开始时间（ISO格式，如2024-01-01T00:00:00）；
- end_time：结束时间（ISO格式）。

响应：

{
  "avg_response_time": 1.2,
  "avg_rating": 4.5,
  "total_interactions": 100
}

7.5 核心代码实现

接下来，我们逐个实现各个模块。由于篇幅限制，我们只展示最核心的代码，完整代码可以在附录的GitHub链接中获取。

8. 关键代码解析与深度剖析 (Key Code Analysis & Deep Dive)

8.1 数据库模块：SQLiteDB

首先，我们实现SQLite数据库模块，用于存储文档信息、交互记录、反馈数据与监控指标。

文件位置：database/sqlite_db.py

import sqlite3
import json
from typing import List, Dict, Optional
from datetime import datetime

class SQLiteDB:
    def __init__(self, db_path: str = "./app.db"):
        self.db_path = db_path
        self.connection = None

    def _get_connection(self):
        """获取数据库连接（单例模式）"""
        if self.connection is None:
            self.connection = sqlite3.connect(self.db_path, check_same_thread=False)
            self.connection.row_factory = sqlite3.Row  # 让返回结果可以用字典访问
        return self.connection

    def init_tables(self):
        """初始化数据库表"""
        conn = self._get_connection()
        cursor = conn.cursor()

        # 1. 文档表：存储上传的文档信息
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS documents (
                document_id TEXT PRIMARY KEY,
                user_id TEXT NOT NULL,
                filename TEXT NOT NULL,
                upload_time TEXT NOT NULL
            )
        """)

        # 2. 交互记录表：存储用户与Agent的交互历史
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS interactions (
                interaction_id TEXT PRIMARY KEY,
                user_id TEXT NOT NULL,
                session_id TEXT NOT NULL,
                user_message TEXT NOT NULL,
                agent_response TEXT NOT NULL,
                sources TEXT,  -- 存储JSON格式的来源信息
                timestamp TEXT NOT NULL
            )
        """)

        # 3. 反馈表：存储用户的反馈
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS feedback (
                feedback_id TEXT PRIMARY KEY,
                user_id TEXT NOT NULL,
                session_id TEXT NOT NULL,
                interaction_id TEXT NOT NULL,
                rating INTEGER NOT NULL,
                comment TEXT,
                timestamp TEXT NOT NULL,
                FOREIGN KEY (interaction_id) REFERENCES interactions(interaction_id)
            )
        """)

        conn.commit()
        print("数据库表初始化成功！")

    # ------------------------------
    # 文档相关操作
    # ------------------------------
    def save_document(self, document_id: str, user_id: str, filename: str, upload_time: str):
        conn = self._get_connection()
        cursor = conn.cursor()
        cursor.execute("""
            INSERT INTO documents (document_id, user_id, filename, upload_time)
            VALUES (?, ?, ?, ?)
        """, (document_id, user_id, filename, upload_time))
        conn.commit()

    def get_documents(self, user_id: str) -> List[Dict]:
        conn = self._get_connection()
        cursor = conn.cursor()
        cursor.execute("""
            SELECT document_id, filename, upload_time FROM documents WHERE user_id = ?
        """, (user_id,))
        rows = cursor.fetchall()
        return [dict(row) for row in rows]

    # ------------------------------
    # 交互与反馈相关操作
    # ------------------------------
    def save_interaction(self, interaction_id: str, user_id: str, session_id: str,
                         user_message: str, agent_response: str, sources: List[Dict], timestamp: str):
        conn = self._get_connection()
        cursor = conn.cursor()
        sources_json = json.dumps(sources) if sources else None
        cursor.execute("""
            INSERT INTO interactions (interaction_id, user_id, session_id, user_message, agent_response, sources, timestamp)
            VALUES (?, ?, ?, ?, ?, ?, ?)
        """, (interaction_id, user_id, session_id, user_message, agent_response, sources_json, timestamp))
        conn.commit()

    def save_feedback(self, feedback_id: str, user_id: str, session_id: str,
                      interaction_id: str, rating: int, comment: Optional[str], timestamp: str):
        conn = self._get_connection()
        cursor = conn.cursor()
        cursor.execute("""
            INSERT INTO feedback (feedback_id, user_id, session_id, interaction_id, rating, comment, timestamp)
            VALUES (?, ?, ?, ?, ?, ?, ?)
        """, (feedback_id, user_id, session_id, interaction_id, rating, comment, timestamp))
        conn.commit()

    # ------------------------------
    # 监控指标相关操作
    # ------------------------------
    def get_metrics(self, start_time: str, end_time: str) -> Dict:
        conn = self._get_connection()
        cursor = conn.cursor()

        # 1. 总交互数
        cursor.execute("""
            SELECT COUNT(*) as total FROM interactions WHERE timestamp BETWEEN ? AND ?
        """, (start_time, end_time))
        total_interactions = cursor.fetchone()["total"] or 0

        # 2. 平均用户评分
        cursor.execute("""
            SELECT AVG(rating) as avg_rating FROM feedback WHERE timestamp BETWEEN ? AND ?
        """, (start_time, end_time))
        avg_rating = cursor.fetchone()["avg_rating"] or 0.0

        # 3. 平均响应时间（这里简化处理，实际项目中应该在交互记录中存储响应时间）
        avg_response_time = 1.5

        return {
            "avg_response_time": round(avg_response_time, 2),
            "avg_rating": round(avg_rating, 2),
            "total_interactions": total_interactions
        }

代码解析：

我们使用SQLite作为关系数据库，因为它轻量、无需额外安装，适合演示项目；
init_tables方法初始化三个核心表：documents（文档）、interactions（交互）、feedback（反馈）；
get_metrics方法计算监控指标，实际项目中应该从交互记录中获取真实的响应时间。

8.2 Agent层：文档处理模块（DocumentProcessor）

这个模块负责解析上传的文档，将内容分块，向量化后存储到ChromaDB向量数据库中。

文件位置：agent/document_processor.py

import tempfile
import os
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import PyPDFLoader, Docx2txtLoader, TextLoader
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma

class DocumentProcessor:
    def __init__(self, chroma_db_path: str = "./chroma_db"):
        # 1. 初始化文本分割器：将长文档分成小块，便于LLM处理
        self.text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=1000,    # 每个块的大小（字符数）
            chunk_overlap=200,  # 块之间的重叠，保持上下文连续性
            length_function=len
        )

        # 2. 初始化嵌入模型：将文本转换为向量
        self.embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

        # 3. 初始化向量存储：ChromaDB
        self.vector_store = Chroma(
            persist_directory=chroma_db_path,
            embedding_function=self.embeddings,
            collection_name="document_chunks"
        )

    def process_document(self, document_id: str, filename: str, content: bytes, user_id: str):
        """
        处理上传的文档：
        1. 根据文件类型选择加载器；
        2. 加载文档内容；
        3. 分割文档为小块；
        4. 向量化并存储到ChromaDB。
        """
        # 获取文件扩展名
        ext = os.path.splitext(filename)[1].lower()

        # 创建临时文件（因为LangChain的加载器需要文件路径）
        with tempfile.NamedTemporaryFile(delete=False, suffix=ext) as temp_file:
            temp_file.write(content)
            temp_file_path = temp_file.name

        try:
            # 1. 选择加载器
            if ext == ".pdf":
                loader = PyPDFLoader(temp_file_path)
            elif ext == ".docx":
                loader = Docx2txtLoader(temp_file_path)
            elif ext in [".md", ".txt"]:
                loader = TextLoader(temp_file_path)
            else:
                raise ValueError(f"不支持的文件类型：{ext}")

            # 2. 加载文档
            documents = loader.load()
            print(f"成功加载文档：{filename}，共{len(documents)}页")

            # 3. 分割文档为小块
            chunks = self.text_splitter.split_documents(documents)
            print(f"文档分割为{len(chunks)}个块")

            # 4. 为每个块添加元数据（便于后续过滤）
            for chunk in chunks:
                chunk.metadata["document_id"] = document_id
                chunk.metadata["filename"] = filename
                chunk.metadata["user_id"] = user_id

            # 5. 向量化并存储到ChromaDB
            self.vector_store.add_documents(chunks)
            print(f"文档向量化完成，已存储到ChromaDB")

        finally:
            # 删除临时文件
            os.unlink(temp_file_path)

代码解析：

文本分割：我们使用RecursiveCharacterTextSplitter，它会优先按照段落、句子、单词的顺序分割文本，保持上下文的完整性；
向量化：我们使用OpenAI的text-embedding-3-small模型，它将文本转换为1536维的向量；
向量存储：ChromaDB是一个轻量级的向量数据库，适合本地开发与演示；
元数据：我们为每个文档块添加了user_id、document_id等元数据，便于后续只检索当前用户的文档。

8.3 Agent层：响应生成模块（ResponseGenerator）

这是Agent的核心模块，负责根据用户的问题，检索相关文档，调用LLM生成响应。

文件位置：agent/response_generator.py

from typing import List, Dict
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.schema import HumanMessage, AIMessage
from agent.rag_retriever import RAGRetriever

class ResponseGenerator:
    def __init__(self, openai_api_key: str, retriever: RAGRetriever):
        # 1. 初始化LLM：使用GPT-4（如果预算有限，可以用GPT-3.5-turbo）
        self.llm = ChatOpenAI(
            model="gpt-4",
            temperature=0.7,  # 控制生成的随机性：0=最确定，1=最随机
            api_key=openai_api_key
        )

        # 2. 初始化RAG检索器
        self.retriever = retriever

        # 3. 设计Prompt模板：这是Harness Engineering的关键！
        self.prompt_template = ChatPromptTemplate.from_messages([
            (
                "system",
                """你是一个专业的智能文档协作助手，帮助用户处理文档相关的任务。
请遵循以下规则：
1. **仅根据上下文回答**：所有回答必须基于提供的文档上下文，不得编造信息。如果上下文中没有相关内容，请诚实告诉用户：“抱歉，我在上传的文档中没有找到相关信息。”
2. **引用来源**：在回答的末尾，引用相关的文档来源（格式：“来源：文档名称”）。
3. **语言友好**：回答要清晰、简洁、口语化，避免使用过于技术化的术语。
4. **多轮对话**：记住之前的对话历史，保持上下文的连贯性。""",
            ),
            MessagesPlaceholder(variable_name="chat_history"),  # 插入对话历史
            (
                "human",
                """以下是来自用户上传文档的相关上下文：
------------------------------
{context}
------------------------------
用户的问题/请求：{question}""",
            ),
        ])

    def generate_response(self, user_message: str, chat_history: List[Dict], user_id: str) -> Dict:
        """
        生成Agent的响应：
        1. 检索相关文档块；
        2. 构建Prompt；
        3. 调用LLM生成响应；
        4. 返回响应与来源。
        """
        # 1. 检索相关文档块（仅当前用户的文档）
        sources = self.retriever.retrieve(user_message, user_id)
        print(f"检索到{len(sources)}个相关文档块")

        # 2. 构建上下文字符串
        context = "\n\n".join([
            f"【来源：{source['filename']}】\n{source['content']}"
            for source in sources
        ])

        # 3. 将对话历史转换为LangChain的消息格式
        langchain_history = []
        for msg in chat_history:
            if msg["role"] == "user":
                langchain_history.append(HumanMessage(content=msg["content"]))
            elif msg["role"] == "assistant":
                langchain_history.append(AIMessage(content=msg["content"]))

        # 4. 构建Prompt
        prompt = self.prompt_template.format_messages(
            chat_history=langchain_history,
            context=context,
            question=user_message
        )

        # 5. 调用LLM生成响应
        print("正在调用LLM生成响应...")
        response = self.llm.invoke(prompt)

        # 6. 返回结果
        return {
            "response": response.content,
            "sources": sources
        }