trending/MathModelAgent
1
0
Fork 0
mirror of https://github.com/jihe520/MathModelAgent.git synced 2026-10-02 02:44:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix docs and interpreter, update tests

Browse Source
This commit is contained in:
Sanjin
2025-06-06 14:25:40 +08:00
parent ce7d8eb456
commit 9e8b5d79c9
5 changed files with 42 additions and 160 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+5 -5
View File
@@ -24,12 +24,12 @@
## ✨ 功能特性
- 🔍 自动分析问题，数学建模，编写代码，纠正错误，撰写论文
- 💻 Code Interperter
- loacl Interperter: 基于 jupyter , 代码保存为 notebook 方便再编辑
- 云端 code interperter: [E2B](https://e2b.dev/) 和 [daytona](https://app.daytona.io/)
- 💻 Code Interpreter
- local Interpreter: 基于 jupyter , 代码保存为 notebook 方便再编辑
- 云端 code interpreter: [E2B](https://e2b.dev/) 和 [daytona](https://app.daytona.io/)
- 📝 生成一份编排好格式的论文
- 🤝 muti-agents: 建模手，代码手，论文手等
- 🔄 muti-llms: 每个 agent 设置不同的、合适的模型
- 🤝 multi-agents: 建模手，代码手，论文手等
- 🔄 multi-llms: 每个 agent 设置不同的、合适的模型
- 🤖 支持所有模型: [litellm](https://docs.litellm.ai/docs/providers)
- 💰 成本低：workflow agentless，不依赖 agent 框架
- 🧩 自定义模板：prompt inject 为每个 subtask 单独设置需求
backend/app/tests/test_cite.py
-146
View File
@@ -1,146 +0,0 @@
import re
import os
import json
def split_footnotes(text: str) -> tuple[str, list[tuple[str, str]]]:
"""分离正文和脚注定义"""
main_text = re.sub(
r"\n\[\^\d+\]:.*?(?=\n\[\^|\n\n|\Z)", "", text, flags=re.DOTALL
).strip()
# 匹配脚注定义
footnotes = re.findall(r"\[\^(\d+)\]:\s*(.+?)(?=\n\[\^|\n\n|\Z)", text, re.DOTALL)
return main_text, footnotes
def extract_footnote_references(text: str) -> list[tuple[str, str]]:
"""提取文本中的脚注引用和对应的脚注内容"""
# 匹配脚注引用模式：[^数字]: 内容（处理多种可能的结束方式）
# 支持以句号、换行或文本结尾结束的脚注
pattern = r"\[\^(\d+)\]:\s*([^[\n]+?)(?=\[\^|\n\n|\n-|\n\d+\.|\n\*|$)"
matches = re.findall(pattern, text, re.DOTALL)
# 清理匹配结果
cleaned_matches = []
for num, content in matches:
# 移除末尾的句号和多余空白
content = content.strip()
if content.endswith("。"):
content = content[:-1]
if content.endswith("."):
content = content[:-1]
cleaned_matches.append((num, content))
return cleaned_matches
class UserOutput:
def __init__(self):
self.res: dict[str, dict] = {}
self.cost_time = 0.0
self.initialized = True
def get_result_to_save(self, ques_count):
# 保存 res.json 文件
# 动态顺序获取拼接res value，正确拼接顺序
ques_str = [f"ques{i}" for i in range(1, ques_count + 1)]
# 修改：调整章节顺序，确保符合论文结构
seq = [
"firstPage", # 标题、摘要、关键词
"RepeatQues", # 一、问题重述
"analysisQues", # 二、问题分析
"modelAssumption", # 三、模型假设
"symbol", # 四、符号说明和数据预处理
"eda", # 四、数据预处理（EDA部分）
*ques_str, # 五、模型的建立与求解（问题1、2...）
"sensitivity_analysis", # 六、模型的分析与检验
"judge", # 七、模型的评价、改进与推广
]
# 全局文献列表，用于去重和统一编号
global_footnotes = [] # 存储所有唯一的文献内容
footnote_content_to_id = {} # 文献内容到全局ID的映射
# 处理后的各节内容
processed_sections = []
for key in seq:
if key not in self.res:
continue
content = self.res[key]["response_content"]
print(f"\n处理节: {key}")
# 1. 提取当前节的文献引用
current_footnotes = extract_footnote_references(content)
print(f"当前节文献: {current_footnotes}")
# 2. 处理每个文献引用
section_footnote_mapping = {} # 当前节内的旧编号到新编号映射
for old_num, footnote_content in current_footnotes:
# 清理文献内容
footnote_content = footnote_content.strip()
if footnote_content.endswith("。"):
footnote_content = footnote_content[:-1]
# 3. 检查是否已存在于全局列表
if footnote_content in footnote_content_to_id:
# 文献已存在，使用已有的全局编号
global_id = footnote_content_to_id[footnote_content]
print(f"文献已存在: {footnote_content} -> 使用编号 {global_id}")
else:
# 新文献，添加到全局列表
global_footnotes.append(footnote_content)
global_id = len(global_footnotes)
footnote_content_to_id[footnote_content] = global_id
print(f"新文献: {footnote_content} -> 分配编号 {global_id}")
section_footnote_mapping[old_num] = global_id
# 4. 更新当前节中的脚注引用编号
processed_content = content
# 更新引用编号
for old_num, new_num in section_footnote_mapping.items():
# 替换文中的引用，但不处理脚注定义
processed_content = re.sub(
rf"\[\^{old_num}\](?!:)", f"[^{new_num}]", processed_content
)
# 移除原有的脚注定义（以换行开头的完整定义行）
processed_content = re.sub(
r"\n\[\^\d+\]:\s*[^[\n]+?(?=\n\[\^|\n\n|\n-|\n\d+\.|\n\*|$)",
"",
processed_content,
flags=re.DOTALL,
).strip()
processed_sections.append(processed_content)
print("processed_sections:", processed_sections)
print(f"节 {key} 脚注映射: {section_footnote_mapping}")
# 5. 合并所有处理后的内容
full_content = "\n\n".join(processed_sections)
# 6. 添加重新编号后的脚注到文档末尾
if global_footnotes:
full_content += "\n\n## 参考文献\n\n"
for i, footnote_content in enumerate(global_footnotes, 1):
full_content += f"[^{i}]: {footnote_content}\n\n"
print(f"\n最终全局文献列表: {global_footnotes}")
return full_content
if __name__ == "__main__":
with open("res.json", "r", encoding="utf-8") as f:
res = json.load(f)
user_output = UserOutput()
user_output.res = res
with open("res.md", "w", encoding="utf-8") as f:
f.write(user_output.get_result_to_save(ques_count=1))
backend/app/tests/test_common_utils.py
+15
View File
@@ -0,0 +1,15 @@
import unittest
from app.utils.common_utils import split_footnotes
class TestCommonUtils(unittest.TestCase):
def test_split_footnotes(self):
text = "Example[^1]\n\n[^1]: Footnote content"
main, notes = split_footnotes(text)
self.assertEqual(main, "Example")
self.assertEqual(notes, [("1", "Footnote content")])
if __name__ == "__main__":
unittest.main()
backend/app/tests/test_e2b.py
+14 -6
View File
@@ -1,22 +1,28 @@
import os
import asyncio
import unittest
from dotenv import load_dotenv
from app.tools.e2b_interpreter import E2BCodeInterpreter
from app.utils.common_utils import create_task_id, create_work_dir
from app.utils.common_utils import create_work_dir
from app.tools.notebook_serializer import NotebookSerializer
class TestE2BCodeInterpreter(unittest.TestCase):
def setUp(self):
load_dotenv()
_, dirs = create_work_dir("20250312-104132-d3625cab")
notebook = NotebookSerializer(dirs["jupyter"])
self.task_id = "20250312-104132-d3625cab"
self.work_dir = create_work_dir(self.task_id)
notebook = NotebookSerializer(self.work_dir)
self.code_interpreter = E2BCodeInterpreter(
dirs, "20250312-104132-d3625cab", notebook
self.task_id, self.work_dir, notebook
)
def test_execute_code(self):
if not os.getenv("E2B_API_KEY"):
self.skipTest("E2B_API_KEY not set")
code = """
import matplotlib.pyplot as plt
import numpy as np
@@ -39,6 +45,8 @@ plt.grid(True)
plt.legend()
# 显示图像
plt.show()
plt.show()
"""
self.code_interpreter.execute_code(code)
asyncio.run(self.code_interpreter.initialize())
asyncio.run(self.code_interpreter.execute_code(code))
backend/app/tools/local_interpreter.py
+8 -3
View File
@@ -219,9 +219,14 @@ class LocalCodeInterpreter(BaseCodeInterpreter):
self.interrupt_signal = True
def restart_jupyter_kernel(self):
self.kernel_client.shutdown()
self.kernel_manager, self.kernel_client = (
jupyter_client.manager.start_new_kernel(kernel_name="python3")
"""Restart the Jupyter kernel and recreate the work directory."""
self.kc.shutdown()
self.km, self.kc = jupyter_client.manager.start_new_kernel(
kernel_name="python3"
)
self.interrupt_signal = False
self._create_work_dir()
def _create_work_dir(self):
"""Ensure the working directory exists after a restart."""
os.makedirs(self.work_dir, exist_ok=True)