1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
|
class PaperQAAgent:
"""使用PaperQA2进行科学文献分析的高级AI代理"""
def __init__(self, papers_directory: str, temperature: float = 0.1):
self.settings = create_gemini_settings(papers_directory, temperature)
self.papers_dir = papers_directory
print(f"PaperQA代理已初始化，论文来源: {papers_directory}")
async def ask_question(self, question: str, use_agent: bool = True):
"""向研究论文提问"""
print(f"\n问题: {question}")
print("正在研究论文中搜索...")
try:
if use_agent:
response = await agent_query(query=question, settings=self.settings)
else:
response = ask(question, settings=self.settings)
return response
except Exception as e:
print(f"问题处理错误: {e}")
return None
def display_answer(self, response):
"""格式化显示答案"""
if response is None:
print("未收到响应")
return
print("\n" + "="*60)
print("答案:")
print("="*60)
answer_text = getattr(response, 'answer', str(response))
print(f"\n{answer_text}")
contexts = getattr(response, 'contexts', getattr(response, 'context', []))
if contexts:
print("\n" + "-"*40)
print("使用的来源:")
print("-"*40)
for i, context in enumerate(contexts[:3], 1):
context_name = getattr(context, 'name', getattr(context, 'doc', f'来源 {i}'))
context_text = getattr(context, 'text', getattr(context, 'content', str(context)))
print(f"\n{i}. {context_name}")
print(f" 文本预览: {context_text[:150]}...")
async def multi_question_analysis(self, questions: list):
"""按顺序分析多个问题"""
results = {}
for i, question in enumerate(questions, 1):
print(f"\n处理问题 {i}/{len(questions)}")
response = await self.ask_question(question)
results = response
if response:
print(f"已完成: {question[:50]}...")
else:
print(f"失败: {question[:50]}...")
return results
async def comparative_analysis(self, topic: str):
"""跨论文执行比较分析"""
questions = [
f"{topic}中的关键创新是什么？",
f"当前{topic}方法的局限性是什么？",
f"{topic}的未来研究方向有哪些建议？",
]
print(f"\n开始比较分析: {topic}")
return await self.multi_question_analysis(questions)
async def basic_demo():
"""演示基本PaperQA功能"""
agent = PaperQAAgent(papers_directory)
question = "什么是Transformer架构？为什么它很重要？"
response = await agent.ask_question(question)
agent.display_answer(response)
print("运行基本演示...")
await basic_demo()
async def advanced_demo():
"""演示高级多问题分析"""
agent = PaperQAAgent(papers_directory, temperature=0.2)
questions = [
"Transformer中的注意力机制如何工作？",
"大语言模型的计算挑战是什么？",
"自然语言处理中的预训练是如何演变的？"
]
print("运行高级多问题分析...")
results = await agent.multi_question_analysis(questions)
for question, response in results.items():
print(f"\n{'='*80}")
print(f"问题: {question}")
print('='*80)
if response:
answer_text = getattr(response, 'answer', str(response))
display_text = answer_text[:300] + "..." if len(answer_text) > 300 else answer_text
print(display_text)
else:
print("无可用答案")
print("\n运行高级演示...")
await advanced_demo()
async def research_comparison_demo():
"""演示比较研究分析"""
agent = PaperQAAgent(papers_directory)
results = await agent.comparative_analysis("神经网络中的注意力机制")
print("\n" + "="*80)
print("比较分析结果")
print("="*80)
for question, response in results.items():
print(f"\n{question}")
print("-" * 50)
if response:
answer_text = getattr(response, 'answer', str(response))
print(answer_text)
else:
print("分析不可用")
print()
print("运行比较研究分析...")
await research_comparison_demo()
|