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
| import json
from datetime import datetime, timedelta
from openai import OpenAI
client = OpenAI(api_key="sk-xxx")
schedule_db = []
tools = [
{
"type": "function",
"function": {
"name": "add_event",
"description": "添加日程事件",
"parameters": {
"type": "object",
"properties": {
"title": {"type": "string", "description": "事件名称"},
"date": {"type": "string", "description": "日期，格式 YYYY-MM-DD"},
"time": {"type": "string", "description": "时间，格式 HH:MM，可选"},
"duration_minutes": {"type": "integer", "description": "持续分钟数"}
},
"required": ["title", "date"]
}
}
},
{
"type": "function",
"function": {
"name": "query_schedule",
"description": "查询指定日期的日程",
"parameters": {
"type": "object",
"properties": {
"date": {"type": "string", "description": "日期，YYYY-MM-DD 格式"}
},
"required": ["date"]
}
}
},
{
"type": "function",
"function": {
"name": "delete_event",
"description": "删除日程",
"parameters": {
"type": "object",
"properties": {
"title": {"type": "string", "description": "要删除的事件名称（支持模糊匹配）"}
},
"required": ["title"]
}
}
}
]
def add_event(title, date, time=None, duration_minutes=60):
event = {"title": title, "date": date, "time": time or "全天",
"duration": duration_minutes}
schedule_db.append(event)
return json.dumps({"status": "ok", "event": event})
def query_schedule(date):
events = [e for e in schedule_db if e["date"] == date]
if not events:
return json.dumps({"date": date, "events": [], "message": "当天没有日程"})
return json.dumps({"date": date, "events": events})
def delete_event(title):
global schedule_db
before = len(schedule_db)
schedule_db = [e for e in schedule_db if title not in e["title"]]
deleted = before - len(schedule_db)
return json.dumps({"status": "ok", "deleted": deleted})
available_functions = {
"add_event": add_event,
"query_schedule": query_schedule,
"delete_event": delete_event,
}
SYSTEM_PROMPT = """你是一个日程管理助手。根据用户描述，调用合适的函数管理日程。
- 如果用户说"明天下午"，请计算出具体日期和时间
- 操作完成后简洁地告知用户结果
- 当前日期是 {}""".format(datetime.now().strftime("%Y-%m-%d"))
def chat(user_input, max_turns=6):
messages = [
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_input}
]
for turn in range(max_turns):
response = client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=tools,
tool_choice="auto"
)
message = response.choices[0].message
if not message.tool_calls:
return message.content
messages.append(message)
for tc in message.tool_calls:
func = available_functions.get(tc.function.name)
if not func:
result = json.dumps({"error": f"未知工具: {tc.function.name}"})
else:
try:
args = json.loads(tc.function.arguments)
result = func(**args)
except Exception as err:
result = json.dumps({"error": f"工具执行异常: {str(err)}"})
messages.append({
"role": "tool",
"tool_call_id": tc.id,
"content": result
})
return "工具调用轮次超过上限，已触发熔断保护。"
print(chat("下周一上午9点开会，讨论项目进度，预计1.5小时"))
print(chat("下周一下午3点去牙科复诊"))
print(chat("告诉我下周一的日程"))
|
