""" 研究报告数据模型 """ import uuid from datetime import datetime from typing import List, Optional, Dict, Any from pydantic import BaseModel, Field class ReportSection(BaseModel): """报告章节""" title: str content: str subsections: List['ReportSection'] = [] sources: List[str] = [] # URL列表 def to_markdown(self, level: int = 1) -> str: """转换为Markdown格式""" header = "#" * level markdown = f"{header} {self.title}\n\n{self.content}\n\n" # 添加子章节 for subsection in self.subsections: markdown += subsection.to_markdown(level + 1) # 添加来源 if self.sources: markdown += f"\n{'#' * (level + 1)} 参考来源\n" for i, source in enumerate(self.sources, 1): markdown += f"{i}. [{source}]({source})\n" markdown += "\n" return markdown # 允许递归引用 ReportSection.model_rebuild() class KeyInsight(BaseModel): """关键洞察""" insight: str supporting_evidence: List[str] = [] source_urls: List[str] = [] confidence: float = 0.0 # 0-1之间 class SubtopicReport(BaseModel): """子主题报告""" subtopic_id: str subtopic_name: str sections: List[ReportSection] = [] key_insights: List[KeyInsight] = [] recommendations: List[str] = [] created_at: datetime = Field(default_factory=datetime.now) word_count: int = 0 def to_markdown(self) -> str: """转换为Markdown格式""" markdown = f"## {self.subtopic_name}\n\n" # 添加各个章节 for section in self.sections: markdown += section.to_markdown(level=3) # 添加关键洞察 if self.key_insights: markdown += "### 关键洞察\n\n" for i, insight in enumerate(self.key_insights, 1): markdown += f"{i}. **{insight.insight}**\n" if insight.supporting_evidence: for evidence in insight.supporting_evidence: markdown += f" - {evidence}\n" if insight.source_urls: markdown += f" - 来源: " markdown += ", ".join([f"[{i+1}]({url})" for i, url in enumerate(insight.source_urls)]) markdown += "\n" markdown += "\n" # 添加建议 if self.recommendations: markdown += "### 建议与展望\n\n" for recommendation in self.recommendations: markdown += f"- {recommendation}\n" markdown += "\n" return markdown class HallucinationCheck(BaseModel): """幻觉检查记录""" content: str source_url: str original_text: Optional[str] = None is_hallucination: bool = False hallucination_type: Optional[str] = None # 夸大/错误归因/无中生有 corrected_content: Optional[str] = None checked_at: datetime = Field(default_factory=datetime.now) class FinalReport(BaseModel): """最终研究报告""" session_id: str title: str executive_summary: str main_findings: List[ReportSection] = [] subtopic_reports: List[SubtopicReport] = [] overall_insights: List[KeyInsight] = [] recommendations: List[str] = [] methodology: Optional[str] = None limitations: List[str] = [] created_at: datetime = Field(default_factory=datetime.now) total_sources: int = 0 total_searches: int = 0 def to_markdown(self) -> str: """转换为完整的Markdown报告""" markdown = f"# {self.title}\n\n" markdown += f"*生成时间: {self.created_at.strftime('%Y-%m-%d %H:%M:%S')}*\n\n" # 执行摘要 markdown += "## 执行摘要\n\n" markdown += f"{self.executive_summary}\n\n" # 主要发现 if self.main_findings: markdown += "## 主要发现\n\n" for finding in self.main_findings: markdown += finding.to_markdown(level=3) # 整体洞察 if self.overall_insights: markdown += "## 综合洞察\n\n" for i, insight in enumerate(self.overall_insights, 1): markdown += f"### {i}. {insight.insight}\n\n" if insight.supporting_evidence: for evidence in insight.supporting_evidence: markdown += f"- {evidence}\n" markdown += "\n" # 建议 if self.recommendations: markdown += "## 建议\n\n" for recommendation in self.recommendations: markdown += f"- {recommendation}\n" markdown += "\n" # 详细子主题报告 markdown += "## 详细分析\n\n" for report in self.subtopic_reports: markdown += report.to_markdown() markdown += "---\n\n" # 研究方法 if self.methodology: markdown += "## 研究方法\n\n" markdown += f"{self.methodology}\n\n" # 局限性 if self.limitations: markdown += "## 研究局限性\n\n" for limitation in self.limitations: markdown += f"- {limitation}\n" markdown += "\n" # 统计信息 markdown += "## 研究统计\n\n" markdown += f"- 总搜索次数: {self.total_searches}\n" markdown += f"- 引用来源数: {self.total_sources}\n" markdown += f"- 分析子主题数: {len(self.subtopic_reports)}\n" return markdown def save_to_file(self, filepath: str): """保存为Markdown文件""" with open(filepath, 'w', encoding='utf-8') as f: f.write(self.to_markdown())