Hello_Agents项目学习笔记（1）实现一个智能旅行助手

来源于开源免费项目链接https://datawhalechina.github.io/hello-agents/#/./READMEPDF版可以在网站下载还有对应的题目和课程是一个不错的入手项目。以下是学习笔记仅供参考持续更新中第一次写博客欢迎大佬指正。实现一个智能旅行助手智能体第一步实现一个查询真实天气的工具。#这是一个查询真实天气的工具 import requests #第三方 HTTP 请求库发送/接收网络数据 #定义一个得到天气的方法传入城市名参数 def get_weather(city:str)-str: #通过调用wttr.in查询真实天气,网址https://wttr.in/城市名 urlfhttps://wttr.in/{city}?formatj1 #?是分隔符j1是返回JSON格式 try: #发起网络请求 response requests.get(url) #检查响应状态码是否为200成功,500是服务器挂了404是页面不存在403是没权限禁止访问301/302是网址跳转 response.raise_for_status() #解析返回的JSON数据 response 既不是字符串也不是 JSON也不是字典它是一个 Response 对象 这个对象里面装了服务器返回的所有信息可以用不同属性拆开 response.status_code → 状态码200/404 等 response.text → 字符串格式原始的 JSON 文本 response.json() → 把 JSON 转成 Python 字典下面会用到 response.headers → 响应头信息 data response.json() data的格式是这样的 { current_condition: [ { temp_C: 20, weatherDesc: [{value: 多云}] #还有很多其他的比如风速风向之类的此处省略 } ] } #提取当前天气情况 current_condition data[current_condition][0]#得到的是{temp_C: 20,.....字典} weather_Desc current_condition[weatherDesc][0][value] temp_c current_condition[temp_c]#得到的输出是20 #格式化成自然语言返回 return f{city}当前天气{weather_Desc},气温:{temp_c}摄氏度 except requests.exceptions.RequestException as e: #处理网络错误 return f错误查询天气时遇到网络问题 - {e} except (KeyError,IndexError) as e: #处理数据解析错误 return f错误解析天气数据失败可能是城市名称无效 - {e}第二步得到某城市的天气之后就可以创建一个根据城市和天气搜索并推荐旅游景点的工具了。#工具2搜索并推荐旅游景点 #pip install requests tavily-python openai -i https://pypi.tuna.tsinghua.edu.cn/simple import os from tavily import TavilyClient def get_attraction(city:str,weather:str) - str: api_key os.environ.get(TAVILY_API_KEY)#从系统环境变量中获取api秘钥 if not api_key: return 你滴环境变量没有配置成功哦 #初始化TavilyClient客户端 tavily TavilyClient(api_keyapi_key) #构造查询 query f{city}在weather天气下最值得去的旅游景点推荐及理由 try: # 带所有常用参数的搜索 response tavily.search( query北京今日天气, # 必选 search_depthbasic,#搜索深度basic默认快速搜索advanced深度搜索 max_results5,#最多返回几条搜索结果默认是5 include_answerTrue,#是否让ai直接总结答案 include_raw_contentFalse,#是否返回网页完整原文默认False只返回摘要开启后数据会很大 include_imagesFalse,#是否返回搜索到的图片链接 topicgeneral#指定搜索主题general通用搜索news只搜新闻finance财经sports体育 ) response tavily.search(queryquery,search_depthbasic, include_answerTrue )#让代码联网搜索内容并把搜索结果存到 response里,response是一个字典用字典的.get()来取值 if response.get(answer): return response[answer] #字典的常用键answer(AI直接总结好的答案) # query 你刚才搜索的关键词 # results 搜索到的网页结果列表多条信息 # images 搜索到的图片链接需开启参数 # raw_content 网页完整原文需开启参数 #如果没有综合回答就格式化原始结果 formatted_result [] for result in response.get(results,[]):#[]是默认值有结果返回结果没有就返回空列表 formatted_result.append(f- {result[title]}:{result[content]}) if not formatted_result:#空列表会返回False return 抱歉没有找到相关的旅游景点推荐 except Exception as e: return f错误执行Tavily搜索时出现问题 - {e}最后将这两个工具函数放入字典工具映射供主循环调用available_tools {get_weather:get_weather,get_attraction:get_attraction}下一步接入大语言模型from openai import OpenAI class OpenAiCompatibleClient: 一个用于调用任何兼容OpenAI接口的大语言模型服务的客户端 def __init__(self,model:str,api_key:str,base_url:str): self.model model self.client OpenAI(api_keyapi_key,base_urlbase_url) def generate(self,prompt:str,system_prompt:str) - str: print(正在调用大语言模型...) try: messages [{role:system,content:system_prompt}, {role:user,content:prompt}] response self.client.chat.completions.create( modelself.model, messagesmessages, streamFalse #一次性全部输出 ) answer response.choices[0].message.content #response长什么样 { # 1. 基础信息没用不用管 id: chatcmpl-xxxxxx, object: chat.completion, created: 1723456789, model: gpt-3.5-turbo, # 2. 核心 choices: [ { index: 0, message: { role: assistant, # 角色AI助手 content: 北京今天晴天气温25℃ # AI 回复的内容 }, finish_reason: stop } ], # 3. 消耗统计没用不用管 usage: { prompt_tokens: 10, completion_tokens: 8, total_tokens: 18 } } print(大语言模型响应成功) return answer except Exception as e: print(f调用大语言模型api时发生错误:{e}) return实例化这个类加入循环#实例化循环 import os import re API_KEY 你的key BASE_URL 你的URL MODEL_ID 你的模型ID TAVILY_API_KEY 你的key os.environ[TAVILY_API_KEY]你的key AGENT_SYSTEM_PROMPT 你是一个智能旅行助手。你的任务是分析用户的请求并使用可用工具一步步地解决问题。 # 可用工具: - get_weather(city: str): 查询指定城市的实时天气。 - get_attraction(city: str, weather: str): 根据城市和天气搜索推荐的旅游景点。 # 输出格式要求: 你的每次回复必须严格遵循以下格式包含一对Thought和Action Thought: [你的思考过程和下一步计划] Action: [你要执行的具体行动] Action的格式必须是以下之一 1. 调用工具function_name(arg_namearg_value) 2. 结束任务Finish[最终答案] # 重要提示: - 每次只输出一对Thought-Action - Action必须在同一行不要换行 - 当收集到足够信息可以回答用户问题时必须使用 Action: Finish[最终答案] 格式结束 请开始吧 llm OpenAiCompatibleClient(modelMODEL_ID,api_keyAPI_KEY,base_urlBASE_URL) user_prompt 你好请帮我查询一下今天北京的天气然后根据天气推荐一个合适的旅游景点。 prompt_history [f用户请求{user_prompt}] print(f用户请求:{user_prompt}) #开始循环 for i in range(5): print(f---循环{i1}---\n) #构建prompt full_prompt \n.join(prompt_history) #调用LLM思考 llm_output llm.generate(full_prompt,system_promptAGENT_SYSTEM_PROMPT) #模型可能会输出多余的Thought-Action,需要截断 match re.search(r(Thought:.*?Action:.*?)(?\n\s*(?:Thought:|Action:|Observation:)|\Z), llm_output, re.DOTALL) if match: truncated match.group(1).strip() if truncated ! llm_output.strip(): llm_output truncated print(已截断多余的 Thought-Action 对) print(f模型输出:\n{llm_output}\n) prompt_history.append(llm_output) # 3.3. 解析并执行行动 action_match re.search(rAction: (.*), llm_output, re.DOTALL) if not action_match: observation 错误: 未能解析到 Action 字段。请确保你的回复严格遵循 Thought: ... Action: ... 的格式。 observation_str fObservation: {observation} print(f{observation_str}\n * 40) prompt_history.append(observation_str) continue action_str action_match.group(1).strip() if action_str.startswith(Finish): final_answer re.match(rFinish\[(.*)\], action_str).group(1) print(f任务完成最终答案: {final_answer}) break tool_name re.search(r(\w)\(, action_str).group(1) args_str re.search(r\((.*)\), action_str).group(1) kwargs dict(re.findall(r(\w)([^]*), args_str)) if tool_name in available_tools: observation available_tools[tool_name](**kwargs) else: observation f错误:未定义的工具 {tool_name} # 3.4. 记录观察结果 observation_str fObservation: {observation} print(f{observation_str}\n * 40) prompt_history.append(observation_str)完整代码import requests #第三方 HTTP 请求库发送/接收网络数据 import os from tavily import TavilyClient from openai import OpenAI import os import re API_KEY 你的key BASE_URL 你的URL MODEL_ID 你的模型ID TAVILY_API_KEY 你的key os.environ[TAVILY_API_KEY]你的key #定义一个得到天气的方法传入城市名参数 def get_weather(city:str)-str: #通过调用wttr.in查询真实天气,网址https://wttr.in/城市名 urlfhttps://wttr.in/{city}?formatj1 #?是分隔符j1是返回JSON格式 try: #发起网络请求 response requests.get(url) #检查响应状态码是否为200成功,500是服务器挂了404是页面不存在403是没权限禁止访问301/302是网址跳转 response.raise_for_status() #解析返回的JSON数据 response 既不是字符串也不是 JSON也不是字典它是一个 Response 对象 这个对象里面装了服务器返回的所有信息可以用不同属性拆开 response.status_code → 状态码200/404 等 response.text → 字符串格式原始的 JSON 文本 response.json() → 把 JSON 转成 Python 字典下面会用到 response.headers → 响应头信息 data response.json() data的格式是这样的 { current_condition: [ { temp_C: 20, weatherDesc: [{value: 多云}] #还有很多其他的比如风速风向之类的此处省略 } ] } #提取当前天气情况 current_condition data[current_condition][0]#得到的是{temp_C: 20,.....字典} weather_Desc current_condition[weatherDesc][0][value] temp_c current_condition[temp_c]#得到的输出是20 #格式化成自然语言返回 return f{city}当前天气{weather_Desc},气温:{temp_c}摄氏度 except requests.exceptions.RequestException as e: #处理网络错误 return f错误查询天气时遇到网络问题 - {e} except (KeyError,IndexError) as e: #处理数据解析错误 return f错误解析天气数据失败可能是城市名称无效 - {e} def get_attraction(city: str, weather: str) - str: api_key os.environ.get(TAVILY_API_KEY) # 从系统环境变量中获取api秘钥 if not api_key: return 你滴环境变量没有配置成功哦 # 初始化TavilyClient客户端 tavily TavilyClient(api_keyapi_key) # 构造查询 query f{city}在weather天气下最值得去的旅游景点推荐及理由 try: # 带所有常用参数的搜索 response tavily.search( query北京今日天气, # 必选 search_depthbasic,#搜索深度basic默认快速搜索advanced深度搜索 max_results5,#最多返回几条搜索结果默认是5 include_answerTrue,#是否让ai直接总结答案 include_raw_contentFalse,#是否返回网页完整原文默认False只返回摘要开启后数据会很大 include_imagesFalse,#是否返回搜索到的图片链接 topicgeneral#指定搜索主题general通用搜索news只搜新闻finance财经sports体育 ) response tavily.search(queryquery, search_depthbasic, include_answerTrue ) # 让代码联网搜索内容并把搜索结果存到 response里,response是一个字典用字典的.get()来取值 if response.get(answer): return response[answer] # 字典的常用键answer(AI直接总结好的答案) # query 你刚才搜索的关键词 # results 搜索到的网页结果列表多条信息 # images 搜索到的图片链接需开启参数 # raw_content 网页完整原文需开启参数 # 如果没有综合回答就格式化原始结果 formatted_result [] for result in response.get(results, []): # []是默认值有结果返回结果没有就返回空列表 formatted_result.append(f- {result[title]}:{result[content]}) if not formatted_result: # 空列表会返回False return 抱歉没有找到相关的旅游景点推荐 except Exception as e: return f错误执行Tavily搜索时出现问题 - {e} available_tools {get_weather: get_weather, get_attraction: get_attraction} class OpenAiCompatibleClient: 一个用于调用任何兼容OpenAI接口的大语言模型服务的客户端,国内外模型都能用 def __init__(self, model: str, api_key: str, base_url: str): self.model model self.client OpenAI(api_keyapi_key, base_urlbase_url) def generate(self, prompt: str, system_prompt: str) - str: print(正在调用大语言模型...) try: messages [{role: system, content: system_prompt}, {role: user, content: prompt}] response self.client.chat.completions.create( modelself.model, messagesmessages, streamFalse # 一次性全部输出,response是一个对象 ) answer response.choices[0].message.content #response长什么样 { # 1. 基础信息没用不用管 id: chatcmpl-xxxxxx, object: chat.completion, created: 1723456789, model: gpt-3.5-turbo, # 2. 核心 choices: [ { index: 0, message: { role: assistant, # 角色AI助手 content: 北京今天晴天气温25℃ # AI 回复的内容 }, finish_reason: stop } ], # 3. 消耗统计没用不用管 usage: { prompt_tokens: 10, completion_tokens: 8, total_tokens: 18 } } print(大语言模型响应成功) return answer except Exception as e: print(f调用大语言模型api时发生错误:{e}) return #实例化循环 AGENT_SYSTEM_PROMPT 你是一个智能旅行助手。你的任务是分析用户的请求并使用可用工具一步步地解决问题。 # 可用工具: - get_weather(city: str): 查询指定城市的实时天气。 - get_attraction(city: str, weather: str): 根据城市和天气搜索推荐的旅游景点。 # 输出格式要求: 你的每次回复必须严格遵循以下格式包含一对Thought和Action Thought: [你的思考过程和下一步计划] Action: [你要执行的具体行动] Action的格式必须是以下之一 1. 调用工具function_name(arg_namearg_value) 2. 结束任务Finish[最终答案] # 重要提示: - 每次只输出一对Thought-Action - Action必须在同一行不要换行 - 当收集到足够信息可以回答用户问题时必须使用 Action: Finish[最终答案] 格式结束 请开始吧 llm OpenAiCompatibleClient(modelMODEL_ID,api_keyAPI_KEY,base_urlBASE_URL) user_prompt 你好请帮我查询一下今天北京的天气然后根据天气推荐一个合适的旅游景点。 prompt_history [f用户请求{user_prompt}] print(f用户请求:{user_prompt}) #开始循环 for i in range(5): print(f---循环{i1}---\n) #构建prompt full_prompt \n.join(prompt_history) #调用LLM思考 llm_output llm.generate(full_prompt,system_promptAGENT_SYSTEM_PROMPT) #模型可能会输出多余的Thought-Action,需要截断 match re.search(r(Thought:.*?Action:.*?)(?\n\s*(?:Thought:|Action:|Observation:)|\Z), llm_output, re.DOTALL) if match: truncated match.group(1).strip() if truncated ! llm_output.strip(): llm_output truncated print(已截断多余的 Thought-Action 对) print(f模型输出:\n{llm_output}\n) prompt_history.append(llm_output) # 3.3. 解析并执行行动 action_match re.search(rAction: (.*), llm_output, re.DOTALL) if not action_match: observation 错误: 未能解析到 Action 字段。请确保你的回复严格遵循 Thought: ... Action: ... 的格式。 observation_str fObservation: {observation} print(f{observation_str}\n * 40) prompt_history.append(observation_str) continue action_str action_match.group(1).strip() if action_str.startswith(Finish): final_answer re.match(rFinish\[(.*)\], action_str).group(1) print(f任务完成最终答案: {final_answer}) break tool_name re.search(r(\w)\(, action_str).group(1) args_str re.search(r\((.*)\), action_str).group(1) kwargs dict(re.findall(r(\w)([^]*), args_str)) if tool_name in available_tools: observation available_tools[tool_name](**kwargs) else: observation f错误:未定义的工具 {tool_name} # 3.4. 记录观察结果 observation_str fObservation: {observation} print(f{observation_str}\n * 40) prompt_history.append(observation_str)