Smart Scholar Agent 是一个以 「精读」研究行为为核心的科研论文解析与理解型智能体。它的目标不是替代研究者阅读论文，而是：作为一个“研究型协作者”，帮助用户更系统、更审慎地理解一篇论文的研究问题、方法假设、证据充分性与结论可靠性。假如用户已经选定一篇论文，并决定投入 30–90 分钟进行精读，希望在有限时间内形成可靠的研究判断。

核心状态机（v0.1）

START
 ├─► parser_node
 ├─► cleaner_node        # 图文对齐 + 结构化
 ├─► deep_analyzer_node  # 核心：研究理解
 ├─► evidence_node       # 证据对齐与可疑点
 ├─► meta_analysis_node  # 可信度 & 冲突
 └─► END

将 LangGraph（大脑） 与 MCP（如 CV眼睛） 结合，是 AI Agent 从“只会聊天”进化到“能感知物理/数字世界”的关键一步。在这个架构中，大脑负责逻辑推理和任务拆解，而 MCP 提供的视觉工具负责精确的感知。构建一个**“多模态论文解析闭环”**：当 LLM 发现论文中的公式或图表模糊不清时，它会主动调用 CV 插件进行精准识别。

整体架构设计：解耦的大脑与感官

在工程上，采用多进程架构。大脑运行在 LangGraph 主进程中，眼睛（CV 模型），嘴巴（TTS模型）作为独立的 MCP Server 运行。

1. LangGraph (Brain): 管理 State，决定何时“看图”。
2. MCP Client: 嵌入在 LangGraph 的工具节点（Tool Node）中，负责与 Server 通信。
3. CV MCP Server (Eyes): 运行 YOLO/OCR 模型，接收图片路径，返回结构化坐标或文本。

---

核心步骤实现

1. 定义视觉状态 (The Visual State)

需要在 LangGraph 的状态中增加一个字段，记录当前正在分析的图片或页面区域。

class ResearchState(TypedDict):
    messages: Annotated[list, add_messages]
    current_image_path: str  # 当前正在分析的论文截图
    vision_results: dict      # 存储 CV 工具返回的坐标或 OCR 结果
    analysis_report: str     # 最终生成的解读

2. 封装 MCP 客户端为 LangGraph 工具

这是连接两者的桥梁。让 Agent 能够通过 MCP 协议向外部 Server 发送请求。

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from langchain_core.tools import tool

# 定义 MCP Server 的启动参数
cv_server_params = StdioServerParameters(
    command="python",
    args=["cv_mcp.py"] # CV Server 叫 cv_mcp.py
)

@tool
async def visual_analyze_tool(image_path: str, task_type: str):
    """
    调用专业的 CV MCP 工具。 task_type 可以是 'ocr' (识别公式) 或 'detect' (分析图表结构)。
    """
    async with stdio_client(cv_server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            
            # 动态调用 MCP Server 上的工具
            if task_type == "ocr":
                result = await session.call_tool("extract_latex_formula", {"path": image_path})
            else:
                result = await session.call_tool("detect_chart_elements", {"path": image_path})
                
            return result.content[0].text

3. 编排工作流 (The Workflow)

在 LangGraph 中，设计了一个循环：如果 LLM 觉得信息不足，就去调用“眼睛”。

from langgraph.graph import StateGraph, END

def brain_node(state: ResearchState):
    # LLM 根据当前消息决定是直接回答，还是调用视觉工具，这里绑定刚才写的 visual_analyze_tool
    llm_with_vision = llm.bind_tools([visual_analyze_tool])
    response = llm_with_vision.invoke(state["messages"])
    return {"messages": [response]}

workflow = StateGraph(ResearchState)
workflow.add_node("brain", brain_node)
workflow.add_node("vision_eyes", ToolNode([visual_analyze_tool]))

workflow.set_entry_point("brain")

# 决策逻辑：如果 LLM 输出了 tool_calls，则跳到 vision_eyes
workflow.add_conditional_edges(
    "brain",
    should_continue, # 检查 tool_calls
    {"tools": "vision_eyes", "end": END}
)
workflow.add_edge("vision_eyes", "brain") # 看完之后，回脑子里分析

app = workflow.compile()

精度互补：VLM（如 GPT-4o）能理解图片的大意，但量化精度极差（比如它看不清坐标轴上的 0.001）。 CV MCP (DBNet/YOLO) 精度极高，能给出确定的像素坐标或 LaTeX 公式。结果：大脑负责定性，眼睛负责定量。

算力解耦：LangGraph 可以跑在普通的 API 环境下。CV MCP Server 可以跑在远程带 GPU 的服务器上。两者通过 Stdio/HTTP 通信， Agent 项目不再需要把几百 MB 的 PyTorch 模型打包进应用里。

调试闭环 (Observability)：通过 LangSmith，可以清晰地看到大脑在什么时候感到“困惑”，以及它调用视觉工具时传输的具体参数。这就像给 AI 系统装了监控探头。

宏观认知：AI Agent 的“中央神经系统”模型

最快理解 Agent 架构的方式是类比。可以将 Agent 看作一个闭环控制系统：

组件	理论定义 (Agent Pattern)	对应论文助手项目中的实践	类比
大脑 (Brain)	核心 LLM。负责决策、反思和推理。	GPT-4o 或 Claude 3.5。	预训练的主干网络 (Backbone)
规划 (Planning)	如何拆解复杂任务（思维链 CoT）。	将“读论文”拆解为“读摘要 -> 找公式 -> 看图表”。	多尺度特征融合 / RPN 网络
记忆 (Memory)	短期：对话上下文；长期：RAG 知识库。	Thread ID 存储的对话、已读过的论文片段。	Feature Map / 缓存帧 (Cache)
工具 (Tools)	与外界交互的能力。	MCP 视觉识别、PDF 搜索、代码执行。	检测头 (Head) / 执行器

论文研读智能体：端到端的工程构建路径

第一步：定义“状态张量” (State Definition)

这是项目的灵魂。所有节点必须知道现在进行到了哪一步。

class PaperState(TypedDict):
    # 基础信息
    messages: list          # 对话流
    paper_path: str         # 论文路径
    # 提取的特征
    current_key_points: dict # 已经提取出的核心方法、实验结果等
    # 控制变量
    remaining_tasks: list    # 待办事项列表（由 Planner 生成）
    quality_score: int       # 对当前回答的自评分 (0-10)

第二步：设计“前向传播” (The Graph Flow)

1. 节点：Planner —— “我看了一下这篇 YOLOv10 的论文，我打算先看它的 BackBone，再看它的 NMS-free 策略。”（更新 remaining_tasks）。
2. 节点：Researcher —— 调用 RAG 工具搜文本，调用 MCP 工具看图表。（更新 current_key_points）。
3. 节点：Grader (反思) —— “Researcher 找的信息不够精准，没看到具体 mAP 数据，重做！”（循环回到 Researcher）。
4. 节点：Summarizer —— “任务完成，输出 Markdown 报告。”

第三步：工程化最佳实践 (Best Practices)

• 状态隔离：不要把所有原始文本都塞进 messages，这会造成 Token 爆炸。把关键信息存在 current_key_points 这种结构化字典里（类似于 特征降维）。
• MCP 解耦：把 YOLO 模型单独作为一个 MCP Server 运行。这样 LangGraph 逻辑就是纯净的逻辑流，不涉及深度学习框架的复杂环境。
• 可观测性 (Observability)：必须接入 LangSmith。在开发 Agent 时，你不能只看最终输出，必须看它的“思考路径”（Trace），就像调试 CV 网络时看每个层的 Feature Map。

LangGraph Demo: 学习如何构建控制流。

RAG 代码: 学习如何让 Agent 拥有外部知识。

MCP 代码: 学习如何让 Agent 接入专业工具。

一、 生产级项目结构设计

这个结构采用了分层架构，并将 MCP 服务与核心逻辑解耦。

smart-scholar-agent/
├── README.md
├── requirements.txt
├── .env                    # 环境配置（API Keys, DB URLs）
├── pyproject.toml          # 项目元数据及依赖管理
│
├── src/
│   ├── main.py             # 入口文件：启动 API 或 CLI
│   │
│   ├── agents/             # LangGraph 核心逻辑
│   │   ├── __init__.py
│   │   ├── state.py        # 定义 TypedDict 状态
│   │   ├── graph.py        # 编排节点和边的连接逻辑
│   │   └── nodes/          # 具体的节点实现
│   │       ├── analyst.py  # 论文深度剖析节点
│   │       ├── researcher.py # 检索与对比节点
│   │       └── generator.py  # PPT/思维导图生成节点
│   │
│   ├── mcp_servers/        # 独立的 MCP 服务模块（可解耦部署）
│   │   ├── vision_service/ # 图像/架构图解析
│   │   ├── ocr_service/    # 高精度 OCR 
│   │   └── doc_parser/     # PDF 结构化解析 (Markdown)
│   │
│   ├── core/               # 核心基础组件
│   │   ├── llm_factory.py  # 模型工厂（支持路由不同模型）
│   │   ├── mcp_client.py   # 通用 MCP 客户端封装
│   │   └── config.py       # 配置加载
│   │
│   ├── rag/                # RAG 知识库模块
│   │   ├── vector_store.py # 向量库连接与检索
│   │   ├── indexer.py      # 文档入库逻辑
│   │   └── embeddings.py   # 嵌入模型配置
│   │
│   ├── tools/              # 执行工具层（被 Agent 调用）
│   │   ├── ppt_gen.py      # PPT 生成工具
│   │   ├── map_gen.py      # 思维导图 (Mermaid/Markdown)
│   │   └── data_exporter.py# 数据对比表导出
│   │
│   └── schema/             # 数据协议与接口定义 (Pydantic models)
│       ├── paper.py        # 论文元数据模型
│       └── analysis.py     # 分析结果模型
│
├── tests/                  # 单元测试与集成测试
└── notebooks/              # 研发过程中的实验脚本

---

二、 核心设计模式应用

为了保证地基稳固，将引入以下设计模式：

1. 策略模式 (Strategy Pattern)：用于 LLM_Factory。根据任务复杂度（如简单提取 vs 深度推理）自动切换 GPT-4o, Claude 3.5 或 DeepSeek。
2. 工厂模式 (Factory Pattern)：用于 MCP_Client。无论底层调用的是视觉服务还是 OCR 服务，Client 层暴露统一的 call_tool 接口。
3. 状态模式 (State Pattern)：LangGraph 本身就是状态机模式的体现。将通过 State 字典管理论文处理的生命周期。
4. 外观模式 (Facade Pattern)：将复杂的 RAG 检索逻辑（多路召回、重排序、精排）封装在 rag/ 下，Agent 只需调用一个 search() 接口。

---

三、 核心接口协议设计 (Contract Design)

这是项目的“交通规则”。将使用 Pydantic 来严格定义接口。

1. 论文解析结果模型 (Schema)

这是 MCP 服务返回给中心大脑的标准格式。

from pydantic import BaseModel, Field
from typing import List, Optional

class PaperSection(BaseModel):
    title: str
    content: str
    figures: List[str] = [] # 图片 ID 或 URL

class PaperStructuredData(BaseModel):
    title: str
    authors: List[str]
    abstract: str
    sections: List[PaperSection]
    equations: List[str]  # 提取出的 LaTeX 公式
    references: List[str]

2. MCP 服务基础抽象 (Abstract Base Class)

确保所有垂类服务遵循统一的接口规范。

from abc import ABC, abstractmethod

class BaseMCPService(ABC):
    @abstractmethod
    async def process(self, input_data: bytes) -> dict:
        """所有服务必须实现此方法用于数据处理"""
        pass

    @abstractmethod
    def get_capabilities(self) -> List[str]:
        """返回该服务支持的能力（如 ['ocr', 'layout_analysis']）"""
        pass

3. LangGraph 状态定义 (State)

这是在不同节点间流转的“接力棒”。

from typing import TypedDict, Annotated, Sequence
from langchain_core.messages import BaseMessage

class AgentState(TypedDict):
    # 原始输入
    pdf_path: str
    # 结构化后的论文数据
    structured_paper: Optional[PaperStructuredData]
    # 已提取的关键点（动机、方法、创新点）
    analysis_results: dict
    # 对比论文列表
    comparison_papers: List[dict]
    # 消息队列（用于对话）
    messages: Annotated[Sequence[BaseMessage], "Add messages"]
    # 当前任务阶段
    current_phase: str

• 在 LangGraph 中，骨架的核心是 State (状态定义) 和 Graph (拓扑结构)。状态决定了信息如何在不同节点间流转，而拓扑结构决定了逻辑的闭环。

1. 定义状态 (src/agents/state.py)

状态是 LangGraph 的核心，它是一个在节点间传递的“内存数据库”。需要它既能承载原始消息流，又能承载高度结构化的论文分析结果。

from typing import Annotated, List, TypedDict, Dict, Any, Optional
from langgraph.graph.message import add_messages
from pydantic import BaseModel, Field

# 定义论文深度分析的结构化模型
class PaperAnalysis(BaseModel):
    motivation: str = Field(description="研究动机")
    problem_statement: str = Field(description="解决的核心问题")
    methodology: str = Field(description="核心方法论")
    innovations: List[str] = Field(description="创新点列表")
    limitations: List[str] = Field(description="局限性分析")

# 定义核心状态
class AgentState(TypedDict):
    # 消息列表，支持 add_messages 增量更新（用于对话历史）
    messages: Annotated[List[Any], add_messages]
    # 论文基本元数据
    paper_metadata: Dict[str, Any]
    # 结构化解析后的全文/片段 (由 MCP 提供)
    structured_content: Optional[Dict[str, Any]]    
    # 核心剖析结果
    analysis: Optional[PaperAnalysis]  
    # 对比分析数据（同类论文对比）
    comparison_data: List[Dict[str, Any]]
    # 当前处理任务的阶段标记
    current_step: str
    # 待生成的产物路径 (如 PPT, PDF 思维导图)
    artifacts: Dict[str, str]

---

2. 节点逻辑打样 (src/agents/nodes/)

为了让骨架跑起来，先创建几个“占位符节点”。每个节点负责状态中的一部分。

A. 解析节点 (src/agents/nodes/parser.py)

负责调用 MCP 服务，将 PDF 变成机器可读的结构。

from mcp import StdioServerParameters
from src.core.mcp_client import MCPClientManager
from src.agents.state import AgentState

# 配置 MCP Server 参数 (假设通过 python 启动一个解析脚本)
PARSER_SERVER_PARAMS = StdioServerParameters(
    command="python",
    args=["src/mcp_servers/doc_parser/server.py"], # 对应工程结构中的路径
    env=None
)

async def paper_parser_node(state: AgentState):
    print(f"--- 正在调用 MCP 服务解析论文: {state['pdf_path']} ---")
    
    try:
        async with MCPClientManager(PARSER_SERVER_PARAMS) as client:
            # 调用名为 'parse_pdf' 的 MCP 工具
            response = await client.call_service_tool(
                "parse_pdf", 
                {"path": state["pdf_path"]}
            )
            
            # 假设返回的是结构化的 Markdown 和元数据
            structured_data = response[0].text # 假设 MCP 返回 text 类型的 content
            
            return {
                "structured_content": {"raw_md": structured_data},
                "current_step": "analysis",
                "messages": [f"成功解析论文，获取到 {len(structured_data)} 字符数据。"]
            }
    except Exception as e:
        print(f"解析出错: {e}")
        return {"messages": [f"解析失败: {str(e)}"]}

B. 剖析节点 (src/agents/nodes/analyzer.py)

负责深入理解内容，提取核心逻辑。使用 high_precision 模型（如 Claude 3.5 Sonnet 或 GPT-4o），因为这里涉及极其复杂的逻辑推理。

from src.agents.state import AgentState
from src.schema.analysis import DetailedAnalysis
from src.core.llm_factory import LLMFactory
from src.agents.prompts.analyzer_prompts import ANALYZER_SYSTEM_PROMPT
from langchain_core.prompts import ChatPromptTemplate

async def research_analyzer_node(state: AgentState):
    # 1. 获取清洗后的结构化数据
    paper_data = state.get("structured_content")
    if not paper_data:
        return {"messages": ["错误：缺少结构化论文数据"]}

    # 2. 构造 Prompt
    prompt = ChatPromptTemplate.from_messages([
        ("system", ANALYZER_SYSTEM_PROMPT),
        ("user", "以下是论文的结构化内容：\n\n{content}")
    ])

    # 3. 调用高精度模型
    llm = LLMFactory.get_model(model_type="high_precision")
    structured_llm = llm.with_structured_output(DetailedAnalysis)
    
    try:
        # 执行推理
        analysis_result = await structured_llm.ainvoke(
            prompt.format(content=str(paper_data))
        )
        return {
            "analysis": analysis_result.dict(),
            "current_step": "output_generation", # 下一步准备生成产物
            "messages": ["论文深度分析已完成，思维导图大纲已生成。"]
        }
    except Exception as e:
        print(f"分析节点故障: {e}")
        return {"messages": [f"分析失败: {str(e)}"]}

---

3. 编排工作流 (src/agents/graph.py)

这是最关键的一步，将节点连接成网。设计一个线性+条件判断的流程：

from langgraph.graph import StateGraph, END
from src.agents.state import AgentState
from src.agents.nodes.parser import paper_parser_node
from src.agents.nodes.cleaner import paper_cleaner_node 
from src.agents.nodes.analyzer import research_analyzer_node
from src.agents.nodes.vision_processor import vision_processor_node # 新增

def create_research_graph():
    # 初始化状态图
    workflow = StateGraph(AgentState)

    # 添加节点
    workflow.add_node("parser", paper_parser_node)
    workflow.add_node("vision", vision_processor_node) # 新增
    workflow.add_node("cleaner", paper_cleaner_node) # 插入清洗节点
    workflow.add_node("analyzer", research_analyzer_node)
    # workflow.add_node("comparer", comparison_node)
    # workflow.add_node("generator", output_generator_node)

    # 构建边（确定逻辑流转）
    workflow.set_entry_point("parser")
    
    # 解析完成后，将图片交给视觉节点
    workflow.add_edge("parser", "vision")
    # 视觉分析完成后，交给清洗节点进行图文整合
    workflow.add_edge("vision", "cleaner") # Parser -> Cleaner
    workflow.add_edge("cleaner", "analyzer") # Cleaner -> Analyzer
    
    # 这里可以添加条件分支 (Conditional Edges) 例如：如果分析发现数据不足，可以跳回到 RAG 检索节点
    workflow.add_edge("analyzer", END)
    # 编译成可执行对象
    return workflow.compile()

# 实例化
app = create_research_graph()

# 在 main.py 中启动时：
# initial_state = {"pdf_path": "path/to/paper.pdf", "messages": []}
# app = create_research_graph()
# await app.ainvoke(initial_state)

MCP (Model Context Protocol) 的核心价值在于：它让“中心大脑”不再需要关心不同工具的具体实现逻辑，只需要通过标准协议调用接口。

4. 构建通用 MCP 客户端 (src/core/mcp_client.py)

设计一个 MCPClientManager，它支持连接到不同的 MCP Server（如 OCR 服务、视觉服务）。

import asyncio
from typing import Optional, Dict, Any
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

class MCPClientManager:
    """
    通用 MCP 客户端管理器 负责与不同的跨进程/跨网络服务（OCR, Vision, etc.）建立连接并调用工具
    """
    def __init__(self, server_params: StdioServerParameters):
        self.server_params = server_params
        self.session: Optional[ClientSession] = None
        self._exit_stack = None

    async def __aenter__(self):
        """支持 context manager 自动管理连接"""
        self._read, self._write = await stdio_client(self.server_params)
        self.session = ClientSession(self._read, self._write)
        await self.session.initialize()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if self.session:
            await self.session.__aexit__(exc_type, exc_val, exc_tb)

    async def call_service_tool(self, tool_name: str, arguments: Dict[str, Any]) -> Any:
        """调用 MCP Server 暴露的具体工具"""
        if not self.session:
            raise RuntimeError("MCP Session 未初始化，请先使用 'async with' 建立连接。")
        
        # 调用工具并返回结果
        try:
            result = await self.session.call_tool(tool_name, arguments)
            if hasattr(result, 'isError') and result.isError:
                raise RuntimeError(f"MCP 工具执行内部错误: {result.content}")
            return result.content
        except Exception as e:
            print(f"[Client Error] 调用工具 {tool_name} 失败: {e}")
            raise

---

5. 编写 MCP Server 逻辑 (src/mcp_servers/doc_parser/server.py)

使用 MCP SDK 的 FastMCP（或标准 Server 接口）来快速暴露工具。为了严谨起见，采用标准接口，这样更利于后期扩展复杂逻辑。

import os
import asyncio
from mcp.server import Server, NotificationOptions
from mcp.server.models import InitializationOptions
import mcp.types as types
from docling.datamodel.base_models import InputFormat
from docling.document_converter import DocumentConverter

# 1. 初始化 MCP Server
server = Server("paper-doc-parser")

# 2. 初始化 Docling 转换器（放在全局避免重复加载模型）
converter = DocumentConverter()

@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
    """列出该服务提供的工具"""
    return [
        types.Tool(
            name="parse_pdf",
            description="将 PDF 论文解析为结构化的 Markdown 文本，保留表格和布局信息",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "PDF 文件的本地绝对路径"},
                },
                "required": ["path"],
            },
        )
    ]

@server.call_tool()
async def handle_call_tool(
    name: str, arguments: dict | None
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
    """处理工具调用请求"""
    if name != "parse_pdf":
        raise ValueError(f"Unknown tool: {name}")

    if not arguments or "path" not in arguments:
        return [types.TextContent(type="text", text="错误：未提供文件路径")]

    pdf_path = arguments["path"]
    
    if not os.path.exists(pdf_path):
        return [types.TextContent(type="text", text=f"错误：文件路径不存在 {pdf_path}")]

    try:
        # 执行 Docling 转换逻辑 注意：Docling 是 CPU/GPU 密集型，这里使用 run_in_executor 防止阻塞异步循环
        loop = asyncio.get_event_loop()
        result = await loop.run_in_executor(None, lambda: converter.convert(pdf_path))
        
        # 导出为 Markdown
        markdown_output = result.document.export_to_markdown()
        
        return [
            types.TextContent(
                type="text",
                text=markdown_output
            )
        ]
    except Exception as e:
        return [types.TextContent(type="text", text=f"解析过程中出现故障: {str(e)}")]

async def main():
    # 启动 Stdio 传输服务，用于与中心大脑通信
    from mcp.server.stdio import stdio_server
    async with stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            InitializationOptions(
                server_name="paper-doc-parser",
                server_version="0.1.0",
                capabilities=server.get_capabilities(
                    notification_options=NotificationOptions(),
                    experimental_capabilities={},
                ),
            ),
        )

if __name__ == "__main__":
    asyncio.run(main())

---

6. 阶段性验证：“冒烟测试”

阶段性验证这个 MCP 服务是否真的能工作。

验证步骤：

1. 手动测试脚本：编写一个临时的 test_parser.py。
2. 模拟调用：利用上文写的 src/core/mcp_client.py 启动这个 Server 并发送一个路径过去。

可以运行以下逻辑（伪代码）来验证：

# test_parser_integration.py
async def test():
    params = StdioServerParameters(command="python", args=["src/mcp_servers/doc_parser/server.py"])
    async with MCPClientManager(params) as client:
        result = await client.call_service_tool("parse_pdf", {"path": "/路径/test_paper.pdf"})
        print(f"解析成功！前100个字符为: {result[0].text[:100]}")

现在已经完成了：

• 骨架 (LangGraph)：知道数据怎么流转。
• 通信协议 (MCP Client)：指挥部能下达命令。
• 实战部队 (DocParser Server)：第一支特种部队就位，且拥有 Docling 这种高精度解析能力。

Docling 提取的 Markdown 虽然保留了布局，但对于 LLM 来说，直接处理几十页的原始文本仍然存在“注意力弥散”和“Token 浪费”的问题。需要一个 Cleaner Node，利用大模型的理解能力，像手术刀一样精准地把这团“乱麻”切分并封装进预定义的 PaperStructuredData 模型中。

7. 完善数据协议 (src/schema/paper.py)

首先，确保“地基”里的图纸是详细的。需要定义解析后的标准模型。

from pydantic import BaseModel, Field
from typing import List, Optional

class PaperSection(BaseModel):
    title: str = Field(description="章节标题，如 'Introduction', 'Methodology'")
    content: str = Field(description="该章节的精简核心内容")

class PaperStructuredData(BaseModel):
    title: str = Field(description="论文标题")
    authors: List[str] = Field(description="作者列表")
    abstract: str = Field(description="摘要内容")
    key_terms: List[str] = Field(description="关键词/术语")
    sections: List[PaperSection] = Field(description="主要章节列表")
    raw_markdown: Optional[str] = Field(None, description="原始Markdown全文")

---

8. 构建 LLM 调用工厂 (src/core/llm_factory.py)

为了严谨，需要一个统一的接口来获取模型实例。使用 LangChain 的 with_structured_output 功能，它可以确保模型返回的数据严格符合 Pydantic 模型。

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
import os

class LLMFactory:
    @staticmethod
    def get_model(model_type: str = "fast"):
        # 实际项目中，配置信息应从 .env 读取
        if model_type == "high_precision":
            # 深度剖析使用 Claude 3.5 或 GPT-4o
            return ChatOpenAI(model="gpt-4o", temperature=0)
        else:
            # 数据清洗使用响应快、成本低的模型
            return ChatOpenAI(model="gpt-4o-mini", temperature=0)

---

9. 实现 Cleaner Node (src/agents/nodes/cleaner.py)

这是核心逻辑所在。该节点会从 AgentState 中取出 raw_md，调用 LLM 进行结构化转换。

from src.agents.state import AgentState
from src.schema.paper import PaperStructuredData
from src.core.llm_factory import LLMFactory
from langchain_core.prompts import ChatPromptTemplate
from src.agents.prompts.cleaner_prompts import FUSION_CLEANER_SYSTEM_PROMPT

# 定义清洗指令
CLEANER_PROMPT = ChatPromptTemplate.from_messages([
    ("system", "你是一个资深论文审稿专家。请将以下从 PDF 中解析出的原始 Markdown 文本进行结构化处理。提取论文的标题、作者、摘要、关键词以及各个核心章节的内容。"),
    ("user", "原始 Markdown 内容如下：\n\n{markdown_content}")
])

async def paper_cleaner_node_old(state: AgentState):
    print("--- 执行：数据清洗与结构化打标 ---")
    
    raw_md = state.get("structured_content", {}).get("raw_md", "")
    if not raw_md:
        return {"messages": ["错误：未获取到原始解析内容"]}

    # 初始化模型并绑定结构化输出
    llm = LLMFactory.get_model(model_type="fast")
    structured_llm = llm.with_structured_output(PaperStructuredData)

    # 执行清洗任务 注意：如果 Markdown 超长，这里需要做简单的分段处理或仅提取关键部分（如前 8k tokens）
    try:
        cleaned_data = await structured_llm.ainvoke(
            CLEANER_PROMPT.format(markdown_content=raw_md[:20000]) # 示例截断，生产环境需优化
        )
        
        # 更新状态
        return {
            "structured_content": cleaned_data.dict(), # 转化为结构化字典
            "current_step": "analysis",
            "messages": [f"已完成结构化提取：{cleaned_data.title}"]
        }
    except Exception as e:
        print(f"清洗节点故障: {e}")
        return {"messages": [f"清洗失败: {str(e)}"]}

async def paper_cleaner_node(state: AgentState):
    print("--- 执行：图文融合深度清洗 ---")
    
    # 提取原始 Markdown 和 视觉分析结果
    raw_md = state.get("structured_content", {}).get("raw_md", "")
    vision_results = state.get("vision_analysis_results", [])
    
    if not raw_md:
        return {"messages": ["错误：未获取到原始 Markdown 内容"]}

    # 格式化视觉分析报告，方便 LLM 阅读
    vision_report = "\n".join([
        f"图片路径/ID: {res['image_id']}\n标题上下文: {res['caption']}\n视觉解读内容: {res['analysis']}\n---"
        for res in vision_results
    ])

    # 构造融合 Prompt
    prompt = ChatPromptTemplate.from_messages([
        ("system", FUSION_CLEANER_SYSTEM_PROMPT),
        ("user", "【原始 Markdown】\n{markdown}\n\n【视觉分析报告】\n{vision_report}")
    ])

    # 调用模型进行融合提取 使用高精度模型以确保复杂的图文对应关系不乱
    llm = LLMFactory.get_model(model_type="high_precision")
    structured_llm = llm.with_structured_output(PaperStructuredData)

    try:
        fused_data = await structured_llm.ainvoke({
            "markdown": raw_md[:25000], # 考虑上下文长度限制
            "vision_report": vision_report
        })

        # 更新状态
        return {
            "structured_content": fused_data.dict(),
            "current_step": "analysis",
            "messages": [f"图文融合清洗完成。整合了 {len(vision_results)} 张插图的分析。"]
        }
    except Exception as e:
        print(f"融合清洗失败: {e}")
        return {"messages": [f"融合清洗失败: {str(e)}"]}

---

在构建 Cleaner Node 时，必须面对一个残酷的现实：长论文的 Token 溢出。 Docling 提取出来的文本可能包含 50,000 个单词，这会超出很多模型的单次输入限制，或者导致处理极慢。

• 分块清洗 (Chunked Cleaning)：如果 raw_md 超过限制，先通过正则匹配（如 # 标题）切分章节，每个章节独立清洗，最后合并。
• 两阶段提取：第一阶段：提取元数据和摘要。第二阶段：根据用户关心的“创新点”或“方法论”，按需从原始文本中检索相关段落（即 Small-to-Big 检索策略）。

在实现 analyzer 节点时，引入思维链（Chain of Thought, CoT）和多角色演化的 Prompt 技术。不仅要提取信息，还要让 LLM 扮演“刁钻的审稿人”来评估局限性，扮演“导师”来梳理思维导图。

10. 细化分析结果模型 (src/schema/analysis.py)

为了支持思维导图和深度剖析，需要更细致的数据结构。

from pydantic import BaseModel, Field
from typing import List, Dict

class MindMapNode(BaseModel):
    id: str
    label: str
    children: List['MindMapNode'] = []

class DetailedAnalysis(BaseModel):
    motivation: str = Field(description="研究背景与为什么要做这项研究")
    core_problem: str = Field(description="论文试图解决的核心痛点")
    key_innovations: List[str] = Field(description="创新点，需注明是算法创新、架构创新还是应用创新")
    methodology_breakdown: Dict[str, str] = Field(description="核心方法的步骤拆解")
    technical_limitations: List[str] = Field(description="技术或实验上的局限性")
    future_work: List[str] = Field(description="作者提到的或你洞察到的后续方向")
    mind_map_mermaid: str = Field(description="用于生成思维导图的 Mermaid 格式代码")

---

11. 设计深度剖析 Prompt (src/agents/prompts/analyzer_prompts.py)

需要引导 LLM 进入“深度思考”模式。

提示词策略： 采用“逻辑溯源法”。要求 LLM 先寻找论文中提到的现有技术的不足，再对应到本文的改进策略。

ANALYZER_SYSTEM_PROMPT = """你现在是一名世界顶级的人工智能实验室首席研究员。
你的任务是基于提供的结构化论文内容，进行透彻的分析。

请遵循以下逻辑进行思考：
1. **溯源**：作者提到了哪些前人工作的痛点？
2. **对症下药**：本文提出的方法是如何精准解决这些痛点的？
3. **推演**：该方法的引入会带来哪些副作用（局限性）？
4. **可视化**：将整篇论文的逻辑结构转化为一个逻辑清晰的 Mermaid 思维导图。

请确保分析严谨、学术，不要使用笼统的词汇。"""

---

12. 思维导图的可视化逻辑

在这一步，生成的 mind_map_mermaid 字段将类似于：

mindmap
  root((论文标题))
    研究动机
      痛点A
      痛点B
    核心方法
      模块1
      模块2
    创新点
      算法优化

这种格式可以直接被前端库（如 Mermaid.js）渲染，也可以被后续的 PPT 生成工具解析。

为了确保 analyzer 的结果可靠，避免 LLM “一本正经地胡说八道”，需要在地基中加入**“证据回溯”机制**：引文锚点：在 Prompt 中要求 LLM 在描述创新点时，必须指明是在 structured_content 的哪个章节发现的。双向校验：可以增加一个“批判性节点”，让它专门寻找分析结果中的逻辑漏洞。到这一步，智能体已经具备了**“读懂”并“拆解”**论文的能力。它不再只是一个搬运工，而是一个能够输出深度见解的研究助手。

视觉服务的设计思路

视觉信息在论文中往往承载了最核心的对比结果（如消融实验柱状图）和方法架构（如模型流程图）。需要构建一个**“视觉特种兵” (Vision MCP Service)，专门负责把这些像素转化为语义信息，补全数据拼图中最关键的一块。为了追求“高精度垂类能力”，这个 Vision MCP 不能只是简单地调用一个通用看图模型。需要对它进行专业化定制**：

1. 图像分类路由：首先判断图片是“数据图表 (Chart/Plot)”还是“架构示意图 (Diagram/Architecture)”。
2. 定向 Prompt 引导：如果是图表，关注坐标轴含义、对比项、关键趋势和消融结果。如果是架构图，关注数据流向、模块交互和核心创新模块。

---

13. 实战：构建视觉 MCP 服务 (src/mcp_servers/vision_service/)

首先安装必要依赖。假设后端接入一个强大的多模态大模型 API（如 OpenAI GPT-4o 或开源的 Qwen-VL-Chat 部署版）。为了演示通用性，用 API 方式实现，可以在实际部署时替换为本地模型调用。

pip install mcp openai pillow

编写服务端代码 src/mcp_servers/vision_service/server.py：

import os
import base64
import asyncio
from mcp.server import Server, NotificationOptions
from mcp.server.models import InitializationOptions
import mcp.types as types
from openai import AsyncOpenAI

# 初始化服务与客户端
server = Server("paper-vision-service")
aclient = AsyncOpenAI()# 注意：实际使用需要配置环境变量 OPENAI_API_KEY，或者替换为本地模型客户端

# 定义针对学术图表的 System Prompt
VISION_SYSTEM_PROMPT = """你是一名资深的学术论文审稿专家，擅长深入分析论文插图。
你的任务是精准解读提供的图像，不要进行泛泛而谈的描述。

- 如果是**数据图表**（折线图、柱状图、表格）：
  1. 明确指出坐标轴、图例的含义。
  2. 详细对比不同方法（特别是 Baseline 和 Ours）的性能差异。
  3. 总结该图表支撑了论文的什么核心论点（例如："证明模块A的有效性"）。

- 如果是**架构图或流程图**：
  1. 描述整体数据流向。
  2. 识别核心模块及其交互方式。
  3. 指出图中强调的创新部分。

请输出清晰、结构化的分析文本。"""

def encode_image(image_path):
    """将本地图片转换为 base64 编码"""
    with open(image_path, "rb") as image_file:
        return base64.b64encode(image_file.read()).decode('utf-8')

@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
    return [
        types.Tool(
            name="analyze_paper_image",
            description="深度分析论文中的插图，识别图表数据趋势或架构图逻辑。",
            inputSchema={
                "type": "object",
                "properties": {
                    "image_path": {"type": "string", "description": "图片的本地绝对路径"},
                    "context_hint": {"type": "string", "description": "可选的上下文提示（如图片标题）"},
                },
                "required": ["image_path"],
            },
        )
    ]

@server.call_tool()
async def handle_call_tool(
    name: str, arguments: dict | None
) -> list[types.TextContent]:
    if name != "analyze_paper_image":
        raise ValueError(f"Unknown tool: {name}")

    image_path = arguments.get("image_path")
    context_hint = arguments.get("context_hint", "")

    if not os.path.exists(image_path):
        return [types.TextContent(type="text", text=f"错误：图片路径不存在 {image_path}")]

    try:
        # 准备 Base64 图片数据
        base64_image = encode_image(image_path)
        # 调用视觉模型进行分析
        response = await aclient.chat.completions.create(
            model="gpt-4o", # 或替换为你部署的高精度本地垂类模型名称
            messages=[
                {"role": "system", "content": VISION_SYSTEM_PROMPT},
                {"role": "user", "content": [
                    {"type": "text", "text": f"请分析这张图片。背景提示：{context_hint}"},
                    {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{base64_image}"}}
                ]}
            ],
            max_tokens=1000
        )
        analysis_result = response.choices[0].message.content
        
        return [types.TextContent(type="text", text=analysis_result)]

    except Exception as e:
        return [types.TextContent(type="text", text=f"视觉分析服务故障: {str(e)}")]

async def main():
    # 启动标准输入输出服务
    from mcp.server.stdio import stdio_server
    async with stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream, write_stream,
            InitializationOptions(
                server_name="paper-vision-service",
                server_version="0.1.0",
                capabilities=server.get_capabilities(notification_options=NotificationOptions(), experimental_capabilities={}),
            ),
        )

if __name__ == "__main__":
    # 实际部署时需确保 API Key 环境变量已设置
    os.environ["OPENAI_API_KEY"] = "key-here"
    asyncio.run(main())

---

14. 集成：在 LangGraph 中引入“视觉处理节点”

现在有了服务，需要在工作流中增加一个环节来调用它。最好的位置是在 Parser（解析出图片文件） 之后，Cleaner（清洗文本） 之前。

步骤 A：更新数据 Schema (src/schema/paper.py)

需要一个地方来存放对图片的分析结果。

from pydantic import BaseModel, Field
from typing import List, Optional

# 新增：单个图像分析结果模型
class ImageAnalysis(BaseModel):
    image_id: str = Field(description="原始文件名或ID")
    caption: str = Field(description="图片标题或上下文")
    analysis: str = Field(description="视觉模型生成的深度解读")


class PaperSection(BaseModel):
    title: str = Field(description="章节标题，如 'Introduction', 'Methodology'")
    content: str = Field(description="该章节的精简核心内容")

class PaperStructuredData(BaseModel):
    title: str = Field(description="论文标题")
    authors: List[str] = Field(description="作者列表")
    abstract: str = Field(description="摘要内容")
    key_terms: List[str] = Field(description="关键词/术语")
    sections: List[PaperSection] = Field(description="主要章节列表")
    raw_markdown: Optional[str] = Field(None, description="原始Markdown全文")
    # 新增字段来存储图像分析列表
    image_analyses: List[ImageAnalysis] = Field(default_factory=list, description="论文插图的深度分析集合")

步骤 B：创建 Vision Processor Node (src/agents/nodes/vision_processor.py)

这个节点负责遍历解析出来的所有图片，并发调用 Vision MCP 进行分析。

import asyncio
from mcp import StdioServerParameters
from src.core.mcp_client import MCPClientManager
from src.agents.state import AgentState
from src.schema.paper import ImageAnalysis

# 配置 Vision MCP Server (假设通过 python 启动)
VISION_SERVER_PARAMS = StdioServerParameters(
    command="python",
    args=["src/mcp_servers/vision_service/server.py"],
    env=None # 确保在此处或全局注入必要的环境变量
)

async def vision_processor_node(state: AgentState):
    print("--- 执行：论文插图深度视觉分析 ---")
    
    # 假设 parser 节点在 state 中存了一个 extracted_images 列表
    # 格式如: [{"path": "/tmp/img1.png", "caption": "Fig 1. Architecture"}]
    extracted_images = state.get("paper_metadata", {}).get("extracted_images", [])
    
    if not extracted_images:
        print("未发现需要分析的图片，跳过视觉处理。")
        return {"messages": ["无图片需要分析，跳过视觉节点。"]}

    processed_images = []
    
    try:
        async with MCPClientManager(VISION_SERVER_PARAMS) as client:
            # 创建并发任务列表
            tasks = []
            for img_info in extracted_images:
                task = client.call_service_tool(
                    "analyze_paper_image",
                    {
                        "image_path": img_info["path"],
                        "context_hint": img_info.get("caption", "")
                    }
                )
                tasks.append(task)
            
            # 并发执行所有图片的分析任务
            print(f"开始并发分析 {len(tasks)} 张图片...")
            results = await asyncio.gather(*tasks, return_exceptions=True)
            
            for i, result in enumerate(results):
                img_info = extracted_images[i]
                if isinstance(result, Exception):
                    print(f"图片 {img_info['path']} 分析失败: {result}")
                else:
                    # 假设 MCP 返回的是 text content
                    analysis_text = result[0].text
                    processed_images.append(ImageAnalysis(
                        image_id=img_info["path"],
                        caption=img_info.get("caption", ""),
                        analysis=analysis_text
                    ))
                    print(f"图片 {img_info['path']} 分析完成。")

        # 将分析结果存入临时状态，供后续 Cleaner 节点整合
        return {
            "vision_analysis_results": [img.dict() for img in processed_images],
            "messages": [f"成功完成 {len(processed_images)} 张核心图片的视觉深度分析。"]
        }
        
    except Exception as e:
        print(f"视觉处理节点整体故障: {e}")
        return {"messages": [f"视觉分析失败: {str(e)}"]}

现在处理链条更加完整了：解析 -> 视觉分析 -> 文本清洗 -> 深度剖析。引入了新的 MCP 服务，还利用了 Python asyncio 的特性实现了多图并发分析，这对于处理包含十几张图的论文来说效率提升巨大。“图文融合” 是迈向多模态理解的最后一公里。如果 analyzer 只看文字，它可能会错过实验图表中的关键证据；如果只看图片分析，它又缺乏理论背景。在 cleaner 节点进行融合，本质上是将视觉模型的“感官结果”嵌入到文本模型的“逻辑流”中。

15. 完善数据协议 (src/schema/paper.py)

为了支撑融合后的数据，要确保 PaperSection 能够承载插图的分析摘要。

from pydantic import BaseModel, Field
from typing import List, Optional, Dict


# 新增：单个图像分析结果模型
class ImageAnalysis(BaseModel):
    image_id: str = Field(description="原始文件名或ID")
    caption: str = Field(description="图片标题或上下文")
    analysis: str = Field(description="视觉模型生成的深度解读")

class PaperSection(BaseModel):
    title: str = Field(description="章节标题，如 'Introduction', 'Methodology'")
    content: str = Field(description="该章节的核心内容，应包含对本章节内相关图表的引用描述")
    related_image_ids: List[str] = Field(default_factory=list, description="本章节涉及到的图片ID/路径")   

class PaperStructuredData(BaseModel):
    title: str = Field(description="论文标题")
    authors: List[str] = Field(description="作者列表")
    abstract: str = Field(description="摘要内容")
    key_terms: List[str] = Field(description="关键词/术语")
    sections: List[PaperSection] = Field(description="主要章节列表")
    raw_markdown: Optional[str] = Field(None, description="原始Markdown全文")
    # 存储全量的视觉分析，供全局索引
    image_metadata: Dict[str, str] = Field(default_factory=dict, description="ID到视觉分析描述的映射")
    # 新增字段来存储图像分析列表
    image_analyses: List[ImageAnalysis] = Field(default_factory=list, description="论文插图的深度分析集合")

---

16. 设计“图文融合”清洗指令 (src/agents/prompts/cleaner_prompts.py)

告诉 LLM：你现在不仅是在整理 Markdown，你还是一个装配工，负责把视觉报告安装到文字的对应位置。

FUSION_CLEANER_SYSTEM_PROMPT = """你是一个专业的论文数据整合专家。
你的任务是将原始的论文 Markdown 文本与针对文中插图的“视觉分析报告”进行深度融合。

输入说明：
1. **原始 Markdown**：包含论文的文字内容。
2. **视觉分析报告**：包含了文中关键图片（如 Fig 1, Chart 等）的内容解读。

处理要求：
- 将视觉分析报告的内容，有机地整合进对应的论文章节中。
- 如果 Markdown 中提到“Figure X”，请在该段落之后插入对应的视觉分析摘要。
- 最终输出一个高度结构化的 JSON，确保文字逻辑与图表证据相辅相成。
- 保持学术严谨性，确保图文对应准确。"""

为了让这个“闭环”真正科学严谨，需要在实现中注意以下几点：

1. 图片 ID 锚点：在 parser 节点解析 PDF 时，Docling 会在 Markdown 里留下类似 的标记。在 vision_processor 中必须保留这些 ID，以便 cleaner 能精准回填。
2. 上下文长度管理：论文 + 视觉分析报告可能非常长。如果超过 LLM 窗口，需要采用**“滑动窗口清洗”或“先摘要后融合”**的策略。
3. 视觉缓存：在开发阶段，视觉模型调用很贵且慢。可以在 vision_processor 节点加入一个简单的本地缓存（基于图片 Hash），避免重复解析同一张图。

智能体可以完成了**“单篇论文深度数字化”**的全过程：

1. 物理拆解 (Parser)
2. 感官解读 (Vision)
3. 逻辑装配 (Cleaner/Fusion)
4. 智慧剖析 (Analyzer)

要实现多论文对比和数据对比表，RAG（检索增强生成）不仅是存储，更是一套知识索引架构。普通 RAG 只是把文本切碎存入数据库，需要构建的是结构化 RAG (Structured-RAG)。要利用之前在 cleaner 节点生成的 PaperStructuredData 进行精准索引。RAG 架构设计：不仅仅是向量。为了支持同类论文的“对比维度分析”，需要双层索引机制：

1. 向量索引 (Vector Index)：用于模糊语义检索（如：寻找使用类似“注意力机制”的论文）。
2. 属性索引 (Property Index)：基于结构化字段（如：年份、数据集、性能指标）进行过滤和精确对比。

17. 核心组件实现 (src/rag/)

A. 向量库抽象 (src/rag/vector_store.py)

可以选择 ChromaDB 作为地基，它们支持 Metadata 过滤。

import chromadb
from chromadb.utils import embedding_functions
from src.schema.paper import PaperStructuredData

class PaperVectorStore:
    def __init__(self, collection_name="research_papers"):
        self.client = chromadb.PersistentClient(path="./db/chroma")
        # 使用高维度的嵌入模型，如 OpenAI 或 HuggingFace
        self.emb_fn = embedding_functions.OpenAIEmbeddingFunction(
            api_key="key",
            model_name="text-embedding-3-large"
        )
        self.collection = self.client.get_or_create_collection(
            name=collection_name, 
            embedding_function=self.emb_fn
        )

    def add_paper(self, structured_data: PaperStructuredData, paper_id: str):
        """将结构化论文存入 RAG"""
        # 将全篇核心内容存入向量
        full_text = f"{structured_data.title} {structured_data.abstract}"
        for section in structured_data.sections:
            full_text += f"\n{section.title}: {section.content}"

        # 存储时附加元数据，方便后续筛选
        self.collection.add(
            ids=[paper_id],
            documents=[full_text],
            metadatas=[{
                "title": structured_data.title,
                "authors": ",".join(structured_data.authors),
                "key_terms": ",".join(structured_data.key_terms)
            }]
        )

    async def search_similar_papers(self, query: str, n_results: int = 5):
        """语义搜索"""
        return self.collection.query(query_texts=[query], n_results=n_results)

---

18. 实现对比节点 (src/agents/nodes/comparer.py)

这是实现“同类论文方法与结果对比”的核心。它会从库中捞出相似论文，然后让 LLM 生成对比表格。

from src.agents.state import AgentState
from src.core.llm_factory import LLMFactory
from langchain_core.prompts import ChatPromptTemplate
from pydantic import BaseModel, Field
from typing import List

class ComparisonTable(BaseModel):
    dimensions: List[str] = Field(description="对比的维度，如：参数量, 精度, 耗时")
    rows: List[dict] = Field(description="每一行代表一个模型/论文的数据项")

async def comparison_node(state: AgentState):
    print("--- 执行：多论文横向对比分析 ---")
    
    current_paper = state["structured_content"]
    # 检索相似论文
    # 实际开发中，这里会调用 PaperVectorStore 获取相关上下文
    similar_papers_context = "..." # Mock 检索到的上下文
    
    # 构造对比 Prompt
    prompt = ChatPromptTemplate.from_messages([
        ("system", "你是一个学术综述专家。请对比当前论文与已有研究，提取核心指标并生成对比表。"),
        ("user", "当前论文：{current}\n\n参考论文集：{references}")
    ])

    llm = LLMFactory.get_model(model_type="high_precision")
    structured_llm = llm.with_structured_output(ComparisonTable)
    
    try:
        table_result = await structured_llm.ainvoke({
            "current": str(current_paper),
            "references": similar_papers_context
        })
        
        return {
            "comparison_data": table_result.dict(),
            "messages": ["多论文横向对比分析表已生成。"]
        }
    except Exception as e:
        return {"messages": [f"对比失败: {str(e)}"]}

---

19. 关键：维度分析（Dimension Analysis）

为了让对比不只是流于表面，需要定义对比维度。在 Prompt 中，可以强制要求 LLM 寻找以下指标：

• $SOTA$ 对标：该方法在哪些指标上超过了之前的 $SOTA$？
• 计算开销：$FLOPs$、$Params$ 的对比。
• 数据效率：是否使用了更少的数据达到了相同的效果？
• 消融差异：其他论文验证过的有效模块，在本文中是否被弃用？
• …

---

20. 完善 LangGraph：多论文处理流

需要在骨架中加入一个“检索-对比”循环。

# src/agents/graph.py
from langgraph.graph import StateGraph, END
from src.agents.state import AgentState
from src.agents.nodes.parser import paper_parser_node
from src.agents.nodes.cleaner import paper_cleaner_node 
from src.agents.nodes.analyzer import research_analyzer_node
from src.agents.nodes.vision_processor import vision_processor_node # 新增


def create_research_graph():
    # 初始化状态图
    workflow = StateGraph(AgentState)

    # 添加节点
    workflow.add_node("parser", paper_parser_node)
    workflow.add_node("vision", vision_processor_node) # 新增
    workflow.add_node("cleaner", paper_cleaner_node) # 插入清洗节点
    workflow.add_node("analyzer", research_analyzer_node)
    workflow.add_node("comparer", comparison_node) # 新增对比节点
    # workflow.add_node("generator", output_generator_node)

    # 构建边（确定逻辑流转）
    workflow.set_entry_point("parser")
    
    # 解析完成后，将图片交给视觉节点
    workflow.add_edge("parser", "vision")
    # 视觉分析完成后，交给清洗节点进行图文整合
    workflow.add_edge("vision", "cleaner") # Parser -> Cleaner
    workflow.add_edge("cleaner", "analyzer") # Cleaner -> Analyzer
    # 分析完当前论文后，自动进入 RAG 进行库内对比
    workflow.add_edge("analyzer", "comparer")
    # 这里可以添加条件分支 (Conditional Edges) 例如：如果分析发现数据不足，可以跳回到 RAG 检索节点
    workflow.add_edge("comparer", END)

    # 4编译成可执行对象
    return workflow.compile()

# 实例化
app = create_research_graph()

# 在 main.py 中启动时：
# initial_state = {"pdf_path": "path/to/paper.pdf", "messages": []}
# app = create_research_graph()
# await app.ainvoke(initial_state)

RAG 性能优化注意要踩的坑

1. 分块冗余：论文中很多文字（如致谢、参考文献）会污染向量库，在 indexer.py 中要利用 structured_content 剔除这些噪声。
2. 多路召回：仅靠向量检索不够，需要 Keyword Search (BM25) 来匹配特定的算法术语（如 “LoRA”, “Transformer”）。
3. 引用闭环：如果当前论文引用了库里的某篇论文，应该赋予该参考论文更高的检索权重。

现在已经实现了从单点到多点的跨越。智能体现在能够：

• 读懂：利用视觉和文本融合。
• 记忆：将知识存入 RAG 向量库。
• 对比：横向拉通多篇论文生成对比表。

将之前在 analyzer（单篇深度）和 comparer（多篇横向）中积累的所有结构化智慧，转化为用户可以直接使用的实物资产。为了实现高质量的产出，将采用模板方法模式 (Template Method Pattern)。这意味着不是让 LLM 盲目生成代码，而是提供标准的学术 PPT 模板和思维导图骨架，让智能体进行“填空式”创作。

21. 产物生成器工具开发 (src/tools/)

将构建两个核心工具类：PPTGenerator 和 MindMapGenerator。

A. 学术对比 PPT 生成器 (src/tools/ppt_gen.py)

需要使用 python-pptx 库。为了美观，建议预设一个包含标题页、目录页、对比图表页和结论页的 template.pptx。

from pptx import Presentation
from pptx.util import Inches, Pt
from src.schema.analysis import DetailedAnalysis
from src.agents.nodes.comparer import ComparisonTable

class PPTGenerator:
    def __init__(self, template_path="assets/academic_template.pptx"):
        self.prs = Presentation(template_path) if template_path else Presentation()

    def add_title_slide(self, title, subtitle):
        slide = self.prs.slides.add_slide(self.prs.slide_layouts[0])
        slide.shapes.title.text = title
        slide.placeholders[1].text = subtitle

    def add_comparison_table(self, table_data: ComparisonTable):
        """将对比维度数据转化为 PPT 表格"""
        slide = self.prs.slides.add_slide(self.prs.slide_layouts[5]) # 标题+内容布局
        slide.shapes.title.text = "横向方法对比分析"
        
        # 定义表格：行数=数据行+1(表头), 列数=维度数+1(论文名)
        rows, cols = len(table_data.rows) + 1, len(table_data.dimensions) + 1
        left, top, width, height = Inches(0.5), Inches(1.5), Inches(9), Inches(5)
        table = slide.shapes.add_table(rows, cols, left, top, width, height).table
        
        # 设置表头
        table.cell(0, 0).text = "论文/指标"
        for j, dim in enumerate(table_data.dimensions):
            table.cell(0, j+1).text = dim
            
        # 填充数据 (此处省略具体循环逻辑)
        self.prs.save("output/research_comparison.pptx")
        return "output/research_comparison.pptx"

B. 思维导图生成器 (src/tools/map_gen.py)

利用 Mermaid.js 语法。它最大的好处是文本即图片，易于在 Web 前端渲染，也方便用户二次修改。

class MindMapGenerator:
    @staticmethod
    def generate_mermaid_code(analysis: DetailedAnalysis, comparison: ComparisonTable):
        """整合单篇深度与多篇对比的思维导图""" # 利用 Markdown 字符串拼接出 Mermaid 格式
        mermaid_template = f"""
mindmap
  root(({analysis.core_problem}))
    研究动机
      {analysis.motivation}
    核心创新
      {"".join([f"      {inn}" for inn in analysis.key_innovations])}
    横向对比
      {"".join([f"      {row['title']}: {row['key_diff']}" for row in comparison.rows])}
        """
        return mermaid_template

---

22. 构建产物生成节点 (src/agents/nodes/generator.py)

这个节点在 LangGraph 中扮演“总装车间”的角色。

from src.agents.state import AgentState
from src.tools.ppt_gen import PPTGenerator
from src.tools.map_gen import MindMapGenerator

async def output_generator_node(state: AgentState):
    print("--- 执行：自动化产物生成 (PPT & MindMap) ---")
    
    analysis = state.get("analysis")
    comparison = state.get("comparison_data")
    
    # 生成思维导图代码
    map_gen = MindMapGenerator()
    mermaid_code = map_gen.generate_mermaid_code(analysis, comparison)
    
    # 生成 PPT 文件
    ppt_gen = PPTGenerator()
    ppt_path = ppt_gen.add_comparison_table(comparison)
    
    return {
        "artifacts": {
            "mermaid_code": mermaid_code,
            "ppt_file_path": ppt_path
        },
        "messages": ["PPT 和思维导图已成功生成，存放在 output 目录。"]
    }

---

23. 更新工作流闭环 (src/agents/graph.py)

现在， LangGraph 已经从“解析”一直走到了“产出”。

from langgraph.graph import StateGraph, END
from src.agents.state import AgentState
from src.agents.nodes.parser import paper_parser_node
from src.agents.nodes.cleaner import paper_cleaner_node 
from src.agents.nodes.analyzer import research_analyzer_node
from src.agents.nodes.vision_processor import vision_processor_node # 新增

def create_research_graph():
    # 初始化状态图
    workflow = StateGraph(AgentState)

    # 添加节点
    workflow.add_node("parser", paper_parser_node)
    workflow.add_node("vision", vision_processor_node) # 新增
    workflow.add_node("cleaner", paper_cleaner_node) # 插入清洗节点
    workflow.add_node("analyzer", research_analyzer_node)
    workflow.add_node("comparer", comparison_node) # 新增对比节点
    workflow.add_node("generator", output_generator_node)

    # 构建边（确定逻辑流转）
    workflow.set_entry_point("parser")
    
    # 解析完成后，将图片交给视觉节点
    workflow.add_edge("parser", "vision")
    # 视觉分析完成后，交给清洗节点进行图文整合
    workflow.add_edge("vision", "cleaner") # Parser -> Cleaner
    workflow.add_edge("cleaner", "analyzer") # Cleaner -> Analyzer
    # 分析完当前论文后，自动进入 RAG 进行库内对比
    workflow.add_edge("analyzer", "comparer")
    # 这里可以添加条件分支 (Conditional Edges)
    # 例如：如果分析发现数据不足，可以跳回到 RAG 检索节点
    workflow.add_edge("comparer", "generator")
    workflow.add_edge("generator", END)

    # 编译成可执行对象
    return workflow.compile()

# 实例化
app = create_research_graph()

# 在 main.py 中启动时：
# initial_state = {"pdf_path": "path/to/paper.pdf", "messages": []}
# app = create_research_graph()
# await app.ainvoke(initial_state)

在生成 PPT 时可以加入一些高阶逻辑：

1. 自动配色方案：根据论文的主题（如：AI 类用深蓝色调，生物类用翠绿色调）自动选择 PPT 模板。
2. 数据可视化插图：如果 comparison_data 中包含数值指标（如 $mAP$ 或 $Accuracy$），可以调用 matplotlib 生成一张对比条形图，并作为图片插入 PPT 中。
3. 多维度分类标签：在 PPT 第一页自动打上“Transformer”、“CVPR 2024”、“SOTA” 等多维分类标签。

如果说 RAG 提供了“局部检索”的能力，那么**知识图谱（Knowledge Graph, KG）则为你的智能体提供了“全局视野”和“历史脉络”。通过构建学术图谱，智能体将能够回答类似于：“这篇文章所改进的 Baseline 在过去三年中是如何演化的？”或者“目前该领域内哪几种技术路径是竞争关系？”这类高阶问题。要实现的是从“碎片化信息”到“体系化知识”**的跃迁。

学术知识图谱模式设计 (Schema Ontology)

为了实现演化关系和多维分类，需要定义一套严谨的图谱本体（Ontology）：

• 节点 (Nodes)：
  • Paper: 论文主体（标题、年份、 venue）。
  • Method/Tech: 技术手段（如：Transformer, Contrastive Learning）。
  • Metric/Dataset: 评估标准与数据集（如：ImageNet, mAP）。
  • Author/Org: 作者与机构。
• 关系 (Edges)：
  • EVOLVED_FROM: 演化关系（A 方法是基于 B 方法的改进）。
  • COMPETES_WITH: 竞争关系（针对同一问题的不同路径）。
  • USES: 使用关系（论文使用了某技术或数据集）。
  • CATEGORIZED_AS: 分类关系（属于某个多维标签）。

---

24、 图谱提取节点实现 (src/agents/nodes/graph_extractor.py)

需要一个新的 LangGraph 节点，专门负责从 DetailedAnalysis 中提取三元组（Subject-Predicate-Object）。

from pydantic import BaseModel, Field
from typing import List
from src.agents.state import AgentState
from src.core.llm_factory import LLMFactory

class Relationship(BaseModel):
    source: str = Field(description="源节点名称，如 'ResNet'")
    target: str = Field(description="目标节点名称，如 'CNN'")
    predicate: str = Field(description="关系类型，如 'INSTANCE_OF', 'IMPROVES_UPON'")
    description: str = Field(description="关系的简要依据")

class GraphExtraction(BaseModel):
    entities: List[dict] = Field(description="识别到的实体列表：名称与类型")
    relationships: List[Relationship] = Field(description="实体间的逻辑关系")

async def graph_extractor_node(state: AgentState):
    print("--- 执行：学术知识图谱三元组提取 ---")
    
    analysis = state.get("analysis")
    # 引导 LLM 进行本体抽取
    prompt = f"""基于以下论文深度分析结果，提取其中的关键实体（技术、模型、数据集）
    及其演化关系（改进了谁、属于什么分类）。
    内容：{analysis}"""

    llm = LLMFactory.get_model(model_type="high_precision")
    structured_llm = llm.with_structured_output(GraphExtraction)
    
    try:
        graph_data = await structured_llm.ainvoke(prompt)
        # 将提取到的图数据存入状态，准备写入数据库
        return {
            "graph_triplets": graph_data.dict(),
            "messages": [f"提取了 {len(graph_data.relationships)} 条学术关系。"]
        }
    except Exception as e:
        return {"messages": [f"图谱提取失败: {str(e)}"]}

---

25、 本地图数据库持久化 (src/database/graph_db.py)

对于本地高性能图处理，建议使用 Neo4j（标准选型）或者更轻量级的 FalkorDB。这里设计一个通用的 GraphStorage 接口。

from neo4j import GraphDatabase

class AcademicGraphDB:
    def __init__(self, uri="bolt://localhost:7687", user="neo4j", password="password"):
        self.driver = GraphDatabase.driver(uri, auth=(user, password))

    def upsert_paper_relation(self, triplet: dict):
        """写入或更新图谱中的关系"""
        with self.driver.session() as session:
            # Cypher 语句示例：创建演化关系
            query = """
            MERGE (s:Entity {name: $source})
            MERGE (t:Entity {name: $target})
            MERGE (s)-[r:RELATION {type: $predicate}]->(t)
            SET r.description = $desc
            """
            session.run(query, source=triplet['source'], target=triplet['target'], 
                        predicate=triplet['predicate'], desc=triplet['description'])

为了实现“演化关系”展示，需要在 comparer 节点之后加入一个**“脉络回溯”**逻辑：

1. 分类对齐：通过 LLM 将论文自动归类到预设的分类树（如：Deep Learning -> NLP -> LLM -> Fine-tuning）。
2. 版本追踪：识别论文中提到的版本迭代（如：V1, V2, Pro）。
3. 技术路径对比：在图谱中标记出“主流路径”与“非主流创新路径”。

---

26、 完善 LangGraph 拓扑结构

将图谱写入作为一个异步的“旁路”任务，或者作为工作流的最后一环。

from langgraph.graph import StateGraph, END
from src.agents.state import AgentState
from src.agents.nodes.parser import paper_parser_node
from src.agents.nodes.cleaner import paper_cleaner_node 
from src.agents.nodes.analyzer import research_analyzer_node
from src.agents.nodes.vision_processor import vision_processor_node # 新增

# ... 导入其他节点

def create_research_graph():
    # 初始化状态图
    workflow = StateGraph(AgentState)

    # 添加节点
    workflow.add_node("parser", paper_parser_node)
    workflow.add_node("vision", vision_processor_node) # 新增
    workflow.add_node("cleaner", paper_cleaner_node) # 插入清洗节点
    workflow.add_node("analyzer", research_analyzer_node)
    workflow.add_node("comparer", comparison_node) # 新增对比节点
    workflow.add_node("generator", output_generator_node)
    workflow.add_node("graph_extractor", graph_extractor_node) # 提取关系

    # 构建边（确定逻辑流转）
    workflow.set_entry_point("parser")
    
    # 解析完成后，将图片交给视觉节点
    workflow.add_edge("parser", "vision")
    # 视觉分析完成后，交给清洗节点进行图文整合
    workflow.add_edge("vision", "cleaner") # Parser -> Cleaner
    workflow.add_edge("cleaner", "analyzer") # Cleaner -> Analyzer
    
    # 这里可以添加条件分支 (Conditional Edges)
    # 例如：如果分析发现数据不足，可以跳回到 RAG 检索节点
    # 在生成产物的同时或之前，将知识沉淀进图谱
    workflow.add_edge("analyzer", "graph_extractor") # 在生成产物的同时或之前，将知识沉淀进图谱
    workflow.add_edge("graph_extractor", "comparer")# 分析完当前论文后，自动进入 RAG 进行库内对比
    workflow.add_edge("comparer", "generator")
    workflow.add_edge("generator", END)

    # 编译成可执行对象
    return workflow.compile()

# 实例化
app = create_research_graph()

# 在 main.py 中启动时：
# initial_state = {"pdf_path": "path/to/paper.pdf", "messages": []}
# app = create_research_graph()
# await app.ainvoke(initial_state)

增量更新：每当你研读一篇新论文，图谱不应只是增加一个点，而是应该去**“激活”**旧节点。例如，新论文说“克服了 ResNet 的梯度消失”，图谱应自动在 ResNet 节点旁建立一条反向连接。

可视化展示：为了方便本地查看，可以集成 Pyvis 库，在 output/ 目录下生成一个交互式的 knowledge_graph.html，可以直接用浏览器拖拽查看论文演化链条。

通过引入知识图谱，这个项目已经从“效率工具”升级为了“科研大脑”。它不仅能做 PPT，还能梳理整个学科的技术地图。现在，地基（LangGraph）、特种兵（MCP）、大脑（LLM）、记忆（RAG）和灵魂（Knowledge Graph）都已就位。

仪表盘（Dashboard） 是整个项目的“指挥中枢”，它将看不见的代码逻辑转化为看得见的“知识资产”。在这一步，将使用 Streamlit 构建前端，因为它能完美兼容 Python 的异步逻辑，并且可以通过 streamlit-agraph 或 pyvis 实现交互式的图谱展示。

27. 仪表盘架构设计 (src/ui/dashboard.py)

仪表盘将分为三个核心区域：

1. 侧边栏 (Sidebar)：负责 PDF 上传、处理进度显示和系统状态监控。
2. 主工作区 (Main Pane)：
   • 标签页 A (Deep Dive)：展示当前论文的深度剖析、图文融合结果和思维导图。
   • 标签页 B (Comparison)：展示多论文对比维度表。
   • 标签页 C (Knowledge Graph)：全量知识图谱的交互式看板。
3. 下载区 (Downloads)：PPT 和思维导图的导出入口。

A. 整体布局与上传逻辑

要确保 Streamlit 能够正确驱动 LangGraph 的异步流程。

import streamlit as st
import asyncio
from src.agents.graph import create_research_graph
from src.tools.map_gen import MindMapGenerator

st.set_page_config(page_title="Smart Scholar Agent", layout="wide")

st.title("🚀 智能论文研读与知识资产中心")

# 侧边栏：上传与配置
with st.sidebar:
    st.header("文件中心")
    uploaded_file = st.file_uploader("上传论文 PDF", type="pdf")
    process_btn = st.button("开始深度研读")
    
    st.divider()
    st.subheader("处理状态")
    status_log = st.empty() # 用于动态显示进度

# 主界面：多维度展示
tab1, tab2, tab3 = st.tabs(["📄 单篇深度剖析", "📊 横向维度对比", "🕸️ 学术知识图谱"])

B. 知识图谱的可视化 (利用 Pyvis)

这是为了实现你提到的“视觉冲击力”，将本地数据库中的关系渲染为可拖拽的节点。

from pyvis.network import Network
import streamlit.components.v1 as components

def render_knowledge_graph(entities, relationships):
    """利用 Pyvis 生成 HTML 图谱并在 Streamlit 中渲染"""
    net = Network(height="600px", width="100%", bgcolor="#ffffff", font_color="black")
    
    # 添加节点
    for ent in entities:
        color = "#3498db" if ent['type'] == 'Paper' else "#e74c3c"
        net.add_node(ent['name'], label=ent['name'], color=color)
        
    # 添加边
    for rel in relationships:
        net.add_edge(rel['source'], rel['target'], title=rel['description'], label=rel['predicate'])
    
    net.save_graph("temp_graph.html")
    with open("temp_graph.html", 'r', encoding='utf-8') as f:
        components.html(f.read(), height=650)

C. 集成 LangGraph 节点输出

在“开始研读”点击后，触发整个后端链路：

if process_btn and uploaded_file:
    # 1. 保存上传文件到临时目录
    with open(f"data/uploads/{uploaded_file.name}", "wb") as f:
        f.write(uploaded_file.getbuffer())
    
    # 2. 启动异步分析逻辑
    async def run_analysis():
        status_log.info("正在启动智能体...")
        graph = create_research_graph()
        initial_state = {
            "pdf_path": f"data/uploads/{uploaded_file.name}",
            "messages": []
        }
        
        # 通过 stream 模式获取中间节点信息
        async for output in graph.astream(initial_state):
            for node, state in output.items():
                status_log.success(f"节点 {node} 处理完成")
                # 这里可以根据 node 名称实时更新 UI 状态
        return state

    final_state = asyncio.run(run_analysis())
    st.session_state['final_state'] = final_state

为了让你真正感受到“知识资产”的增长，需要在仪表盘中加入以下设计：

1. 资产统计计数器 (Counters)：在首页上方显示：已研读论文：128篇 | 发现技术路径：42条 | 提取创新点：315个。这种数据增长的即时反馈非常激励人。
2. Mermaid 在线渲染：在 tab1 中使用 streamlit-mermaid 插件，直接将 artifacts 里的 Mermaid 代码渲染成可缩放的思维导图。
3. 对比表高亮：在 tab2 的数据表中，用颜色高亮当前论文优于 Baseline 的指标（例如精度提升用绿色加粗）。

---

为了长期稳健运行，在地基中还差最后一环：持久化缓存管理。

• 问题：Streamlit 每次刷新页面都会重新运行脚本。
• 对策：需要将解析好的 final_state 存入 SQLite 或 JSON 文件。下次打开 Dashboard 时，直接加载已有的分析结果，而不是重新调用昂贵的 LLM。

一个基于 LangGraph + MCP + RAG + KG 的闭环智能体研读系统。

1. 大脑：LangGraph 驱动的复杂逻辑。
2. 眼耳：MCP 驱动的视觉与文档解析。
3. 记忆：RAG 驱动的跨论文检索。
4. 体系：知识图谱驱动的演化脉络。
5. 展示：Streamlit 驱动的仪表盘产出。

通过引入 Arxiv 自动追踪能力，将这套系统从一个“被动响应的工具”升级为了一个“主动进化的智能体”。每天早晨打开 Dashboard，系统已经自动为你研读了昨晚发布的 5 篇相关领域论文，并把它们连接到了学术图谱中，甚至为你准备好了对比简报。这种**“知识动复利”是科研工作者的梦幻场景。为了实现这个功能，需要构建一个全新的 Arxiv MCP Server，并设计一套无人值守的自动化管线**。

28. 构建 Arxiv 抓取特种兵 (src/mcp_servers/arxiv_monitor/)

需要集成 arxiv 官方 API。这个服务负责根据关键词搜索最新论文，并过滤出你还没读过的部分。

# 安装依赖
pip install mcp arxiv

编写 src/mcp_servers/arxiv_monitor/server.py：

import arxiv
import asyncio
from mcp.server import Server
import mcp.types as types

server = Server("arxiv-monitor-service")

@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
    return [
        types.Tool(
            name="fetch_latest_papers",
            description="根据关键词搜索 Arxiv 上最新的论文列表",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "搜索关键词，如 'LLM Agent'"},
                    "max_results": {"type": "integer", "default": 5},
                },
                "required": ["query"],
            },
        )
    ]

@server.call_tool()
async def handle_call_tool(name: str, arguments: dict | None) -> list[types.TextContent]:
    if name != "fetch_latest_papers":
        raise ValueError(f"Unknown tool: {name}")

    query = arguments.get("query")
    max_results = arguments.get("max_results", 5)

    # 调用 Arxiv API
    search = arxiv.Search(
        query=query,
        max_results=max_results,
        sort_by=arxiv.SortCriterion.SubmittedDate
    )

    results = []
    for result in search.results():
        results.append({
            "title": result.title,
            "pdf_url": result.pdf_url,
            "published": result.published.strftime("%Y-%m-%d"),
            "entry_id": result.entry_id
        })
    
    import json
    return [types.TextContent(type="text", text=json.dumps(results))]

# ... (启动逻辑与之前类似)

---

29. 设计无人值守管线 (src/agents/batch_processor.py)

需要一个脱离 UI 运行的后台脚本，它像一个“守夜人”一样定期工作。

import asyncio
from src.core.mcp_client import MCPClientManager
from src.agents.graph import create_research_graph
from mcp import StdioServerParameters

ARXIV_PARAMS = StdioServerParameters(command="python", args=["src/mcp_servers/arxiv_monitor/server.py"])

async def autonomous_growth_task(interest_topics: list):
    """无人值守增长任务"""
    graph = create_research_graph()
    
    async with MCPClientManager(ARXIV_PARAMS) as client:
        for topic in interest_topics:
            print(f"--- 正在巡检话题: {topic} ---")
            # 发现新论文
            papers = await client.call_service_tool("fetch_latest_papers", {"query": topic})
            
            for paper in papers: # 这里需要一个去重逻辑，检查 RAG 或图谱中是否已存在
                # 触发完整的研读工作流
                print(f"--- 发现新论文，开始自动研读: {paper['title']} ---")
                initial_state = {
                    "pdf_path": paper['pdf_url'], # 的 Parser 节点需要支持直接下载 URL
                    "messages": [f"Arxiv 自动追踪触发"]
                }
                # 运行 LangGraph
                await graph.ainvoke(initial_state)
                print(f"--- 论文 {paper['title']} 已自动并入知识版图 ---")

# 可以使用 schedule 库或简单的 while 循环配合 sleep

---

30. Parser 节点的“远程适应”增强

为了支持无人值守，要补全 src/agents/nodes/parser.py 的功能，让它能处理 http:// 形式的 PDF 路径。

import httpx
import os
from mcp import StdioServerParameters
from src.core.mcp_client import MCPClientManager
from src.agents.state import AgentState
# 配置 MCP Server 参数 (假设通过 python 启动一个解析脚本)
PARSER_SERVER_PARAMS = StdioServerParameters(
    command="python",
    args=["src/mcp_servers/doc_parser/server.py"], # 对应工程结构中的路径
    env=None
)

async def paper_parser_node(state: AgentState):
    """
    补全后的解析节点：
    1. 建立与解析服务的连接
    2. 调用解析工具处理 PDF
    3. 更新状态机中的 structured_content
    """
    print(f"--- 正在调用 MCP 服务解析论文: {state['pdf_path']} ---")
    path = state["pdf_path"]
    if path.startswith("http"):
        # 下载 PDF 到本地临时目录
        local_path = f"data/cache/{path.split('/')[-1]}.pdf"
        async with httpx.AsyncClient() as client:
            resp = await client.get(path)
            with open(local_path, "wb") as f:
                f.write(resp.content)
        state["pdf_path"] = local_path
    
    try:
        async with MCPClientManager(PARSER_SERVER_PARAMS) as client:
            # 调用名为 'parse_pdf' 的 MCP 工具
            response = await client.call_service_tool(
                "parse_pdf", 
                {"path": state["pdf_path"]}
            )
            
            # 假设返回的是结构化的 Markdown 和元数据
            structured_data = response[0].text # 假设 MCP 返回 text 类型的 content
            
            return {
                "structured_content": {"raw_md": structured_data},
                "current_step": "analysis",
                "messages": [f"成功解析论文，获取到 {len(structured_data)} 字符数据。"]
            }
    except Exception as e:
        print(f"解析出错: {e}")
        return {"messages": [f"解析失败: {str(e)}"]}

仪表盘的“资产动向”通知，在 Dashboard 中，要增加一个“今日动态”通知栏：

• 今日新增资产：自动收录了 3 篇关于 LoRA 优化的论文。
• 图谱新连接：发现新论文 A 与你库中的论文 B 存在显著的演化关系。

当开启“自动追踪”后，系统会面临新的挑战：

1. 并发压力：如果 Arxiv 一下子更新了 20 篇相关论文， GPU 或 API Key 可能会爆。需要在 batch_processor 中引入队列机制，确保论文是一个接一个有序处理。
2. 质量过滤：不是所有 Arxiv 论文都值得进入图谱。可以增加一个 Pre-Filter 节点，让 LLM 只根据“摘要”判断论文质量，只有高分论文才进入昂贵的“视觉解析”和“深度剖析”环节。

通过结合 RAG（局部语义） 和 知识图谱（全局关联），实现的将是业界最前沿的 GraphRAG 模式。普通的搜索只能告诉你“谁提到了这个词”，而你的全局搜索能告诉你“这个问题的技术演化脉络是什么，库中哪几篇论文形成了对立的观点”。为了实现高质量的回答，需要构建一个**多路召回（Multi-Route Retrieval）**引擎：

1. 向量路径 (Vector Route)：检索具体的段落、实验数据。
2. 图谱路径 (Graph Route)：检索论文间的引用关系、作者链条、技术演化路径。
3. 结构化路径 (SQL/Metadata Route)：按年份、期刊、关键词进行精确过滤。

---

31. 实现搜索路由逻辑 (src/agents/nodes/search_engine.py)

需要一个“意图识别”逻辑，判断用户是在问具体细节，还是在问宏观趋势。

from src.core.llm_factory import LLMFactory
from src.rag.vector_store import PaperVectorStore
from src.database.graph_db import AcademicGraphDB

async def global_search_engine(query: str):
    """全局搜索中枢"""
    llm = LLMFactory.get_model(model_type="high_precision")
    
    # 意图分析：用户到底在问什么？
    intent_prompt = f"分析用户查询意图，决定检索策略：'{query}'。输出：VECTOR, GRAPH, 或 HYBRID。"
    # (此处省略意图识别模型调用)

    # 执行双路检索 向量检索：寻找语义相似的片段
    vector_results = await PaperVectorStore().search_similar_papers(query, n_results=5)
    
    # 图谱检索：寻找关联的实体和演化路径 例如查询：MATCH (p:Paper)-[:EVOLVED_FROM]->(parent) WHERE p.mentions = '注意力弥散' ...
    graph_results = AcademicGraphDB().query_related_nodes(query)

    # 结果聚合与推理生成
    final_prompt = f"""
    你是一个拥有全局视野的学术助手。基于以下检索到的信息回答用户问题：
    
    【局部细节 (RAG)】: {vector_results}
    【知识图谱 (KG)】: {graph_results}
    
    用户问题：{query}
    
    要求：回答要专业，引用库中的论文标题，并指出不同论文间的关联或矛盾点。
    """
    
    response = await llm.ainvoke(final_prompt)
    return response.content

---

32. Dashboard 交互界面实现 (src/ui/dashboard.py)

在 Streamlit 中，将这个功能设计成一个**“学术对话悬浮窗”**。

# 在 tab3 之后增加 tab4
with tab4:
    st.header("🔍 全局学术对话")
    st.caption("基于你本地的 RAG 库和知识图谱进行深度回答")
    
    # 使用 Streamlit 的聊天交互组件
    if "messages" not in st.session_state:
        st.session_state.messages = []

    # 展示历史对话
    for message in st.session_state.messages:
        with st.chat_message(message["role"]):
            st.markdown(message["content"])

    # 接收新提问
    if prompt := st.chat_input("问问你的论文库：例如 '注意力弥散问题的解决方案有哪些？'"):
        st.session_state.messages.append({"role": "user", "content": prompt})
        with st.chat_message("user"):
            st.markdown(prompt)

        with st.chat_message("assistant"):
            with st.spinner("正在穿越知识图谱并调取 RAG 资产..."):
                # 调用上面写的搜索引擎
                response = asyncio.run(global_search_engine(prompt))
                st.markdown(response)
        
        st.session_state.messages.append({"role": "assistant", "content": response})

• 从“找”到“问”：你不再需要打开 10 个 PDF 搜索同一个关键词。智能体通过 GraphRAG 直接告诉你：“论文 A 提出了问题，论文 B 改进了它，论文 C 的消融实验证明了该方案在长文本场景下的局限性。”
• 证据溯源：每一个回答都会带上**【来源论文】**的标签。点击标签，Dashboard 自动跳转到该论文的深度剖析页面。
• 消除幻觉：因为所有回答都严格基于你本地的 PaperStructuredData 和图谱关系，LLM “瞎编”的可能性被降到了最低。

现在的系统能力清单：

1. 全自动巡检：Arxiv 每天自动喂入最新鲜的知识。
2. 多模态解析：Docling 和视觉 MCP 配合，不放过任何公式和图表。
3. 体系化存储：RAG 存细节，KG 存脉络。
4. 专业化输出：一键生成对比 PPT、思维导图。
5. 全局化交互：通过对话窗口，与你过去几年积累的所有知识进行即时对谈。

AI代理是系统，通过为大型语言模型（LLMs）提供工具和知识的访问权限，扩展其能力，使其能够执行操作。ai-agents-for-beginners/translations/zh/01-intro-to-ai-agents/README.md at main · microsoft/ai-agents-for-beginners · GitHub 将这个定义拆分为几个部分：

• 系统-重要的是要将代理视为一个由多个组件组成的系统，而不是单一组件。AI代理的基本组件包括：
  • 环境 - AI代理运行的定义空间。例如，如果有一个旅行预订AI代理，环境可能是AI代理用来完成任务的旅行预订系统。
  • 传感器 - 环境提供信息和反馈。AI代理使用传感器收集并解释有关环境当前状态的信息。在旅行预订代理的例子中，旅行预订系统可以提供酒店可用性或航班价格等信息。
  • 执行器 - 一旦AI代理接收到环境的当前状态，代理会根据当前任务决定采取什么行动来改变环境。对于旅行预订代理来说，这可能是为用户预订一个可用的房间。
• 大型语言模型 - 代理的概念在LLMs出现之前就已经存在。使用LLMs构建AI代理的优势在于它们能够解释人类语言和数据。这种能力使LLMs能够解释环境信息并制定改变环境的计划。
• 执行操作 - 在AI代理系统之外，LLMs的作用通常局限于根据用户提示生成内容或信息。在AI代理系统内，LLMs可以通过解释用户请求并使用环境中可用的工具来完成任务。
• 工具访问权限 - LLM可以访问的工具由1）其运行的环境和2）AI代理开发者定义。在旅行代理示例中，代理的工具受限于预订系统的操作，开发者也可以限制代理的工具访问权限，例如仅限于航班。
• 记忆+知识 - 记忆可以是短期的，例如用户与代理之间的对话内容。长期来看，除了环境提供的信息外，AI代理还可以从其他系统、服务、工具甚至其他代理中检索知识。在旅行代理示例中，这些知识可能是客户数据库中用户旅行偏好的信息。

论文研读智能体的“三层开发范式”

1. 数据与感知层 (The Sensing Layer) —— 解决“看”的问题

• 核心技术：RAG + MCP (Vision)。
• 理论意义：解决 LLM 无法处理长文本、无法精准识图的问题。
• 工程作用：提供“原材料”。RAG 提供文本片段，MCP 提供通过 YOLO/OCR 处理后的精准视觉坐标和公式。
• 代码连接：这是 Agent 调用的外部 API。

2. 工作流编排层 (The Workflow Layer) —— 解决“走”的问题

• 核心技术：LangGraph (State Machine)。
• 理论意义：通过“图”限制 Agent 的随机性。Agent 不能乱跑，必须先规划，再行动，最后评估。
• 工程作用：定义 State（状态）。就像 PyTorch 里的 Tensor 在各层流动，State 在节点间流动。
• 代码连接：这是 workflow.add_node 和 add_edge。

3. 反思与自愈层 (The Evaluation Layer) —— 解决“对”的问题

• 核心技术：Self-Correction / Grader。
• 理论意义：闭环控制。如果 Agent 发现 RAG 回来的内容不对，它会重新搜。
• 工程作用：引入 Conditional Edges (条件边)。
• 代码连接：这是 should_continue 逻辑。

1. state.py：定义系统的“共享内存”

这是所有节点（Node）通信的唯一协议。可以把它类比为在网络中传递的 Feature Map。

from typing import Annotated, List, TypedDict, Optional
from pydantic import BaseModel, Field
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages

class PaperSummary(BaseModel):
    """结构化的论文摘要模型"""
    title: str = Field(description="论文标题")
    methodology: str = Field(description="核心方法论，如网络架构改动")
    experiments: List[str] = Field(description="关键实验指标，如 mAP, FPS")
    is_complete: bool = Field(description="当前获取的信息是否足以回答用户")

class AgentState(TypedDict):
    # 自动追加的消息列表 (加法累积)
    messages: Annotated[List[BaseMessage], add_messages]
    # 当前研究的 PDF 路径
    paper_path: str
    # 结构化笔记
    summary: PaperSummary
    # 待办任务清单
    todo_list: List[str]

---

2. cv_mcp_server.py：眼睛 (感知层)

这是一个独立的进程，专门运行 CV 模型（如 YOLO 或 OCR）。它与大脑解耦。

from mcp.server.fastmcp import FastMCP
import PIL.Image

mcp = FastMCP("Paper-Vision-Server")

@mcp.tool()
def analyze_paper_figure(image_path: str, query: str) -> str:
    """
    输入图像路径和查询内容（如：分析图 5 的曲线趋势），
    返回 CV 模型（OCR/检测）的结构化分析结果。
    """
    # 这里接入 YOLO/DBNet 模型代码
    # 示例返回
    return f"检测到图中存在坐标轴。X轴为 Epoch，Y轴为 mAP。曲线在 50 epoch 处达到 0.45 峰值。"

if __name__ == "__main__":
    mcp.run()

---

3. tools/：感官接口 (Tools Layer)

这里存放大脑如何调用外部能力的定义，包括 RAG 检索和连接 MCP。

# tools/mcp_client.py
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from langchain_core.tools import tool

server_params = StdioServerParameters(command="python", args=["cv_mcp_server.py"])

@tool
async def vision_tool(image_path: str, query: str):
    """调用远程 CV MCP Server 识别论文图表、公式或布局"""
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            result = await session.call_tool("analyze_paper_figure", {"image_path": image_path, "query": query})
            return result.content[0].text

# tools/rag_tool.py
@tool
def rag_search_tool(query: str):
    """在论文文本数据库中进行语义搜索"""
    # 实现 ChromaDB 检索逻辑
    return "检索到的论文片段内容..."

---

4. nodes.py：大脑逻辑 (Logic Layer)

定义每个节点的具体行为。

from langchain_openai import ChatOpenAI
from state import AgentState, PaperSummary

llm = ChatOpenAI(model="gpt-4o")
# 将所有工具绑定给 LLM
tools = [vision_tool, rag_search_tool]
llm_with_tools = llm.bind_tools(tools)

async def researcher_node(state: AgentState):
    """执行者节点：负责根据当前消息，决定是查 RAG 还是看图"""
    response = await llm_with_tools.ainvoke(state["messages"])
    return {"messages": [response]}

def analyzer_node(state: AgentState):
    """分析者节点：负责将收到的信息整理进 structured_summary"""
    # 使用 structured_output 提取信息
    extractor = llm.with_structured_output(PaperSummary)
    # 结合最近的消息更新笔记
    new_summary = extractor.invoke(state["messages"][-1].content)
    return {"summary": new_summary}

---

5. graph.py：神经中枢 (Workflow Layer)

将所有节点连成图，实现闭环。

from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from nodes import researcher_node, analyzer_node, tools

def should_continue(state: AgentState):
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tools"
    if not state["summary"].is_complete:
        return "continue"
    return "end"

workflow = StateGraph(AgentState)

workflow.add_node("researcher", researcher_node)
workflow.add_node("analyzer", analyzer_node)
workflow.add_node("tools", ToolNode(tools))

workflow.set_entry_point("researcher")

# 循环逻辑：researcher -> 如果有工具调用 -> tools -> researcher
workflow.add_conditional_edges(
    "researcher",
    should_continue,
    {"tools": "tools", "continue": "analyzer", "end": END}
)

workflow.add_edge("tools", "researcher")
workflow.add_edge("analyzer", "researcher") # 整理完笔记后，看是否需要进一步搜索

app = workflow.compile()