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
136
137
138
139
140
| class PersonalAssistantAgent:
"""个人助理Agent - 集成多种工具"""
def __init__(self, api_key: str):
self.registry = ToolRegistry()
self.engine = ToolExecutionEngine(self.registry)
self.tool_caller = ToolUsingAgent(self.engine)
self.resilient_caller = ResilientToolCaller(self.engine)
self._register_all_tools()
self.llm = ChatOpenAI(model="gpt-4-turbo", api_key=api_key)
def _register_all_tools(self):
"""注册所有工具"""
@self.registry.register(
name="create_event",
description="创建日历事件",
parameters={
"type": "object",
"properties": {
"title": {"type": "string", "description": "事件标题"},
"start_time": {"type": "string", "description": "开始时间"},
"end_time": {"type": "string", "description": "结束时间"},
"location": {"type": "string", "description": "地点"}
},
"required": ["title", "start_time"]
}
)
def create_calendar_event(title: str, start_time: str,
end_time: str = None, location: str = None) -> str:
return f"已创建日历事件: {title}\n时间: {start_time}" + \
(f" - {end_time}" if end_time else "")
@self.registry.register(
name="send_email",
description="发送电子邮件",
parameters={
"type": "object",
"properties": {
"to": {"type": "string", "description": "收件人邮箱"},
"subject": {"type": "string", "description": "邮件主题"},
"body": {"type": "string", "description": "邮件正文"}
},
"required": ["to", "subject", "body"]
}
)
async def send_email(to: str, subject: str, body: str) -> str:
return f"邮件已发送至: {to}\n主题: {subject}"
@self.registry.register(
name="web_search",
description="搜索网络信息",
parameters={
"type": "object",
"properties": {
"query": {"type": "string", "description": "搜索关键词"},
"limit": {"type": "number", "description": "结果数量", "default": 5}
},
"required": ["query"]
}
)
async def web_search(query: str, limit: int = 5) -> str:
return f"搜索 '{query}' 找到 {limit} 条结果"
@self.registry.register(
name="calculate",
description="执行数学计算",
parameters={
"type": "object",
"properties": {
"expression": {"type": "string", "description": "数学表达式"}
},
"required": ["expression"]
}
)
def calculate(expression: str) -> str:
try:
result = eval(expression)
return f"计算结果: {result}"
except Exception as e:
return f"计算错误: {str(e)}"
@self.registry.register(
name="read_file",
description="读取文件",
parameters={
"type": "object",
"properties": {
"path": {"type": "string", "description": "文件路径"},
"lines": {"type": "number", "description": "读取行数"}
},
"required": ["path"]
}
)
def read_file(path: str, lines: int = 100) -> str:
try:
with open(path, 'r') as f:
return f.read()[:lines * 100]
except FileNotFoundError:
return f"文件未找到: {path}"
async def chat(self, message: str) -> str:
"""处理用户消息"""
result = await self.tool_caller.process(message, self.llm)
if result["type"] == "tool_result":
tool_result = result["result"]
response = await self._generate_response(message, tool_result)
return response
return result["content"]
async def _generate_response(self, user_message: str,
tool_result: Dict) -> str:
"""生成基于工具结果的回复"""
prompt = f"""用户消息: {user_message}
工具执行结果:
{json.dumps(tool_result, ensure_ascii=False, indent=2)}
请用自然语言向用户解释工具执行结果。
"""
response = await self.llm.agenerate([{"role": "user", "content": prompt}])
return response.generations[0].text
|
