1
from typing import List, Dict, Optional
2
from langchain_core.tools import tool
3
import requests
4
import os
5

6
class SearchTools:
7
    """搜索工具集"""
8

9
    def __init__(self):
10
        self.serper_api_key = os.getenv("SERPER_API_KEY")
11
        self.tavily_api_key = os.getenv("TAVILY_API_KEY")
12

13
    @tool
14
    def web_search(self, query: str, num_results: int = 5) -> List[Dict]:
15
        """
16
        搜索网络获取信息。
17

18
        Args:
19
            query: 搜索关键词
20
            num_results: 返回结果数量
21

22
        Returns:
23
            搜索结果列表，包含标题、链接、摘要
24
        """
25
        url = "https://google.serper.dev/search"
26
        headers = {
27
            "X-API-KEY": self.serper_api_key,
28
            "Content-Type": "application/json"
29
        }
30
        payload = {"q": query, "num": num_results}
31

32
        response = requests.post(url, json=payload, headers=headers)
33
        results = response.json().get("organic", [])
34

35
        return [
36
            {
37
                "title": r.get("title", ""),
38
                "link": r.get("link", ""),
39
                "snippet": r.get("snippet", ""),
40
                "position": r.get("position", 0)
41
            }
42
            for r in results[:num_results]
43
        ]
44

45
    @tool
46
    def deep_search(self, query: str, search_depth: str = "basic") -> Dict:
47
        """
48
        深度搜索，获取更全面的信息。
49

50
        Args:
51
            query: 搜索关键词
52
            search_depth: 搜索深度 (basic/advanced)
53

54
        Returns:
55
            包含搜索结果和答案的字典
56
        """
57
        from tavily import TavilyClient
58

59
        client = TavilyClient(api_key=self.tavily_api_key)
60
        result = client.search(
61
            query=query,
62
            search_depth=search_depth,
63
            max_results=10
64
        )
65

66
        return {
67
            "answer": result.get("answer", ""),
68
            "results": result.get("results", []),
69
            "follow_up_questions": result.get("follow_up_questions", [])
70
        }
71

72
    @tool
73
    def fetch_webpage(self, url: str) -> str:
74
        """
75
        获取网页内容。
76

77
        Args:
78
            url: 网页 URL
79

80
        Returns:
81
            网页文本内容
82
        """
83
        from bs4 import BeautifulSoup
84

85
        response = requests.get(url, timeout=30)
86
        soup = BeautifulSoup(response.text, 'html.parser')
87

88
        # 移除脚本和样式
89
        for script in soup(["script", "style"]):
90
            script.decompose()
91

92
        text = soup.get_text(separator='\n', strip=True)
93

94
        # 清理多余空白
95
        lines = [line.strip() for line in text.splitlines() if line.strip()]
96
        return '\n'.join(lines[:200])  # 限制长度
97

98

99
# 创建工具实例
100
search_tools = SearchTools()

1
from typing import List, Dict, Any, Optional
2
from langchain_core.tools import tool
3
import sqlite3
4
import json
5

6
class DatabaseTools:
7
    """数据库工具集"""
8

9
    def __init__(self, db_path: str = "./data/research.db"):
10
        self.db_path = db_path
11
        self._init_database()
12

13
    def _init_database(self):
14
        """初始化数据库表"""
15
        conn = sqlite3.connect(self.db_path)
16
        cursor = conn.cursor()
17

18
        # 研究记录表
19
        cursor.execute("""
20
            CREATE TABLE IF NOT EXISTS research_records (
21
                id INTEGER PRIMARY KEY AUTOINCREMENT,
22
                topic TEXT NOT NULL,
23
                query TEXT,
24
                result TEXT,
25
                sources TEXT,
26
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
27
            )
28
        """)
29

30
        # 用户偏好表
31
        cursor.execute("""
32
            CREATE TABLE IF NOT EXISTS user_preferences (
33
                id INTEGER PRIMARY KEY AUTOINCREMENT,
34
                user_id TEXT,
35
                preference_key TEXT,
36
                preference_value TEXT,
37
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
38
                UNIQUE(user_id, preference_key)
39
            )
40
        """)
41

42
        conn.commit()
43
        conn.close()
44

45
    @tool
46
    def execute_query(self, sql: str) -> List[Dict]:
47
        """
48
        执行 SQL 查询。
49

50
        Args:
51
            sql: SELECT 查询语句
52

53
        Returns:
54
            查询结果列表
55
        """
56
        if not sql.strip().upper().startswith("SELECT"):
57
            return {"error": "仅支持 SELECT 查询"}
58

59
        conn = sqlite3.connect(self.db_path)
60
        conn.row_factory = sqlite3.Row
61

62
        try:
63
            cursor = conn.execute(sql)
64
            results = [dict(row) for row in cursor.fetchall()]
65
            return results
66
        except Exception as e:
67
            return {"error": str(e)}
68
        finally:
69
            conn.close()
70

71
    @tool
72
    def save_research(self, topic: str, query: str, result: str,
73
                     sources: List[str]) -> int:
74
        """
75
        保存研究记录。
76

77
        Args:
78
            topic: 研究主题
79
            query: 查询内容
80
            result: 研究结果
81
            sources: 信息来源列表
82

83
        Returns:
84
            记录 ID
85
        """
86
        conn = sqlite3.connect(self.db_path)
87
        cursor = conn.cursor()
88

89
        cursor.execute("""
90
            INSERT INTO research_records (topic, query, result, sources)
91
            VALUES (?, ?, ?, ?)
92
        """, (topic, query, result, json.dumps(sources)))
93

94
        record_id = cursor.lastrowid
95
        conn.commit()
96
        conn.close()
97

98
        return record_id
99

100
    @tool
101
    def get_research_history(self, topic: Optional[str] = None,
102
                            limit: int = 10) -> List[Dict]:
103
        """
104
        获取研究历史记录。
105

106
        Args:
107
            topic: 可选的主题筛选
108
            limit: 返回记录数量
109

110
        Returns:
111
            历史记录列表
112
        """
113
        conn = sqlite3.connect(self.db_path)
114
        conn.row_factory = sqlite3.Row
115

116
        if topic:
117
            cursor = conn.execute("""
118
                SELECT * FROM research_records
119
                WHERE topic LIKE ?
120
                ORDER BY created_at DESC
121
                LIMIT ?
122
            """, (f"%{topic}%", limit))
123
        else:
124
            cursor = conn.execute("""
125
                SELECT * FROM research_records
126
                ORDER BY created_at DESC
127
                LIMIT ?
128
            """, (limit,))
129

130
        results = [dict(row) for row in cursor.fetchall()]
131
        conn.close()
132

133
        return results
134

135
    @tool
136
    def save_preference(self, user_id: str, key: str, value: str) -> bool:
137
        """保存用户偏好"""
138
        conn = sqlite3.connect(self.db_path)
139
        cursor = conn.cursor()
140

141
        cursor.execute("""
142
            INSERT OR REPLACE INTO user_preferences
143
            (user_id, preference_key, preference_value)
144
            VALUES (?, ?, ?)
145
        """, (user_id, key, value))
146

147
        conn.commit()
148
        conn.close()
149

150
        return True
151

152
    @tool
153
    def get_preferences(self, user_id: str) -> Dict[str, str]:
154
        """获取用户偏好"""
155
        conn = sqlite3.connect(self.db_path)
156
        conn.row_factory = sqlite3.Row
157

158
        cursor = conn.execute("""
159
            SELECT preference_key, preference_value
160
            FROM user_preferences
161
            WHERE user_id = ?
162
        """, (user_id,))
163

164
        preferences = {row["preference_key"]: row["preference_value"]
165
                      for row in cursor.fetchall()}
166
        conn.close()
167

168
        return preferences

1
from typing import List, Dict, Optional
2
from langchain_core.tools import tool
3
import os
4
import json
5
from datetime import datetime
6

7
class FileTools:
8
    """文件处理工具集"""
9

10
    def __init__(self, output_dir: str = "./output"):
11
        self.output_dir = output_dir
12
        os.makedirs(output_dir, exist_ok=True)
13

14
    @tool
15
    def save_report(self, filename: str, content: str,
16
                   format: str = "markdown") -> str:
17
        """
18
        保存研究报告。
19

20
        Args:
21
            filename: 文件名（不含扩展名）
22
            content: 报告内容
23
            format: 输出格式 (markdown/json/txt)
24

25
        Returns:
26
            保存的文件路径
27
        """
28
        ext_map = {
29
            "markdown": ".md",
30
            "json": ".json",
31
            "txt": ".txt"
32
        }
33

34
        ext = ext_map.get(format, ".txt")
35
        filepath = os.path.join(self.output_dir, f"{filename}{ext}")
36

37
        with open(filepath, 'w', encoding='utf-8') as f:
38
            f.write(content)
39

40
        return filepath
41

42
    @tool
43
    def read_file(self, filepath: str) -> str:
44
        """
45
        读取文件内容。
46

47
        Args:
48
            filepath: 文件路径
49

50
        Returns:
51
            文件内容
52
        """
53
        if not os.path.exists(filepath):
54
            return f"错误：文件不存在 {filepath}"
55

56
        with open(filepath, 'r', encoding='utf-8') as f:
57
            return f.read()
58

59
    @tool
60
    def list_files(self, directory: str = None,
61
                  pattern: str = None) -> List[str]:
62
        """
63
        列出目录下的文件。
64

65
        Args:
66
            directory: 目录路径，默认为输出目录
67
            pattern: 文件名模式（支持通配符）
68

69
        Returns:
70
            文件列表
71
        """
72
        dir_path = directory or self.output_dir
73

74
        if not os.path.exists(dir_path):
75
            return []
76

77
        files = []
78
        for f in os.listdir(dir_path):
79
            if pattern:
80
                import fnmatch
81
                if fnmatch.fnmatch(f, pattern):
82
                    files.append(os.path.join(dir_path, f))
83
            else:
84
                files.append(os.path.join(dir_path, f))
85

86
        return sorted(files, key=lambda x: os.path.getmtime(x), reverse=True)
87

88
    @tool
89
    def generate_markdown_report(self, title: str, sections: Dict[str, str],
90
                                 metadata: Dict = None) -> str:
91
        """
92
        生成 Markdown 格式报告。
93

94
        Args:
95
            title: 报告标题
96
            sections: 各章节内容 {章节名: 内容}
97
            metadata: 元数据（作者、日期等）
98

99
        Returns:
100
            Markdown 格式的报告内容
101
        """
102
        lines = []
103

104
        # 标题
105
        lines.append(f"# {title}\n")
106

107
        # 元数据
108
        if metadata:
109
            lines.append("---")
110
            for key, value in metadata.items():
111
                lines.append(f"{key}: {value}")
112
            lines.append("---\n")
113

114
        # 目录
115
        lines.append("## 目录\n")
116
        for i, section_name in enumerate(sections.keys(), 1):
117
            anchor = section_name.lower().replace(" ", "-")
118
            lines.append(f"{i}. [{section_name}](#{anchor})")
119
        lines.append("\n")
120

121
        # 内容
122
        for section_name, content in sections.items():
123
            lines.append(f"## {section_name}\n")
124
            lines.append(content)
125
            lines.append("\n")
126

127
        # 时间戳
128
        lines.append(f"\n---\n*生成时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}*")
129

130
        return "\n".join(lines)

1
from typing import List, Dict, Union
2
from langchain_core.tools import tool
3
import statistics
4

5
class CalculatorTools:
6
    """计算工具集"""
7

8
    @tool
9
    def calculate(self, expression: str) -> Union[float, str]:
10
        """
11
        执行数学计算。
12

13
        Args:
14
            expression: 数学表达式，如 "2 + 3 * 4"
15

16
        Returns:
17
            计算结果
18
        """
19
        try:
20
            # 安全计算（仅允许数学运算）
21
            allowed_chars = set('0123456789+-*/.() ')
22
            if not all(c in allowed_chars for c in expression):
23
                return "错误：表达式包含非法字符"
24

25
            result = eval(expression)
26
            return result
27
        except Exception as e:
28
            return f"计算错误：{str(e)}"
29

30
    @tool
31
    def analyze_statistics(self, numbers: List[float]) -> Dict:
32
        """
33
        计算统计指标。
34

35
        Args:
36
            numbers: 数值列表
37

38
        Returns:
39
            统计结果字典
40
        """
41
        if not numbers:
42
            return {"error": "数据为空"}
43

44
        return {
45
            "count": len(numbers),
46
            "sum": sum(numbers),
47
            "mean": statistics.mean(numbers),
48
            "median": statistics.median(numbers),
49
            "min": min(numbers),
50
            "max": max(numbers),
51
            "std_dev": statistics.stdev(numbers) if len(numbers) > 1 else 0,
52
            "variance": statistics.variance(numbers) if len(numbers) > 1 else 0
53
        }
54

55
    @tool
56
    def calculate_growth_rate(self, start_value: float,
57
                             end_value: float,
58
                             periods: int = 1) -> Dict:
59
        """
60
        计算增长率。
61

62
        Args:
63
            start_value: 起始值
64
            end_value: 结束值
65
            periods: 周期数
66

67
        Returns:
68
            增长率信息
69
        """
70
        total_growth = (end_value - start_value) / start_value * 100
71
        avg_growth = total_growth / periods
72
        cagr = ((end_value / start_value) ** (1 / periods) - 1) * 100
73

74
        return {
75
            "total_growth_rate": f"{total_growth:.2f}%",
76
            "average_growth_rate": f"{avg_growth:.2f}%",
77
            "cagr": f"{cagr:.2f}%"
78
        }
79

80
    @tool
81
    def compare_values(self, values: Dict[str, float]) -> Dict:
82
        """
83
        对比多个数值。
84

85
        Args:
86
            values: {名称: 数值} 的字典
87

88
        Returns:
89
            对比结果
90
        """
91
        sorted_values = sorted(values.items(), key=lambda x: x[1], reverse=True)
92
        max_name, max_val = sorted_values[0]
93
        min_name, min_val = sorted_values[-1]
94

95
        return {
96
            "ranking": [{"name": k, "value": v} for k, v in sorted_values],
97
            "max": {"name": max_name, "value": max_val},
98
            "min": {"name": min_name, "value": min_val},
99
            "range": max_val - min_val
100
        }

1
from typing import List, Dict
2
from collections import deque
3

4
class ShortTermMemory:
5
    """短期记忆：对话历史管理"""
6

7
    def __init__(self, max_messages: int = 50, max_tokens: int = 8000):
8
        self.max_messages = max_messages
9
        self.max_tokens = max_tokens
10
        self.messages: deque = deque(maxlen=max_messages)
11
        self.summary = ""
12

13
    def add_message(self, role: str, content: str):
14
        """添加消息"""
15
        self.messages.append({
16
            "role": role,
17
            "content": content,
18
            "token_count": self._estimate_tokens(content)
19
        })
20

21
        # 检查是否需要压缩
22
        if self._get_total_tokens() > self.max_tokens:
23
            self._compress()
24

25
    def get_context(self, include_summary: bool = True) -> List[Dict]:
26
        """获取上下文"""
27
        context = []
28

29
        if include_summary and self.summary:
30
            context.append({
31
                "role": "system",
32
                "content": f"[历史摘要] {self.summary}"
33
            })
34

35
        context.extend(list(self.messages))
36
        return context
37

38
    def clear(self):
39
        """清空记忆"""
40
        self.messages.clear()
41
        self.summary = ""
42

43
    def _estimate_tokens(self, text: str) -> int:
44
        """估算 token 数量"""
45
        # 简单估算：中文约 1.5 字/token，英文约 4 字符/token
46
        return len(text) // 2
47

48
    def _get_total_tokens(self) -> int:
49
        """获取总 token 数"""
50
        return sum(m["token_count"] for m in self.messages)
51

52
    def _compress(self):
53
        """压缩历史消息"""
54
        # 保留最近 10 条，其余生成摘要
55
        if len(self.messages) <= 10:
56
            return
57

58
        to_compress = list(self.messages)[:-10]
59

60
        # 这里可以用 LLM 生成摘要
61
        # 简化处理：直接拼接关键信息
62
        compressed_summary = " | ".join([
63
            f"{m['role']}: {m['content'][:100]}..."
64
            for m in to_compress[-5:]
65
        ])
66

67
        self.summary = f"{self.summary}\n{compressed_summary}" if self.summary else compressed_summary
68

69
        # 只保留最近的消息
70
        self.messages = deque(list(self.messages)[-10:], maxlen=self.max_messages)

1
from typing import List, Dict, Optional
2
import chromadb
3
from chromadb.config import Settings
4
from openai import OpenAI
5
import os
6

7
class LongTermMemory:
8
    """长期记忆：向量数据库存储"""
9

10
    def __init__(self, collection_name: str = "research_memory",
11
                 persist_dir: str = "./data/chroma"):
12
        # 初始化 ChromaDB
13
        self.client = chromadb.Client(Settings(
14
            chroma_db_impl="duckdb+parquet",
15
            persist_directory=persist_dir
16
        ))
17

18
        self.collection = self.client.get_or_create_collection(
19
            name=collection_name,
20
            metadata={"hnsw:space": "cosine"}
21
        )
22

23
        self.embedder = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
24

25
    def store(self, content: str, metadata: Dict = None) -> str:
26
        """存储记忆"""
27
        import uuid
28

29
        memory_id = str(uuid.uuid4())
30
        embedding = self._get_embedding(content)
31

32
        self.collection.add(
33
            ids=[memory_id],
34
            embeddings=[embedding],
35
            documents=[content],
36
            metadatas=[metadata or {}]
37
        )
38

39
        return memory_id
40

41
    def recall(self, query: str, n_results: int = 5,
42
               filter_metadata: Dict = None) -> List[Dict]:
43
        """检索相关记忆"""
44
        query_embedding = self._get_embedding(query)
45

46
        results = self.collection.query(
47
            query_embeddings=[query_embedding],
48
            n_results=n_results,
49
            where=filter_metadata,
50
            include=["documents", "metadatas", "distances"]
51
        )
52

53
        memories = []
54
        for i in range(len(results["ids"][0])):
55
            memories.append({
56
                "id": results["ids"][0][i],
57
                "content": results["documents"][0][i],
58
                "metadata": results["metadatas"][0][i],
59
                "distance": results["distances"][0][i]
60
            })
61

62
        return memories
63

64
    def _get_embedding(self, text: str) -> List[float]:
65
        """生成 embedding"""
66
        response = self.embedder.embeddings.create(
67
            model="text-embedding-3-small",
68
            input=text
69
        )
70
        return response.data[0].embedding
71

72
    def store_research_experience(self, topic: str, query: str,
73
                                  result: str, lessons: str = None):
74
        """存储研究经验"""
75
        content = f"""
76
研究主题：{topic}
77
查询内容：{query}
78
研究结果：{result}
79
经验总结：{lessons or '无'}
80
"""
81
        self.store(
82
            content,
83
            metadata={
84
                "type": "research_experience",
85
                "topic": topic,
86
                "timestamp": datetime.now().isoformat()
87
            }
88
        )
89

90
    def get_similar_research(self, topic: str, limit: int = 3) -> List[Dict]:
91
        """获取相似研究"""
92
        return self.recall(
93
            topic,
94
            n_results=limit,
95
            filter_metadata={"type": "research_experience"}
96
        )

1
from abc import ABC, abstractmethod
2
from typing import List, Dict, Any, Optional
3
from langchain_openai import ChatOpenAI
4
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage
5
from langchain_core.tools import BaseTool
6

7
class BaseAgent(ABC):
8
    """Agent 基类"""
9

10
    def __init__(self,
11
                 name: str,
12
                 llm: ChatOpenAI,
13
                 tools: List[BaseTool] = None,
14
                 system_prompt: str = None):
15

16
        self.name = name
17
        self.llm = llm
18
        self.tools = tools or []
19
        self.system_prompt = system_prompt or self._default_system_prompt()
20

21
        # 绑定工具
22
        if self.tools:
23
            self.llm_with_tools = self.llm.bind_tools(self.tools)
24
        else:
25
            self.llm_with_tools = self.llm
26

27
    @abstractmethod
28
    def _default_system_prompt(self) -> str:
29
        """默认系统提示词"""
30
        pass
31

32
    @abstractmethod
33
    def process(self, input_data: Any) -> Any:
34
        """处理任务"""
35
        pass
36

37
    def _call_llm(self, messages: List) -> str:
38
        """调用 LLM"""
39
        full_messages = [SystemMessage(content=self.system_prompt)] + messages
40
        response = self.llm_with_tools.invoke(full_messages)
41
        return response
42

43
    def _execute_tools(self, tool_calls: List) -> List[Dict]:
44
        """执行工具调用"""
45
        results = []
46

47
        for tool_call in tool_calls:
48
            tool_name = tool_call["name"]
49
            tool_args = tool_call["args"]
50

51
            # 查找工具
52
            tool = next((t for t in self.tools if t.name == tool_name), None)
53

54
            if tool:
55
                try:
56
                    result = tool.invoke(tool_args)
57
                    results.append({
58
                        "tool": tool_name,
59
                        "result": result,
60
                        "success": True
61
                    })
62
                except Exception as e:
63
                    results.append({
64
                        "tool": tool_name,
65
                        "error": str(e),
66
                        "success": False
67
                    })
68

69
        return results

1
from typing import List, Dict, Any
2
from langchain_openai import ChatOpenAI
3
from langchain_core.tools import BaseTool
4
from .base import BaseAgent
5

6
class ResearcherAgent(BaseAgent):
7
    """研究员 Agent：负责信息收集"""
8

9
    def __init__(self, llm: ChatOpenAI, tools: List[BaseTool]):
10
        super().__init__(
11
            name="Researcher",
12
            llm=llm,
13
            tools=tools
14
        )
15

16
    def _default_system_prompt(self) -> str:
17
        return """你是一位专业的研究员，负责收集和整理信息。
18

19
你的职责：
20
1. 根据研究主题，确定需要搜集的信息类型
21
2. 使用搜索工具获取相关信息
22
3. 验证信息的可靠性
23
4. 整理和归类收集到的信息
24

25
工作原则：
26
- 信息来源要可靠
27
- 尽可能全面
28
- 标注信息来源
29
- 发现矛盾时要标注"""
30

31
    def process(self, topic: str, depth: str = "normal") -> Dict:
32
        """
33
        执行研究任务。
34

35
        Args:
36
            topic: 研究主题
37
            depth: 研究深度 (quick/normal/deep)
38

39
        Returns:
40
            研究结果
41
        """
42
        # 第一步：规划研究
43
        plan = self._plan_research(topic, depth)
44

45
        # 第二步：执行搜索
46
        collected_info = []
47
        for query in plan["queries"]:
48
            search_result = self._search(query)
49
            collected_info.append(search_result)
50

51
        # 第三步：整理结果
52
        organized = self._organize_info(collected_info)
53

54
        return {
55
            "topic": topic,
56
            "plan": plan,
57
            "collected_info": collected_info,
58
            "organized_info": organized,
59
            "sources": plan["sources"]
60
        }
61

62
    def _plan_research(self, topic: str, depth: str) -> Dict:
63
        """规划研究步骤"""
64
        prompt = f"""
65
研究主题：{topic}
66
研究深度：{depth}
67

68
请规划研究步骤：
69
1. 需要搜索哪些方面的信息？
70
2. 使用什么关键词搜索？
71
3. 预期获取什么类型的数据？
72

73
以 JSON 格式返回：
74
{{
75
    "aspects": ["方面1", "方面2", ...],
76
    "queries": ["搜索词1", "搜索词2", ...],
77
    "expected_data": ["数据类型1", ...]
78
}}
79
"""
80
        response = self.llm.invoke([HumanMessage(content=prompt)])
81
        # 解析 JSON...
82
        return self._parse_plan(response.content)
83

84
    def _search(self, query: str) -> Dict:
85
        """执行搜索"""
86
        messages = [
87
            HumanMessage(content=f"搜索：{query}")
88
        ]
89

90
        response = self._call_llm(messages)
91

92
        # 处理工具调用
93
        if hasattr(response, 'tool_calls') and response.tool_calls:
94
            tool_results = self._execute_tools(response.tool_calls)
95
            return {
96
                "query": query,
97
                "results": tool_results
98
            }
99

100
        return {"query": query, "results": []}
101

102
    def _organize_info(self, collected_info: List) -> Dict:
103
        """整理收集的信息"""
104
        prompt = f"""
105
请整理以下收集到的信息：
106

107
{collected_info}
108

109
要求：
110
1. 按主题分类
111
2. 去除重复信息
112
3. 标注来源
113
4. 突出关键信息
114
"""
115
        response = self.llm.invoke([HumanMessage(content=prompt)])
116
        return {"summary": response.content}

1
from typing import List, Dict, Any
2
from langchain_openai import ChatOpenAI
3
from langchain_core.tools import BaseTool
4
from .base import BaseAgent
5

6
class AnalystAgent(BaseAgent):
7
    """分析师 Agent：负责数据分析"""
8

9
    def __init__(self, llm: ChatOpenAI, tools: List[BaseTool]):
10
        super().__init__(
11
            name="Analyst",
12
            llm=llm,
13
            tools=tools
14
        )
15

16
    def _default_system_prompt(self) -> str:
17
        return """你是一位专业的数据分析师，负责分析和解读数据。
18

19
你的职责：
20
1. 分析研究数据
21
2. 发现趋势和模式
22
3. 计算关键指标
23
4. 得出分析结论
24

25
分析原则：
26
- 数据驱动
27
- 结论有依据
28
- 考虑多种可能性
29
- 识别数据局限性"""
30

31
    def process(self, research_data: Dict, analysis_type: str = "comprehensive") -> Dict:
32
        """
33
        执行分析任务。
34

35
        Args:
36
            research_data: 研究数据
37
            analysis_type: 分析类型 (quick/comprehensive/deep)
38

39
        Returns:
40
            分析结果
41
        """
42
        # 第一步：理解数据
43
        understanding = self._understand_data(research_data)
44

45
        # 第二步：执行分析
46
        if analysis_type == "quick":
47
            analysis = self._quick_analysis(research_data)
48
        else:
49
            analysis = self._comprehensive_analysis(research_data)
50

51
        # 第三步：生成洞察
52
        insights = self._generate_insights(analysis)
53

54
        return {
55
            "understanding": understanding,
56
            "analysis": analysis,
57
            "insights": insights
58
        }
59

60
    def _quick_analysis(self, data: Dict) -> Dict:
61
        """快速分析"""
62
        prompt = f"""
63
对以下数据进行快速分析，提取关键信息：
64

65
{data}
66

67
请提供：
68
1. 核心发现（3-5 点）
69
2. 关键数据
70
3. 初步结论
71
"""
72
        response = self.llm.invoke([HumanMessage(content=prompt)])
73
        return {"analysis": response.content}
74

75
    def _comprehensive_analysis(self, data: Dict) -> Dict:
76
        """深度分析"""
77
        # 使用工具进行计算
78
        # 生成趋势分析
79
        # 对比分析
80
        pass
81

82
    def _generate_insights(self, analysis: Dict) -> List[str]:
83
        """生成洞察"""
84
        prompt = f"""
85
基于以下分析，生成关键洞察：
86

87
{analysis}
88

89
要求：
90
1. 突出最重要的发现
91
2. 解释发现的意义
92
3. 提出可能的行动建议
93
"""
94
        response = self.llm.invoke([HumanMessage(content=prompt)])
95
        return response.content.split('\n')