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
| import requests import os from tavily import TavilyClient from openai import OpenAI import re
AGENT_SYSTEM_PROMPT = """ 你是一个智能旅行助手。你的任务是分析用户的请求,并使用可用工具一步步地解决问题。
# 可用工具: - `get_weather(city: str)`: 查询指定城市的实时天气。 - `get_attraction(city: str, weather: str)`: 根据城市和天气搜索推荐的旅游景点。
# 行动格式: 你的回答必须严格遵循以下格式。首先是你的思考过程,然后是你要执行的具体行动,每次回复只输出一对Thought-Action: Thought: [这里是你的思考过程和下一步计划] Action: [这里是你要调用的工具,格式为 function_name(arg_name="arg_value")]
# 任务完成: 当你收集到足够的信息,能够回答用户的最终问题时,你必须在`Action:`字段后使用 `finish(answer="...")` 来输出最终答案。
请开始吧! """
def get_weather(city: str) -> str: """ 通过调用 wttr.in API 查询真实的天气信息。 """ url = f"https://wttr.in/{city}?format=j1" try: response = requests.get(url) response.raise_for_status() data = response.json() current_condition = data['current_condition'][0] weather_desc = current_condition['weatherDesc'][0]['value'] temp_c = current_condition['temp_C'] 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: """ 根据城市和天气,使用Tavily Search API搜索并返回优化后的景点推荐。 """ api_key = os.environ.get("TAVILY_API_KEY") if not api_key: return "错误:未配置TAVILY_API_KEY环境变量。"
tavily = TavilyClient(api_key=api_key) query = f"'{city}' 在'{weather}'天气下最值得去的旅游景点推荐及理由" try: response = tavily.search(query=query, search_depth="basic", include_answer=True) if response.get("answer"): return response["answer"] formatted_results = [] for result in response.get("results", []): formatted_results.append(f"- {result['title']}: {result['content']}") if not formatted_results: return "抱歉,没有找到相关的旅游景点推荐。"
return "根据搜索,为您找到以下信息:\n" + "\n".join(formatted_results)
except Exception as e: return f"错误:执行Tavily搜索时出现问题 - {e}"
available_tools = { "get_weather": get_weather, "get_attraction": get_attraction, }
class OpenAICompatibleClient: """ 一个用于调用任何兼容OpenAI接口的LLM服务的客户端。 """ def __init__(self, model: str, api_key: str, base_url: str): self.model = model self.client = OpenAI(api_key=api_key, base_url=base_url)
def generate(self, prompt: str, system_prompt: str) -> str: """调用LLM API来生成回应。""" print("正在调用大语言模型...") try: messages = [ {'role': 'system', 'content': system_prompt}, {'role': 'user', 'content': prompt} ] response = self.client.chat.completions.create( model=self.model, messages=messages, stream=False ) answer = response.choices[0].message.content print("大语言模型响应成功。") return answer except Exception as e: print(f"调用LLM API时发生错误: {e}") return "错误:调用语言模型服务时出错。"
API_KEY = "ms" BASE_URL = "https://api-inference.modelscope.cn/v1" MODEL_ID = "Qwen/Qwen2.5-7B-Instruct" TAVILY_API_KEY="tvly-" os.environ['TAVILY_API_KEY'] = TAVILY_API_KEY
llm = OpenAICompatibleClient( model=MODEL_ID, api_key=API_KEY, base_url=BASE_URL )
user_prompt = "你好,请帮我查询一下今天北京的天气,然后根据天气推荐一个合适的旅游景点。" prompt_history = [f"用户请求: {user_prompt}"]
print(f"用户输入: {user_prompt}\n" + "="*40)
for i in range(5): print(f"--- 循环 {i+1} ---\n") full_prompt = "\n".join(prompt_history) llm_output = llm.generate(full_prompt, system_prompt=AGENT_SYSTEM_PROMPT) 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) action_match = re.search(r"Action: (.*)", llm_output, re.DOTALL) if not action_match: print("解析错误:模型输出中未找到 Action。") break action_str = action_match.group(1).strip()
if action_str.startswith("finish"): final_answer = re.search(r'finish\(answer="(.*)"\)', 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}'"
observation_str = f"Observation: {observation}" print(f"{observation_str}\n" + "="*40) prompt_history.append(observation_str)
|