用 Python 从零手写你的第一个 AI Agent!手把手带你开发一个智能旅行助手(附完整源码)

def get_weather(city: str) -> str:    # 调用 wttr.in 接口获取实时数据    url = f"https://wttr.in/{city}?format=j1"    # ... (省略网络请求代码，详见文末完整源码)    return f"{city}当前天气:{desc}，气温{temp}摄氏度"def get_attraction(city: str, weather: str) -> str:    # 调用 Tavily Search API 联网搜索    query = f"'{city}' 在'{weather}'天气下最值得去的旅游景点推荐及理由"    # ...    return response["answer"]

AGENT_SYSTEM_PROMPT = """你是一个智能旅行助手。# 可用工具:- get_weather(city: str)- get_attraction(city: str, weather: str)...# 行动格式:Thought: [思考过程]Action: [函数调用，如 get_weather(city="Beijing")]"""

# 伪代码逻辑history = ["用户请求: 帮我查郑州天气并推荐景点"]while True:    # 1. 让 LLM 思考    response = llm.generate(history)    # 2. 解析 LLM 想要调用的工具    if "Action:" in response:        tool_name, args = parse(response)        # 3. 执行工具 (Python 真正执行函数的地方)        result = tools[tool_name](**args)        # 4. 将结果作为 Observation 存入历史        history.append(f"Observation: {result}")    # 5. 如果 LLM 决定结束    if "finish" in response:        print("最终答案:", final_answer)        break

import requestsimport jsonimport osimport sysfrom tavily import TavilyClientfrom openai import OpenAIimport reimport configAGENT_SYSTEM_PROMPT = """你是一个智能旅行助手。你的任务是分析用户的请求，并使用可用工具一步步地解决问题。根据用户的输入，确定需要调用哪些工具来获取信息，并帮助用户规划旅行过程。# 可用工具:- `get_attraction(city: str, weather: str)`: 根据城市和天气搜索推荐的旅游景点。- `get_forecast(city: str, days: int)`: 查询指定城市未来若干天的天气预报。- `get_weather(city: str)`: 查询指定城市的实时天气。# 行动格式:你的回答必须严格遵循以下格式。首先是你的思考过程，然后是你要执行的具体行动，每次回复只输出一对Thought-Action：Thought: [这里是你的思考过程和下一步计划]Action: [这里是你要调用的工具，格式为 function_name(arg_name="arg_value")]# 任务完成:当你收集到足够的信息，能够回答用户的最终问题时，你必须在`Action:`字段后使用 `finish(answer="...")` 来输出最终答案。请开始吧！"""def get_weather(city: str) -> str:    """    通过调用 wttr.in API 查询真实的天气信息。    """    # API端点，我们请求JSON格式的数据    url = f"https://wttr.in/{city}?format=j1"    try:        # 发起网络请求        response = requests.get(url)        # 检查响应状态码是否为200 (成功)        response.raise_for_status()         # 解析返回的JSON数据        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_forecast(city: str, days: str = "3") -> str:    """    查询未来几天的天气。`days`可以是字符串或整数，默认为3天，最大支持7天。    """    try:        days_int = int(days)    except Exception:        return "错误:参数 days 必须为整数。"    if days_int < 1:        return "错误:days 必须至少为1。"    if days_int > 7:        days_int = 7    url = f"https://wttr.in/{city}?format=j1"    try:        response = requests.get(url)        response.raise_for_status()        data = response.json()        weather_list = data.get("weather", [])        if not weather_list:            return "错误:未能获取到天气预报信息。"        days_available = min(days_int, len(weather_list))        parts = [f"{city} 接下来 {days_available} 天的天气预报:"]        for i in range(days_available):            day = weather_list[i]            date = day.get("date", "未知日期")            maxtemp = day.get("maxtempC", "?")            mintemp = day.get("mintempC", "?")            desc = ""            # 有时 hourly 中会包含 weatherDesc            try:                hourly = day.get("hourly", [])                if hourly:                    desc = hourly[0].get("weatherDesc", [{"value": ""}])[0].get("value", "")            except Exception:                desc = ""            parts.append(f"- {date}: {desc}，最高{maxtemp}°C，最低{mintemp}°C")        return "\n".join(parts)    except requests.exceptions.RequestException as e:        return f"错误:查询天气预报时遇到网络问题 - {e}"    except Exception as e:        return f"错误:解析天气预报数据时出错 - {e}"def get_attraction(city: str, weather: str) -> str:    """    根据城市和天气，使用Tavily Search API搜索并返回优化后的景点推荐。    """    # 1. 从环境变量中读取API密钥    api_key = config.TAVILY_API_KEY    if not api_key:        return "错误:未配置TAVILY_API_KEY环境变量。"    # 2. 初始化Tavily客户端    tavily = TavilyClient(api_key=api_key)    # 3. 构造一个精确的查询    query = f"'{city}' 在'{weather}'天气下最值得去的旅游景点推荐及理由"    try:        # 4. 调用API，include_answer=True会返回一个综合性的回答        response = tavily.search(query=query, search_depth="basic", include_answer=True)        # 5. Tavily返回的结果已经非常干净，可以直接使用        # response['answer'] 是一个基于所有搜索结果的总结性回答        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_forecast": get_forecast,    "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 "错误:调用语言模型服务时出错。"# --- 1. 配置LLM客户端 ---# 请根据您使用的服务，将这里替换成对应的凭证和地址llm = OpenAICompatibleClient(    model=config.AI_MODEL,    api_key=config.AI_KEY,    base_url=config.AI_URL)# --- 2. 初始化 ---user_prompt = input("请输入您的请求（例如：查询郑州的天气并推荐景点）： ").strip()if not user_prompt:    print("未输入内容，程序退出。")    sys.exit(0)prompt_history = [f"用户请求: {user_prompt}"]print(f"用户输入: {user_prompt}\n" + "="*40)# --- 3. 运行主循环 ---for i in range(10): # 设置最大循环次数    print(f"--- 循环 {i+1} ---\n")    # 3.1. 构建Prompt    full_prompt = "\n".join(prompt_history)    # 3.2. 调用LLM进行思考    llm_output = llm.generate(full_prompt, system_prompt=AGENT_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(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"):        # 支持多种格式的 finish(answer=..)，包含单/双引号并允许跨行内容        finish_match = re.search(r'finish\s*\(\s*answer\s*=\s*(?P<q>[\'\"])(?P<ans>.*?)(?P=q)\s*\)', action_str, re.DOTALL)        if finish_match:            final_answer = finish_match.group('ans')        else:            # 退回策略：尝试提取括号内的原始内容            inner = re.search(r'finish\s*\((.*)\)\s*$', action_str, re.DOTALL)            if inner:                final_answer = inner.group(1).strip()            else:                final_answer = action_str        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 = f"Observation: {observation}"    print(f"{observation_str}\n" + "="*40)    prompt_history.append(observation_str)# print("\nPrompt History:", prompt_history)

#config.pyfrom dotenv import load_dotenvimport os# 加载 .env 文件load_dotenv()# 从 .env 获取环境变量AI_KEY = os.getenv('AI_KEY')AI_URL = os.getenv('AI_URL')AI_MODEL = os.getenv('AI_MODEL')TAVILY_API_KEY = os.getenv('TAVILY_API_KEY')# 打印加载的配置if __name__ == "__main__":    print("=== 环境变量配置 ===")    print(f"AI_KEY: {AI_KEY}")    print(f"AI_URL: {AI_URL}")    print(f"AI_MODEL: {AI_MODEL}")    print(f"TAVILY_API_KEY: {TAVILY_API_KEY}")