-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathtrip_advisor.py
358 lines (331 loc) · 17.1 KB
/
trip_advisor.py
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
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# File : trip_advisor.py
# Author : Yan <[email protected]>
# Date : 01.03.2024
# Last Modified Date: 18.03.2024
# Last Modified By : Yan <[email protected]>
from collections import namedtuple
from http import HTTPStatus
import json
import os
import re
import logging
import requests
import dashscope
import openxlab
logger = logging.getLogger(__name__)
Prompt = namedtuple('Prompt', ['name', 'instruction', 'examples'])
TRIP_ADVISE_PROMPT = Prompt(
name='trip_advise',
instruction='你具有丰富的旅行经验，了解各地著名或小众旅游景点，擅长规划旅游行程。你可以帮助我规划旅行行程、提供各种类型的旅游目的地建议，包括著名景点和小众独特的去处，并且可以根据我的偏好、预算、时间以及特定的兴趣点（如历史文化探索、自然风光欣赏、美食之旅等）来定制旅行方案。制定旅行攻略应遵守的原则包括：每天至少安排1个景点，至多安排5个景点；每个景点只游览一次，前一天游览过的景点不要在后续行程中；每天游览景点的顺序要考虑景点特点，例如，适合看日出的景点要安排在早上，适合徒步游览的景点尽量安排在上午，适合看日落的景点要安排在傍晚，适合看夜景的景点应当安排在晚上；安排景点时要根据当天的天气状况，选择适合晴朗天气游览的景点和适合雨雪天气游览的景点；安排在相邻时间段游览的景点之间的距离不要太远；如果天气未知，安排行程时不必考虑天气情况。你的任务是根据旅游天数、目的地和天气等出行信息制定旅游攻略，用合法的JSON格式返回结果，不要添加注释。',
examples=[
{
'city': '绍兴',
'duration': '2天',
'weathers': [{'day_weather': '晴', 'night_weather': '多云'}, {'day_weather': '阴转小雨', 'night_weather': '晴'}],
'trip_advise': {
'city': '绍兴',
'days': [
{
'date': '第1天',
'day_weather': '晴',
'night_weather': '多云',
'schedule': [
{
'time': '上午',
'location': '鲁迅故里',
'description': '游览鲁迅故居、鲁迅纪念馆、百草园和三味书屋。这些景点较为集中，适合一早参观，了解历史文化，并且在清晨时分体验江南水乡的宁静。'
},
{
'time': '中午',
'location': '咸亨酒店',
'description': '午餐可以在鲁迅故居附近的咸亨酒店享用，品尝绍兴地方特色菜肴，如臭豆腐、茴香豆和绍兴黄酒等。'
},
{
'time': '下午',
'location': '东湖景区',
'description': '午后阳光明媚，适宜在户外活动，东湖是理想的游览地点，可以乘乌篷船欣赏湖光山色。'
},
{
'time': '傍晚',
'location': '会稽山风景度假区',
'description': '若时间允许，可在傍晚前前往会稽山，虽然当天不适合看日落，但可以在多云的傍晚感受山间静谧氛围。'
},
{
'time': '晚上',
'location': '安昌古镇',
'description': '晚上入住安昌古镇内客栈，感受古镇夜景，体验当地生活气息，并为第二天早上游览古镇预留充足时间。'
}
]
},
{
'date': '第2天',
'day_weather': '阴转小雨',
'night_weather': '晴',
'schedule': [
{
'time': '上午',
'location': '安昌古镇',
'description': '清晨游览古镇，欣赏雨后朦胧的江南水乡风貌，享受当地特色的早餐。'
},
{
'time': '中午',
'location': '兰亭景区',
'description': '由于预计白天有小雨，选择室内或半开放型的兰亭景区是个不错的选择，可以参观王羲之书法文化，避雨同时沉浸于艺术气息中。'
},
{
'time': '下午',
'location': '大禹陵',
'description': '如果雨势不大，可以考虑游览大禹陵，雨中的自然景观别有一番风味，带上雨具爬山或在景区内漫步，呼吸清新空气。'
},
{
'time': '傍晚',
'location': '书圣故里',
'description': '傍晚时分，若天气好转，可以去书圣故里，观赏文笔塔并登高远眺，体验绍兴城市风光。'
},
{
'time': '晚上',
'location': '绍兴古城',
'description': '鉴于晚上天气预报转晴，可以选择夜游绍兴古城，逛逛历史街区，体验古城墙、古桥及河两岸的夜景灯光秀。'
}
]
}
]
}
},
{
'city': '北京',
'duration': '2天',
'weathers': [{'day_weather': '大雨', 'night_weather': '多云'}, {'day_weather': '小雨', 'night_weather': '晴'}],
'trip_advise': {
'city': '北京',
'days': [
{
'date': '第1天',
'day_weather': '大雨',
'night_weather': '多云',
'schedule': [
{
'time': '上午',
'location': '国家博物馆',
'description': '鉴于全天大部分时间有雨，首日上午安排参观中国国家博物馆，这里丰富的馆藏和室内环境适合在雨天游览。'
},
{
'time': '中午',
'location': '王府井步行街',
'description': '在附近著名的王府井步行街享用午餐，并可在此购买一些北京特色小吃或纪念品。'
},
{
'time': '下午',
'location': '故宫博物院',
'description': '尽管当天有雨，但故宫内部游览不受影响。由于下雨可能减少室外游客数量，此时游览故宫可以避开部分人流，体验更佳。请提前通过网络预订门票，避免现场排队。'
},
{
'time': '傍晚',
'location': '南锣鼓巷',
'description': '若傍晚时分雨势减弱至多云，可以选择前往南锣鼓巷或后海地区，体验老北京胡同文化，同时可以在酒吧、茶馆或餐厅中稍作休息，等待夜幕降临。'
}
]
},
{
'date': '第2天',
'day_weather': '小雨',
'night_weather': '晴',
'schedule': [
{
'time': '上午',
'location': '798艺术区',
'description': '上午安排去798艺术区，这里的室内艺术展览和创意店铺可以提供充足的避雨空间，且小雨天气下的艺术区别有一番韵味。'
},
{
'time': '中午',
'location': '三里屯商圈',
'description': '前往三里屯商圈，在那里找一家餐厅享用午餐，同时享受现代都市的繁华氛围。'
},
{
'time': '下午',
'location': '颐和园',
'description': '根据天气预报，下午可能会有小雨，可以选择在颐和园内乘坐游船观赏昆明湖及佛香阁等主要景观，即便下雨也能在长廊等遮蔽处欣赏园林美景。'
},
{
'time': '晚上',
'location': '奥林匹克公园',
'description': '考虑到晚上的天气会转晴，可在傍晚时分前往奥林匹克公园，参观鸟巢和水立方的夜景。如果时间允许，还可以在晴朗的夜晚观赏一场灯光秀表演。'
}
]
}
]
}
}
]
)
class TripAdvisor(object):
def get_trip_brief(self, trip):
city = '目的地:' + trip['city']
duration = '\n旅游天数:' + trip['duration']
weather = '\n天气情况:'
for i, w in enumerate(trip['weathers']):
weather += f'第{i+1}天'
day_weather, night_weather = w['day_weather'], w['night_weather']
weather += f'白天{day_weather}, 晚上{night_weather}。'
return city + duration + weather
def create_prompt(self, trip):
prompt = TRIP_ADVISE_PROMPT.instruction
if TRIP_ADVISE_PROMPT.examples:
prompt += '\n以下是根据出行信息制定旅游攻略的示例。'
for i, example in enumerate(TRIP_ADVISE_PROMPT.examples):
prompt += f'\n示例{i+1}:'
example_brief = self.get_trip_brief(example)
example_advise = json.dumps(
example['trip_advise'], ensure_ascii=False).encode('utf8').decode()
prompt += f'\n出行信息如下:\n{example_brief}\n旅游攻略如下:\n{example_advise}'
prompt += '\n请你根据以下出行信息制定旅游攻略:'
prompt += '\n' + self.get_trip_brief(trip)
return prompt
class QwenTripAdvisor(TripAdvisor):
def __init__(self, model_name):
self.model_name = model_name # e.g. qwen-max, qwen-max-longcontext
def generate_advise(self, trip):
advise = {}
if not trip:
logger.warning('No trip brief provided to generate advise.')
return advise
prompt = self.create_prompt(trip)
try:
response = dashscope.Generation.call(
model=self.model_name,
prompt=prompt
)
if response.status_code == HTTPStatus.OK:
logger.info(
'Qwen output: {}, usage info: {}'.format(
response.output, response.usage)
)
text = response.output['text']
# Sometimes the text not only contains valid JSON but also
# contains some contents like ```json {} ```. Stupid LLM.
m = re.search(r'```json(.*)```', text, re.DOTALL)
if m:
text = m.group(1)
advise = json.loads(text)
else:
logger.error(
'Qwen request failed. Request id: {}, status code: {},'
' error code: {}, error message: {}'.format(
response.request_id, response.status_code,
response.code, response.message)
)
except Exception as e:
logger.error('Qwen generation failed: {}'.format(e))
return advise
class InternTripAdvisor(TripAdvisor):
def __init__(self, model_name, model_url, temperature=0.95, top_p=0.9):
self.access_key = os.environ['OPENXLAB_AK']
self.secret_key = os.environ['OPENXLAB_SK']
self.model_url = model_url
self.model_name = model_name
self.temperature = temperature
self.top_p = top_p
openxlab.login(ak=self.access_key, sk=self.secret_key)
def _get_token(self):
return openxlab.xlab.handler.user_token.get_jwt(self.access_key,
self.secret_key)
def generate_advise(self, trip):
advise = {}
if not trip:
logger.warning('No trip brief provided to generate advise.')
return advise
prompt = self.create_prompt(trip)
headers = {
'Authorization': self._get_token(),
'Content-Type': 'application/json'
}
payload = {
'model': self.model_name,
'messages': [{'role': 'user', 'text': prompt}],
'temperature': self.temperature,
'top_p': self.top_p
}
try:
response = requests.post(self.model_url, headers=headers,
data=json.dumps(payload))
content = response.json()
if response.status_code == HTTPStatus.OK:
logger.info('InternLM output: {}'.format(content))
text = content['data']['choices'][0]['text']
m = re.search(r'{.+}', text, re.DOTALL)
if m:
text = m.group(0)
advise = json.loads(text)
else:
logger.error(
'InternLM request failed. Status code: {},'
' code: {}, message: {}, error: {}'.format(
response.status_code, content['code'], content['msg'],
content['error'])
)
except Exception as e:
logger.error(f'InternLM generation failed: {e}')
return advise
class YiTripAdvisor(TripAdvisor):
def __init__(self, auth_url, model_url, temperature=0.9, top_p=0.8,
penalty_score=2.0):
self.auth_url = auth_url
self.model_url = model_url
self.temperature = temperature
self.top_p = top_p
self.penalty_score = penalty_score
self.api_key = os.environ['BAIDU_API_KEY']
self.secret_key = os.environ['BAIDU_SK']
def _get_token(self):
headers = {
'Content-Type': 'application/json',
'Accept': 'application/json'
}
payload = {
'grant_type': 'client_credentials',
'client_id': self.api_key,
'client_secret': self.secret_key
}
try:
response = requests.post(self.auth_url, headers=headers, params=payload)
except Exception as e:
logger.error('Get access token failed: {}'.format(e))
return response.json().get("access_token")
def generate_advise(self, trip):
advise = {}
if not trip:
logger.warning('No trip brief provided to generate advise.')
return advise
prompt = self.create_prompt(trip)
headers = {'Content-Type': 'application/json'}
params = {'access_token': self._get_token()}
data = json.dumps({
"messages": [
{
"role": "user",
"content": prompt
}
]
})
advise = {}
try:
response = requests.post(
self.model_url, params=params, headers=headers, data=data
)
content = response.json()
if not content.get('error_code'):
logger.info('Yi output: {}'.format(content))
text = content['result']
m = re.search(r'{.+}', text, re.DOTALL)
if m:
text = m.group(0)
advise = json.loads(text)
else:
logger.error('Yi request failed. '
'Error code: {}, error message: {}'.format(
response['error_code'], response['error_msg']))
except Exception as e:
logger.error('Yi generation failed: {}'.format(e))
return advise