| 这个项目属于哪个课程 | 2025数据采集与融合技术实践 |
|---|---|
| 组名、项目简介 | 组名:暴虎 项目需求:“京东多智能体挑战赛”旨在推动人工智能（AI）技术在实际生产与生活场景中的应用创新，探索多智能体在复杂业务场景下的协同决策能力。基于京东零售开源的多智能体（Multi-Agent）协作框架OxyGent，构建高效协作的多智能体系统，以应对现实环境中的不确定性和多目标优化问题。 项目目标:采用公开验证集和私有测试集相结合的评测方式，确保模型调优的科学性与最终能力的公正评估。为全面考察多智能体系统的综合能力，设计了分级任务体系，并覆盖多种实际应用场景， 项目开展技术路线 :基于OxyGent框架，以数据采集为核心，先通过多模态/网页采集器获取并预处理数据，经任务分发Agent分配后，由理解/检索/推理/执行Agent协同处理，接入DeepSeek LLM并通过超强清洗器优化输出格式，最终生成符合要求的提交结果。 |
| 团队成员学号 | 102302117冯大河，102302154杨如意，102302155张怡旋，102302156李子贤，102202110叶志杰 |
| 这个项目的目标 | 分级任务体系:1.要求智能体依托大语言模型（LLM）或基础工具，完成基本推理与信息处理任务；2.要求智能体能够准确调用多类工具，并整合多源信息，提升任务完成的准确性和效率；3.要求智能体实现复杂的多步规划、深度推理及高效协作，解决更具挑战性的实际问题。多模态内容理解： 精准解析文本、图像、音频、PDF文档、表格及网页等多模态信息，实现商品详情页图文、用户评论、视频内容的深度理解与结构化信息提取。 |
| 其他参考文献 | 京东多智能体挑战赛 |
一 项目背景
本实践报告基于京东多智能体挑战赛，该竞赛旨在推动多智能体系统在复杂场景下的协同决策与创新应用。竞赛数据集包含100条验证集样本和200条测试集样本，涵盖多模态理解、信息检索、逻辑推理、代码生成与终端交互等多种任务类型，难度分为基础、进阶与复杂三个等级。
作为本科生参赛者，目标是设计并实现一个基于开源框架OxyGent的多智能体系统，重点解决数据采集与智能体协作问题。
二 项目架构设计
项目采用分层架构设计，主要包括以下三个层次：
- 数据采集与预处理层
- 多模态数据采集器（处理图像、PDF、表格等格式）
- 网页数据采集器（支持静态和动态网页）
- 数据预处理器（标准化和清洗数据）
- 多智能体协作层（基于OxyGent理念）
- 任务分发Agent（根据难度分配策略）
- 理解Agent（多模态内容解析）
- 检索Agent（信息搜索与整合）
- 推理Agent（逻辑分析与决策）
- 执行Agent（代码生成与运行）
- 结果生成与优化层
- 答案后处理器（清洗和标准化）
- 结果验证器（格式检查）
- 性能分析器（准确率计算）
三 项目分工
- 冯大河：PPT 讲解、项目演示视频录制与后期：对接杨如意熟悉 PPT 内容，编写讲解脚本；协调叶志杰完成系统演示操作；视频录制。
- 叶志杰：项目源代码开发、调试与优化：提供代码核心逻辑、技术架构说明（支撑 PPT 和实践报告）；编写代码注释与 README.md；配合演示视频录制，确保系统可正常演示。
- 杨如意：最终汇报 PPT 全流程制作：对接叶志杰提取代码技术细节，对接张怡旋、李子贤提取实践报告核心成果；设计 PPT 框架与可视化元素；优化 PPT 逻辑与演示适配性。
- 张怡旋：项目总报告撰写、实践报告统筹：主导撰写总报告（赛题分析、系统设计、成果总结）；整合李子贤实践报告内容，确保前后一致；负责项目源代码归档和复现文档编写，详细整理了依赖配置与运行流程说明，保障了项目的可复现性和代码管理的规范性。
- 李子贤：实践报告核心撰写、数据与文档整理：撰写技术实现细节、实验过程、问题解决方案；整理实验数据、测试报告（支撑总报告）；校验提交材料的完整性与合规性
四 个人贡献
技术实现细节
- 多模态数据采集模块
点击查看代码
python
class MultiModalCollector:def auto_collect(self, file_path: str) -> Dict[str, Any]:"""自动识别文件类型并采集支持格式：图像(.jpg, .png, .gif等)、PDF、Excel、CSV、文本文件"""ext = os.path.splitext(file_path)[1].lower()if ext in ['.png', '.jpg', '.jpeg']:return self.collect_image(file_path)elif ext == '.pdf':return self.collect_pdf(file_path)elif ext in ['.xlsx', '.xls']:return self.collect_excel(file_path)elif ext == '.csv':return self.collect_csv(file_path)elif ext in ['.txt', '.md']:return self.collect_text(file_path)else:return {"type": "unknown", "path": file_path}def collect_excel(self, excel_path: str) -> Dict[str, Any]:"""采集Excel数据，支持多sheet读取"""try:excel_file = pd.ExcelFile(excel_path)all_data = {}for sheet_name in excel_file.sheet_names:df = pd.read_excel(excel_path, sheet_name=sheet_name)all_data[sheet_name] = {"shape": df.shape,"columns": df.columns.tolist(),"data": df.to_dict('records')}return {"type": "table","format": "excel","sheets": list(all_data.keys()),"shape": all_data[first_sheet]["shape"],"columns": all_data[first_sheet]["columns"],"head": all_data[first_sheet]["data"][:20]}except Exception as e:return {"type": "error", "message": str(e)}
- 任务分发策略
任务分发器根据任务难度和类型选择不同的处理策略:
点击查看代码
python
class TaskDispatcher:def dispatch(self, task: Dict[str, Any]) -> Dict[str, Any]:query = task.get("query", "").lower()level = task.get("level", "1")has_file = "file_name" in task and task.get("file_name")# 基础策略配置strategy = {"type": "general","require_agents": ["understanding", "reasoning"],"priority": "normal"}# 计算类任务if any(kw in query for kw in ["计算", "多少", "总和", "平均"]):strategy["type"] = "calculation"strategy["require_agents"] = ["understanding", "reasoning", "execution"]# 多模态任务elif has_file:strategy["type"] = "multimodal_understanding"strategy["require_agents"] = ["understanding", "retrieval", "reasoning"]# 复杂任务（level≥3）elif int(level) >= 3:strategy["type"] = "complex"strategy["require_agents"] = ["understanding", "retrieval", "reasoning", "execution"]strategy["priority"] = "high"return strategy
- 推理Agent实现
推理Agent是系统的核心，负责执行逻辑推理：
点击查看代码
python
class ReasoningAgent:def reason(self, task: Dict[str, Any], understanding: Dict[str, Any],retrieval_results: Dict[str, Any] = None) -> Dict[str, Any]:"""执行推理，支持多种问题类型"""question_type = understanding.get("query_analysis", {}).get("question_type", "general")if question_type == "calculation":return self._reason_calculation(task, understanding)elif question_type == "comparison":return self._reason_comparison(task, understanding, retrieval_results)elif question_type == "extraction":return self._reason_extraction(task, understanding, retrieval_results)else:return self._reason_general(task, understanding, retrieval_results)def _reason_calculation(self, task: Dict[str, Any], understanding: Dict[str, Any]) -> Dict[str, Any]:"""计算类推理：优先使用精确计算器"""query = task.get("query", "")collected_data = task.get("collected_data", {})# 提取数字numbers = self._extract_numbers(query)# 从表格数据中提取数字if collected_data and "file" in collected_data:file_data = collected_data["file"]if file_data.get("type") == "table":numbers.extend(self._extract_from_table(file_data, query))# 执行计算if numbers:# 识别运算类型if "总和" in query or "求和" in query:result = sum(numbers)elif "平均" in query:result = sum(numbers) / len(numbers)elif "最大" in query:result = max(numbers)elif "最小" in query:result = min(numbers)else:result = numbers[0]# 格式化结果if isinstance(result, float):if result.is_integer():result_str = str(int(result))else:result_str = f"{result:.2f}"else:result_str = str(result)else:result_str = "无法计算"return {"task_id": task.get("task_id"),"reasoning_steps": ["提取数字", "识别运算", "执行计算"],"final_answer": result_str,"confidence": 0.9}def _extract_numbers(self, text: str) -> List[float]:"""从文本中提取所有数字"""import renumbers = re.findall(r'\d+\.?\d*', text)return [float(n) for n in numbers if n]
- 答案后处理器
为了确保答案格式符合比赛要求，实现了答案后处理器:
点击查看代码
python
class AnswerProcessor:def process(self, raw_answer: Any) -> str:"""处理原始答案，确保格式符合要求"""if raw_answer is None:return ""answer = str(raw_answer).strip()# 1. 移除常见前缀prefixes = ["答案是", "答案为", "答案：", "答案:", "结果是", "结果为"]for prefix in prefixes:if answer.lower().startswith(prefix.lower()):answer = answer[len(prefix):].strip()# 2. 移除标点符号answer = answer.rstrip('。.！!？?，,')# 3. 提取纯数字（如果是数字答案）if self._is_numeric_answer(answer):answer = self._extract_number(answer)# 4. 提取选项（如果是选择题）if self._is_choice_answer(answer):answer = self._extract_choice(answer)return answer.strip()def _is_numeric_answer(self, text: str) -> bool:"""判断是否是数字答案"""import rehas_number = bool(re.search(r'\d', text))is_short = len(text) < 50return has_number and is_shortdef _extract_number(self, text: str) -> str:"""提取数字"""import renumbers = re.findall(r'[-+]?\d+(?:\.\d+)?', text)if numbers:# 返回最后一个数字（通常是结果）number = numbers[-1]try:num = float(number)if num.is_integer():return str(int(num))else:return f"{num:.2f}".rstrip('0').rstrip('.')except:return numberreturn text
实验过程
- 实验设置
- 硬件环境：
- CPU: Intel Core i7-12700H
- 内存: 32GB
- 操作系统: Windows 11
- 软件环境：
- Python 3.10
- 主要依赖库：pandas, numpy, openai, jsonlines, Pillow
- 数据设置：
- 验证集：100个任务（包含答案）
- 测试集：200个任务（不包含答案）
- 实践步骤
- 环境搭建
点击查看代码
bash
# 创建虚拟环境
python -m venv venv
source venv/bin/activate # Windows: venv\Scripts\activate# 安装依赖
pip install pandas numpy jsonlines Pillow requests beautifulsoup4 openpyxl PyPDF2
- 数据准备
点击查看代码
python
# 将比赛数据组织为以下结构
# data/valid/data.jsonl # 验证集
# data/valid/files/ # 验证集附加文件
# data/test/data.jsonl # 测试集
# data/test/files/ # 测试集附加文件
- 数据处理
点击查看代码
bash
# 运行数据预处理
python main.py# 输出结果在cache_dir/目录下
# - processed_valid.jsonl # 处理后的验证集
# - processed_test.jsonl # 处理后的测试集
- 模型训练与验证
点击查看代码
bash
# 在验证集上训练和调试
python train.py# 查看训练结果
cat output/valid_results.jsonl
- 生成提交文件
点击查看代码
bash
# 在测试集上推理
python inference.py# 生成的提交文件
# output/result.jsonl
问题解决方案
- LLM输出格式不规范
问题描述：
在初步测试中，发现DeepSeek API返回的答案经常包含多余的解释性文字，例如：
输入："1+1等于多少？"
输出："1+1的计算结果是2。这是一个基本的数学运算。"
解决方案：
设计了专门的答案后处理器
点击查看代码
python
class AnswerProcessor:def process(self, raw_answer: Any) -> str:"""清洗和标准化答案"""if raw_answer is None:return ""answer = str(raw_answer).strip()# 1. 移除常见前缀prefixes_to_remove = ["答案是", "答案为", "答案：", "答案:", "结果是", "结果为"]for prefix in prefixes_to_remove:if answer.lower().startswith(prefix.lower()):answer = answer[len(prefix):].strip()# 2. 移除常见后缀suffixes_to_remove = ["。", ".", "！", "!", "是正确的", "是答案"]for suffix in suffixes_to_remove:if answer.endswith(suffix):answer = answer[:-len(suffix)].strip()# 3. 提取数字（如果是数字答案）if self._is_numeric_answer(answer):answer = self._extract_number(answer)# 4. 提取选项（如果是选择题）if self._is_choice_answer(answer):answer = self._extract_choice(answer)return answer.strip()
- 表格数据处理不充分
问题描述：
在处理包含表格文件的任务时，系统未能有效提取相关数据，导致答案准确率低。
解决方案：
增强表格数据提取逻辑:
点击查看代码
python
def extract_from_table(self, table_data: Dict, query: str) -> List[float]:"""从表格智能提取相关数字"""numbers = []# 提取查询中的关键词keywords = self._extract_keywords(query)if "head" in table_data:for row in table_data["head"]:# 检查行是否相关row_text = str(row).lower()is_relevant = any(kw.lower() in row_text for kw in keywords if len(kw) > 1)if is_relevant or not keywords: # 如果没有关键词，提取所有数字for key, value in row.items():try:# 清理并转换value_str = str(value).replace(',', '').replace('¥', '').replace('$', '')num = float(value_str)numbers.append(num)except:passreturn numbers
- API调用不稳定
问题描述：
在大量调用DeepSeek API时，偶尔会出现超时或连接错误。
解决方案：
实现重试机制和错误处理：
点击查看代码
python
def chat_with_retry(self, messages: List[Dict[str, str]], max_retries: int = 3) -> str:"""带重试的API调用"""for attempt in range(max_retries):try:response = self.client.chat.completions.create(model=self.model,messages=messages,temperature=0.0, # 使用确定性输出max_tokens=100)return response.choices[0].message.contentexcept Exception as e:if attempt < max_retries - 1:import timetime.sleep(1) # 等待1秒后重试continueelse:return f"API调用失败: {str(e)}"
五 个人心得
参与京东多智能体挑战赛的综合实践项目，让我在数据采集与融合技术领域收获了宝贵的实战经验。从项目搭建到问题攻坚，每一步都充满了成长与启发。
在技术实现上，我深刻体会到分层架构设计的重要性。基于 OxyGent 框架构建的多智能体系统，通过数据采集、智能体协作、结果优化三层架构，让复杂任务得以有序拆解。尤其是多模态数据采集模块的开发，让我熟练掌握了图像、PDF、表格等异构数据的处理方法，理解了数据预处理对于后续任务的关键支撑作用。
项目推进中，我也遇到了不少挑战。但通过团队协作，我们成功攻克了这些难题。这个过程让我明白，实际项目中技术落地往往需要不断调试优化，而问题解决能力和创新思维正是工程实践的核心。
此外，团队分工协作的模式让我受益匪浅。从代码开发、报告撰写到 PPT 制作、视频录制，每位成员各司其职又相互配合。在与队友对接技术细节、整合项目成果的过程中，我不仅提升了技术表达能力，更学会了从全局视角看待项目，理解了团队协作对于复杂项目成功的重要意义。
)
