传统的检索增强生成(RAG)系统虽然能够高效地查找和总结事实信息,但其核心缺陷在于缺乏真正的思考能力。这就像是一个拥有海量书籍却不会辩证思考的图书管理员——它能快速找到相关段落,却无法像人类专家那样理解问题本质、识别潜在歧义或进行因果推理。
我们的目标是构建一个Agentic RAG系统,它能够模拟金融分析师处理复杂问题的完整认知流程。这个系统将具备以下核心能力:
在实际测试中,当被问及"微软云业务表现如何"时,基础RAG系统直接返回了Azure营收数据。而我们的Agentic RAG则会先确认用户关注的是增长趋势、市场份额还是盈利能力,然后分别调用趋势分析工具、竞争情报检索和财务比率计算来构建完整答案。
系统采用模块化设计,各组件通过明确定义的接口通信:
code复制[用户问题] →
Gatekeeper(歧义检测) →
Planner(任务规划) →
[专家工具执行] →
Auditor(质量检查) →
Strategist(洞察合成) →
[最终回答]
每个组件都是可独立升级的Python类,通过状态字典(AgentState)共享上下文。这种设计使得我们可以单独优化某个认知环节(如改进规划算法)而不影响其他模块。
与传统RAG的最大区别在于我们的多层次知识表示方法:
python复制# 元数据生成示例
class ChunkMetadata(BaseModel):
summary: str = Field(description="1-2句内容摘要")
keywords: List[str] = Field(description="5-7个关键主题词")
hypothetical_questions: List[str] = Field(description="本片段能回答的3-5个问题")
table_summary: Optional[str] = Field(default=None)
这种富元数据策略使向量检索的准确率提升了47%(基于我们的基准测试),因为搜索可以同时匹配原始文本和LLM提炼的高阶概念。
金融文档检索面临两大挑战:(1)专业术语众多 (2)需要跨文档关联信息。我们的解决方案是三级检索流程:
python复制"微软云业务怎么样?" → "分析Microsoft Intelligent Cloud板块收入增长,重点关注Azure和云服务的贡献"
针对财务数据表开发了两类分析工具:
精确查询工具:
python复制@tool
def analyst_sql_tool(query: str) -> str:
"""回答特定时间点的财务指标查询"""
# 示例:SELECT revenue FROM financials WHERE quarter='Q4-2023'
return execute_sql(query)
趋势分析工具:
python复制def calculate_growth(df):
# 自动计算季度环比(QoQ)和年度同比(YoY)
df['QoQ'] = df['revenue'].pct_change()
df['YoY'] = df['revenue'].pct_change(4)
return df
测试显示,这种分工使财务问题回答速度提升60%,因为简单查询不再需要启动复杂的趋势分析流程。
通过Tavily API实现:
特别优化了金融领域搜索的提示词工程,确保返回结果与商业分析相关。
AgentState是整个系统的"工作记忆",采用TypedDict确保类型安全:
python复制class AgentState(TypedDict):
original_request: str
clarification_question: Optional[str]
plan: List[str]
intermediate_steps: List[Dict]
verification_history: List[Dict]
final_response: str
关键创新在于verification_history字段,它记录了:
模糊问题检测算法流程:
采用基于链式思考(CoT)的规划方法:
python复制def create_plan(request):
# 步骤1:问题分类
if "趋势" in request:
return ["analyst_trend_tool", "FINISH"]
elif "对比" in request:
return ["librarian_rag_tool", "analyst_sql_tool", "FINISH"]
...
实际部署中,我们使用LLM生成更动态的计划,但保留上述规则引擎作为回退方案。
SEC文件中的财务表格是重要信息源,但传统RAG处理存在:
问题:表格被拆分成多个片段后失去语义完整性
解决方案:
当多个工具适合回答同一问题时:
问题:如何避免重复/冲突结果
解决方案:
python复制def check_consistency(tool1_output, tool2_output):
# 使用LLM比较两个结果的语义一致性
return llm.compare(f"是否一致:\n1.{tool1_output}\n2.{tool2_output}")
问题:SEC文件有3个月延迟,如何提供最新分析
解决方案:
分层索引:
预过滤策略:
python复制# 先按时间范围过滤,再执行语义搜索
filter = Filter(year="2023")
client.search(..., filter=filter)
python复制async def parallel_tools(tool1, tool2):
return await asyncio.gather(tool1(), tool2())
我们在金融QA测试集上对比了三种系统:
| 指标 | 传统RAG | Agentic RAG | 人类专家 |
|---|---|---|---|
| 答案准确率 | 62% | 89% | 95% |
| 澄清问题提出率 | 5% | 78% | 85% |
| 平均响应时间(秒) | 2.1 | 6.7 | 120+ |
复合问题处理:
"比较Azure和AWS过去两年的营收增长"
隐含假设识别:
"按当前增速预测明年云收入"
短期:
长期:
在实际部署中,这套系统已经成功应用于:
通过持续迭代,我们正逐步缩小AI系统与人类专家在商业分析领域的差距。未来的工作将聚焦于提升系统的解释能力和预测准确性,使其真正成为决策者的智能助手而非仅仅是信息检索工具。