1. 项目概述
作为一名长期从事AI系统开发的工程师,我最近完成了一个面向科研场景的论文RAG问答系统的架构升级。这个项目最初只是一个简单的检索增强生成(RAG)系统,但随着功能需求的增加,逐渐演变成一个支持多轮对话、工具调用和复杂任务编排的智能Agent架构。本文将详细分享这次架构升级的技术细节和工程实践。
科研工作者在日常工作中经常面临这样的困境:当需要引用某个理论时,记不清具体出自哪篇论文;当需要理解某个方法时,不得不反复查阅多篇文献;当需要对比不同研究时,又得花费大量时间整理笔记。我们的系统正是为了解决这些痛点而设计的。
2. 基础RAG系统的局限性
2.1 单轮问答的不足
传统的RAG系统通常只能处理单轮问答场景:用户提问→系统检索→生成回答。但在真实的科研工作中,研究者往往需要进行多轮深入探讨。例如:
- 先问"这篇论文的主要贡献是什么?"
- 接着追问"实验部分用了哪些数据集?"
- 然后继续"这些数据集在哪里可以获取?"
这种连续性的知识探索过程,是简单RAG系统难以支持的。
2.2 工具能力的缺失
科研场景中经常需要结合多种工具能力:
- 时间查询:"这篇论文是什么时候发表的?"
- 简单计算:"两组实验结果的p值是多少?"
- 概念解释:"什么是注意力机制?"
这些需求超出了纯文本检索的范畴,需要系统具备调用外部工具的能力。
2.3 状态管理的挑战
随着功能增加,系统需要维护的对话状态也越来越复杂:
- 当前会话历史
- 已检索的论文片段
- 工具调用记录
- 临时计算结果
这些状态如果管理不当,很容易导致系统行为不一致或难以调试。
3. 引入LangGraph的决策过程
3.1 前期的手工实现
在项目初期,我们通过Python函数直接实现了基本流程:
python复制def handle_query(query, history):
# 1. 工具选择
tool, params = choose_tool(query, history)
# 2. 工具执行
result = execute_tool(tool, params)
# 3. 生成回答
answer = generate_answer(result)
return answer
这种方式在小规模时还能应付,但随着功能增加,代码变得越来越难以维护。
3.2 状态管理的痛点
主要遇到三个问题:
- 所有逻辑都堆在一个函数里,难以单独测试某部分功能
- 中间状态分散在各个变量中,调试困难
- 添加新功能时需要修改多处代码,容易引入错误
3.3 LangGraph的优势
LangGraph是一个基于状态图的AI工作流编排框架,它提供了:
- 清晰的节点边界:每个处理步骤都是独立的
- 显式的状态管理:所有中间结果都集中存储
- 灵活的工作流:支持条件分支和循环
这些特性正好解决了我们面临的问题,因此决定用它重构系统。
4. 系统架构设计
4.1 整体架构层次
系统采用分层设计,从上到下依次为:
- 接口层:FastAPI实现的REST API
- 会话管理层:维护对话历史和上下文
- 编排层:LangGraph实现的工作流引擎
- 工具层:各种功能模块(RAG、LLM、计算器等)
- 检索层:向量数据库和检索逻辑
- 数据层:论文PDF处理和索引构建
4.2 核心组件交互
mermaid复制graph TD
A[用户] -->|提问| B(FastAPI接口)
B --> C{会话管理}
C -->|新建/获取会话| D[LangGraph工作流]
D --> E[工具选择节点]
E --> F[工具执行节点]
F --> G[回答生成节点]
G --> H[返回结果]
F -->|RAG查询| I[向量数据库]
I --> F
注意:实际实现中每个节点都是独立的Python函数,通过共享状态对象进行通信。
4.3 状态对象设计
我们定义了一个AgentState类来封装所有中间状态:
python复制class AgentState(TypedDict):
session_id: str
query: str
chat_history: List[Dict]
decision: Optional[Dict]
tool_result: Optional[Dict]
final_answer: Optional[str]
error: Optional[str]
这种强类型设计有助于早期发现状态不一致的问题。
5. 工作流实现细节
5.1 节点划分原则
我们将整个处理流程划分为三个核心节点:
- choose_tool:分析用户意图,决定使用哪个工具
- execute_tool:执行选定的工具并获取结果
- generate_answer:将工具结果转化为自然语言回答
每个节点都遵循单一职责原则,只关注自己的特定任务。
5.2 状态流转示例
假设用户问:"Attention机制最早是在哪篇论文提出的?"
- 初始状态:
json复制{
"session_id": "123",
"query": "Attention机制最早是在哪篇论文提出的?",
"chat_history": []
}
- choose_tool后:
json复制{
"decision": {
"tool": "rag",
"input": "Attention机制最早是在哪篇论文提出的?"
}
}
- execute_tool后:
json复制{
"tool_result": {
"tool_name": "rag",
"tool_input": "...",
"tool_output": {
"answer": "Attention机制最早出现在...",
"sources": ["论文A", "论文B"]
}
}
}
- generate_answer后:
json复制{
"final_answer": "根据检索结果,Attention机制最早是在论文A(2014年)中提出的..."
}
5.3 错误处理机制
每个节点都可能失败,因此我们设计了统一的错误处理流程:
python复制def choose_tool(state):
try:
# 正常处理逻辑
state["decision"] = {...}
except Exception as e:
state["error"] = str(e)
return state
错误信息会一直传递到最终节点,生成友好的错误提示。
6. 工程实践与优化
6.1 会话管理实现
会话管理器维护了两个核心数据结构:
python复制class SessionManager:
def __init__(self):
self.sessions = {} # session_id -> Session
self.locks = {} # session_id -> Lock
使用读写锁确保并发安全,避免多请求同时修改同一会话。
6.2 检索性能优化
针对RAG模块做了以下优化:
- 分层索引:按论文领域建立多个FAISS索引
- 动态重排序:使用ColBERT进行结果重排
- 缓存机制:高频查询结果缓存5分钟
这些优化使平均检索延迟从1200ms降到了400ms。
6.3 工具扩展模式
新增工具的步骤非常标准化:
- 在
choose_tool中添加识别逻辑 - 实现工具函数
- 在
execute_tool中添加调用入口
例如添加时间查询工具:
python复制# choose_tool.py
if "什么时候" in query or "时间" in query:
return {"tool": "time", "input": query}
# tools/time.py
def get_current_time():
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
7. 实际应用效果
7.1 典型使用场景
-
文献回顾:
- 用户:""这篇论文的方法部分讲了什么?"
- 系统:检索并总结方法章节
- 用户:"实验设置是怎样的?"
- 系统:基于上文理解这是同一篇论文的追问
-
概念溯源:
- 用户:"Transformer架构是谁提出的?"
- 系统:"Vaswani等人在2017年的论文..."
- 用户:"这篇论文被引用了多少次?"
- 系统:调用学术API查询引用数
7.2 性能指标
在1000次API调用的压力测试中:
- 平均响应时间:1.2s
- 成功率:98.7%
- 最长会话轮次:15轮
7.3 用户反馈
收集的典型评价包括:
- "比直接搜索论文高效多了"
- "能记住之前的对话上下文很实用"
- "计算结果直接嵌入回答很方便"
8. 经验总结与改进方向
8.1 关键设计决策
- 状态集中管理:所有中间结果都存储在AgentState中,极大简化了调试
- 节点松耦合:每个节点只依赖状态对象,不直接调用其他节点
- 显式错误处理:错误作为正常流程的一部分,而不是异常
8.2 遇到的挑战
- 长对话状态膨胀:解决方案是自动总结历史对话
- 工具冲突:两个工具可能同时匹配同一问题,需要优先级机制
- LLM响应不一致:通过严格的输出格式约束来改善
8.3 后续优化方向
- 支持复合工具:组合多个工具完成复杂任务
- 加入验证节点:检查工具结果可信度
- 实现工作流版本控制:便于AB测试不同流程
这个架构已经证明了其在科研辅助场景的价值,团队正在将其扩展应用到更多专业领域。对于想要构建类似系统的开发者,我的建议是从小规模验证开始,逐步迭代,重点关注状态设计和节点边界。
