杂七杂八的学习记录。一些学习笔记和一些debug记录。

教程来自datawhale的:第十三章 智能旅行助手

1环境配置

如果没有安Node.js和npm就去https://nodejs.org/,一起安装的。教程我用的是Node.js详细安装教程-腾讯云开发者社区-腾讯云

启动后端:

开一个新的虚拟环境,我的就叫travel_agent。

# 1. 进入后端目录

cd helloagents-trip-planner/backend

# 2. 安装依赖 pip install -r requirements.txt

# 3. 启动后端服务 uvicorn app.api.main:app --reload # 或者 python run.py

 成功启动后,访问 http://localhost:8000/docs 可以看到 API 文档。

HelloAgents V0.2.9是学习版,装这个

requirements里面有安装uv,而我安装的时候uv不知道为什么pip装不上,被卡出来了只好用conda安装。装完uv之后就可以uv pip install -r requirements.txt了,比pip快得多。

启动前端:

打开新的终端窗口:

# 1. 进入前端目录

cd helloagents-trip-planner/frontend

# 2. 安装依赖

npm install

# 3. 启动前端服务

npm run dev

成功启动后,访问 http://localhost:5173 即可使用应用。


2快速体验遇到的bug(13.1.3)

运行后端:

关于API文档:

它是FastAPI框架带来的。

用开餐厅来比方:

1. 它是餐厅的“点单菜单” (接口清单)

看下面第一张截图,里面有绿色的 POST 和蓝色的 GET。

前端(网页/小程序)是顾客,它只会发请求。

Agent 是后厨,负责查天气、算预算、写计划。

这个页面(API)就是菜单。它告诉前端:如果你想让后厨帮你“生成旅行计划”(/api/trip/plan),就必须填好菜单上的那些必填项。

2. 它是 Agent 的测试器

不用写前端界面,直接能在这里测试Agent聪不聪明。

做法:

点第二张截图右上角有一个Try it out;

那个黑色的 JSON 文本框原本是锁定的,点完之后就可以编辑了;

把里面的数据改成自己喜欢的,比如:把 "city": "北京" 改成 "city": "东京",或者把天数改成 5;

往下滑,点Execute。

我自己遇到的报错:

1.尝试路径规划的时候(地图服务-路径规划-try it out)报错

提示:

写测试脚本:

1.测试高德api是否可用
import requests

def test_amap_weather(api_key, city="110000"): # 110000 是北京的 adcode
    url = f"https://restapi.amap.com/v3/weather/weatherInfo?city={city}&key={api_key}"
    response = requests.get(url)
    print("高德官方返回的原始数据:")
    print(response.json())

# 把这里替换成你的高德 Key!
my_key = "真实key" 
test_amap_weather(my_key)

结果是api可用,运行得到了北京的详细地理信息。

2.测试map.py

在backend/app/api/routes里面找到map.py,搜索找到“规划路线”,把原本的规划路线函数注释掉,换成测试函数:

#这是一个断点测试脚本
# @router.post(
#     "/route",
#     response_model=RouteResponse,
#     summary="规划路线",
#     description="规划两点之间的路线"
# )
# async def plan_route(request: RouteRequest):
#     try:
#         service = get_amap_service()
#         route_info = service.plan_route(
#             origin_address=request.origin_address,
#             destination_address=request.destination_address,
#             origin_city=request.origin_city,
#             destination_city=request.destination_city,
#             route_type=request.route_type
#         )
        
#         import json
#         import re
#         import traceback
        
#         # === 🚨 深度解剖探针开始 🚨 ===
#         print("\n" + "="*40)
#         print("🕵️‍♂️ [1] 数据类型检测:")
#         print(f"route_info 的 type 是: {type(route_info)}")
        
#         print("\n🕵️‍♂️ [2] 原生内容扫描 (repr):")
#         # repr() 会把所有隐藏的回车、截断、乱码全部暴露出来
#         print(repr(route_info))
        
#         parsed_dict = {}
        
#         if isinstance(route_info, dict):
#             print("\n🕵️‍♂️ [3] 路线清晰: 这是一个完美字典!直接使用。")
#             parsed_dict = route_info
            
#         elif isinstance(route_info, str):
#             print("\n🕵️‍♂️ [3] 路线清晰: 这是一个字符串,准备正则提取...")
#             match = re.search(r'\{.*\}', route_info, re.DOTALL)
#             if match:
#                 json_str = match.group(0)
#                 print(f"\n🕵️‍♂️ [4] 正则提取成功!即将喂给 json.loads 的字符串长度: {len(json_str)}")
#                 try:
#                     parsed_dict = json.loads(json_str)
#                     print("🕵️‍♂️ [5] 🎉 JSON 解析大获成功!")
#                 except Exception as e:
#                     print(f"\n🕵️‍♂️ [5] ❌ 致命错误:JSON 解析崩溃!")
#                     print(f"崩溃原因: {e}")
#                     print(f"提取出的问题字符串片段 (前100字): {json_str[:100]}...")
#                     print(f"提取出的问题字符串片段 (最后100字): {json_str[-100:]}")
#             else:
#                 print("\n🕵️‍♂️ [4] ❌ 致命错误:正则没有找到任何包裹在 {} 里的内容!")
#         else:
#             print("\n🕵️‍♂️ [3] ❌ 致命错误:这既不是字典也不是字符串!")
            
#         print("="*40 + "\n")
#         # === 🚨 深度解剖探针结束 🚨 ===

#         # --- 正常的数据拼装流程 ---
#         distance_val = 0
#         duration_val = 0
#         desc_val = "后端强行喂饭成功!但数据是兜底测试值。"
        
#         paths = parsed_dict.get("route", {}).get("paths", [])
#         if paths and len(paths) > 0:
#             first_path = paths[0]
#             distance_val = int(first_path.get("distance", 0))
#             duration_val = int(first_path.get("duration", 0))
#             desc_val = f"真实数据提取成功!距离约 {distance_val/1000:.1f} 公里"
            
#         clean_data = {
#             "distance": distance_val,
#             "duration": duration_val,
#             "route_type": request.route_type,
#             "description": desc_val
#         }

#         return RouteResponse(
#             success=True,
#             message="路线规划成功",
#             data=clean_data
#         )
        
#     except Exception as e:
#         print(f"❌ 路线规划发生异常: {str(e)}")
#         raise HTTPException(status_code=500, detail=f"路线规划失败: {str(e)}")

运行得到

可以看到它第二步传递了一个没有信息的{},这就是导致pydantic报错的原因。

3.找amap_service.py的问题

上一步发现,这不是map.py的问题,map.py输入的就是{},转去backend/app/services里面找到amap_service.py,找到plan_route函数第182行,原代码写的是return {},改成return result。

改完之后运行map.py测试版本,得到真实地理数据:

4.回去删掉测试函数,继续在api文档测试路径规划。

得到报错如图:

这是因为修改了 amap_service.py,现在它把一长串字符串,开头是 工具 'maps_direction...' 执行结果:\n{...}原封不动地传给了原版的 map.py。原版 map.py 拿到这串字符串后,直接扔给了 Pydantic。

报错的这一行:Input should be a valid dictionary... [input_type=str]。Pydantic说他要的是一个字典dic,但得到了一串str,直接拒收。

修改plan_route 函数的 try 块:

try:
        # 1. 拿到服务层传来的包裹 (带中文前缀的字符串)
        service = get_amap_service()
        route_info = service.plan_route(
            origin_address=request.origin_address,
            destination_address=request.destination_address,
            origin_city=request.origin_city,
            destination_city=request.destination_city,
            route_type=request.route_type
        )
        
        # 2. 生产级数据清洗
        import json
        import re
        
        clean_data = {}
        
        if isinstance(route_info, str):
            # 用正则精准切除中文字符,只提取 {} 里的 JSON
            match = re.search(r'\{.*\}', route_info, re.DOTALL)
            if match:
                try:
                    parsed_dict = json.loads(match.group(0))
                    paths = parsed_dict.get("route", {}).get("paths", [])
                    if paths:
                        # 完美组装 Pydantic 需要的四个字段
                        clean_data = {
                            "distance": int(paths[0].get("distance", 0)),
                            "duration": int(paths[0].get("duration", 0)),
                            "route_type": request.route_type,
                            "description": f"获取成功!全程距离约 {int(paths[0].get('distance', 0))/1000:.1f} 公里"
                        }
                except Exception as e:
                    print(f"数据清洗异常: {e}")
                    
        # 3. 终极兜底 (防止高德偶尔抽风没查到数据)
        if not clean_data:
            clean_data = {
                "distance": 0,
                "duration": 0,
                "route_type": request.route_type,
                "description": "已连通服务,但未获取到有效距离数据"
            }

        # 4. 把干干净净的字典递给 Pydantic
        return RouteResponse(
            success=True,
            message="路线规划成功",
            data=clean_data
        )
        
    except Exception as e:
        print(f"❌ 路线规划失败: {str(e)}")
        raise HTTPException(
            status_code=500,
            detail=f"路线规划失败: {str(e)}"
        )

继续测试得到正确运行了的结果。

2.尝试生成旅行计划(旅行规划-生成旅行计划-try it out):

报错,提示用不了搜索工具。

1.测试工具是否注册。

按照之前的思路,写一个测试脚本看看他到底有没有正确注册工具可以用:

新建backen下一个test_mcp_tools.py:

# test_mcp_tools.py
import json
from app.services.amap_service import get_amap_mcp_tool

def run_xray_test():
    print("====== 🚀 MCP 插件底层 X光扫描 ======\n")
    
    try:
        # 1. 启动并获取工具
        tool = get_amap_mcp_tool()
        tools_list = tool._available_tools
        
        if not tools_list:
            print("❌ 致命错误:没有读取到任何可用工具!")
            return

        print(f"✅ 成功连接 MCP,共扫描到 {len(tools_list)} 个底层工具\n")
        
        # 2. 扒出搜索和天气的“真实名字”和“参数清单”
        print("====== 🔍 重点排查:工具真实身份与参数要求 ======")
        for t in tools_list:
            name = t.get('name', '未知')
            schema = t.get('inputSchema', {})
            # 只过滤出包含天气、搜索、路线相关的工具
            if 'weather' in name or 'search' in name or 'poi' in name:
                print(f"👉 发现工具: {name}")
                print(f"   需要参数: {json.dumps(schema, ensure_ascii=False)}")
                print("-" * 50)
                
        # 3. 暴力裸调:天气查询
        print("\n====== ⚡ 暴力裸调测试:天气 (maps_weather) ======")
        try:
            res_weather = tool.run({
                "action": "call_tool",
                "tool_name": "maps_weather",
                "arguments": {"city": "北京"} # 看看只传北京行不行
            })
            print(f"☁️ 天气返回结果: {str(res_weather)[:200]}...")
        except Exception as e:
            print(f"❌ 天气调用崩溃: {e}")

        # 4. 暴力裸调:景点搜索
        print("\n====== ⚡ 暴力裸调测试:搜索 (maps_text_search) ======")
        try:
            res_search = tool.run({
                "action": "call_tool",
                "tool_name": "maps_text_search",
                "arguments": {"keywords": "故宫", "city": "北京"}
            })
            print(f"🏛️ 搜索返回结果: {str(res_search)[:200]}...")
        except Exception as e:
            print(f"❌ 搜索调用崩溃: {e}")

    except Exception as e:
        print(f"❌ X光机启动失败: {e}")

if __name__ == "__main__":
    run_xray_test()

得到输出结果:

后面是长长的报错,就不截图了。

可以看到p1需要参数: {},这就是问题。大模型(LLM)每次要调用工具前,都必须看一眼工具的“说明书”(也就是参数列表),而这里amap-mcp-server只告诉他{},他就不行了。

p2调试可以看到高德api是可以用的。

p3及后面的报错是因为Python asyncio 的一个祖传老 Bug。当你的测试脚本运行完毕准备退出时,Python 试图去关闭那个叫 uvx 的子进程,但发现管道已经被系统提前关了。不用理他就行。

2.加入追踪看到底是哪里有问题。

找到backend/app/agents里面的trip_planner_agent.py,找到def plan_trip里面的四个步骤,注释掉原本的,替换成追踪代码:

from app.services.amap_service import get_amap_mcp_tool
#这个放最前面import
# =================================================================
            # 获取全局的 MCP 工具实例
            mcp_tool = get_amap_mcp_tool()

            print("\n" + "🔥"*25)
            print("====== 📡 开启全链路追踪 (上帝视角) ======")
            print("🔥"*25)

            # =========================================================
            # (追踪) 步骤 1: 景点搜索
            # =========================================================
            print("\n📍 (追踪) 步骤1: 搜索景点...")
            print(f"⚙️ [追踪 - Python直接发给高德]: 请求 {request.city} 的景点")
            try:
                raw_poi = mcp_tool.run({
                    "action": "call_tool",
                    "tool_name": "maps_text_search",
                    "arguments": {"keywords": "历史文化 景点", "city": request.city}
                })
                print(f"📥 [追踪 - 高德API原生返回]: {str(raw_poi)[:200]}...")
            except Exception as e:
                raw_poi = f"高德调用失败: {e}"
                print(f"❌ [追踪 - 高德崩溃]: {e}")

            attraction_query = f"用户想去{request.city}。这是高德API返回的真实景点数据:{raw_poi}。请根据这些数据整理出推荐景点。"
            print(f"📤 [追踪 - 发给大模型翻译]: (已隐藏长串原生数据,让大模型开始排版...)")
            attraction_response = self.attraction_agent.run(attraction_query)
            print(f"🤖 [追踪 - 大模型最终输出]: {attraction_response[:200]}...\n")


            # =========================================================
            # (追踪) 步骤 2: 天气查询
            # =========================================================
            print("\n🌤️ (追踪) 步骤2: 查询天气...")
            print(f"⚙️ [追踪 - Python直接发给高德]: 请求 {request.city} 的天气")
            try:
                raw_weather = mcp_tool.run({
                    "action": "call_tool",
                    "tool_name": "maps_weather",
                    "arguments": {"city": request.city}
                })
                print(f"📥 [追踪 - 高德API原生返回]: {str(raw_weather)[:200]}...")
            except Exception as e:
                raw_weather = f"高德调用失败: {e}"
                print(f"❌ [追踪 - 高德崩溃]: {e}")

            weather_query = f"这是高德API返回的{request.city}真实天气数据:{raw_weather}。请帮我总结成自然连贯的天气预报。"
            print(f"📤 [追踪 - 发给大模型翻译]: (强制要求大模型阅读原生天气数据...)")
            weather_response = self.weather_agent.run(weather_query)
            print(f"🤖 [追踪 - 大模型最终输出]: {weather_response[:200]}...\n")


            # =========================================================
            # (追踪) 步骤 3: 酒店推荐
            # =========================================================
            print("\n🏨 (追踪) 步骤3: 搜索酒店...")
            print(f"⚙️ [追踪 - Python直接发给高德]: 请求 {request.city} 的 {request.accommodation} 酒店")
            try:
                raw_hotel = mcp_tool.run({
                    "action": "call_tool",
                    "tool_name": "maps_text_search",
                    "arguments": {"keywords": f"{request.accommodation} 酒店", "city": request.city}
                })
                print(f"📥 [追踪 - 高德API原生返回]: {str(raw_hotel)[:200]}...")
            except Exception as e:
                raw_hotel = f"高德调用失败: {e}"
                print(f"❌ [追踪 - 高德崩溃]: {e}")

            hotel_query = f"这是高德API返回的{request.city}酒店数据:{raw_hotel}。请挑选出适合的酒店并进行推荐。"
            print(f"📤 [追踪 - 发给大模型翻译]: (强制要求大模型阅读原生酒店数据...)")
            hotel_response = self.hotel_agent.run(hotel_query)
            print(f"🤖 [追踪 - 大模型最终输出]: {hotel_response[:200]}...\n")


            # =========================================================
            # (追踪) 步骤 4: 行程总规划
            # =========================================================
            print("\n📋 (追踪) 步骤4: 生成行程计划...")
            planner_query = self._build_planner_query(request, attraction_response, weather_response, hotel_response)
            print(f"📤 [追踪 - 终极长文发给大模型]: 整合前三步结果,准备生成最终游记... (这步最容易断连,祈祷!)")
            try:
                planner_response = self.planner_agent.run(planner_query)
                print(f"🤖 [追踪 - 大模型最终输出]: {planner_response[:300]}...\n")
            except Exception as e:
                print(f"❌ [追踪 - 终极崩溃]: DeepSeek 最终连接断开死因: {e}")
                planner_response = f"行程生成失败: {e}"

            # 解析最终计划 (保留原代码)
            trip_plan = self._parse_response(planner_response, request)

            print(f"\n{'='*60}")
            print(f"✅ 全链路测试结束!旅行计划生成完成!")
            print(f"{'='*60}\n")

            return trip_plan

再次测试得到:

可以看到其实他都能调用高德api来搜索。

那就直接保留这个测试版本就好了。

原版是这么写的:agent.run("请查询北京天气")。 它完全交给了大模型,大模型收到指令-去翻看 MCP 插件提供的“工具说明书”-说明书是空的{}-大模型不知道怎么调用就告诉你:“抱歉,工具不可用”。

现在Python 直接接管:用 Python 直接硬编码去调用高德 API (mcp_tool.run(...))。Python 是不需要看大模型说明书的,它拿到真实参数直接去调高德,100% 能拿到原生 JSON 数据。

大模型只做翻译官: 把拿到的原生 JSON 连同指令一起打包,强行塞给大模型(weather_query = f"这是真实数据:{raw_weather}。请帮我总结...")。

大模型这时候根本不需要去调用工具,它只负责把JSON 翻译成人类语言。

运行前端:

没有地图图片和景点真实图片:

可以看到地图完全空白,景区配图12直接是文字,3并不是夫子庙,反而像是上海。

怀疑是backend/app/services/里面的unsplash_service.py有问题。

1.先写一个测试脚本

看看unsplash api key能不能正常工作。放在backend下面:

import os
import requests
from dotenv import load_dotenv

# 加载 .env 文件里的变量
load_dotenv()

def test_unsplash_search():
    print("====== 📸 Unsplash API 裸调探针 ======\n")
    
    # 获取你的 API Key (请确保你的 .env 里这个名字是对的)
    # 有些教程叫 UNSPLASH_API_KEY,有些叫 UNSPLASH_ACCESS_KEY
    api_key = os.getenv("UNSPLASH_API_KEY") or os.getenv("UNSPLASH_ACCESS_KEY")
    
    if not api_key:
        print("❌ 致命错误:在 .env 文件中没有找到 Unsplash API Key!")
        return
        
    print(f"✅ 成功获取 API Key: {api_key[:5]}...{api_key[-5:]}\n")

    # 你要测试的三个景点
    test_locations = ["水游城", "夫子庙", "玄武湖景区", "Confucius Temple Nanjing"]
    
    for location in test_locations:
        print(f"🔍 正在搜索: 【{location}】...")
        
        # Unsplash 官方搜索 API 接口
        url = "https://api.unsplash.com/search/photos"
        params = {
            "query": location,
            "client_id": api_key,
            "per_page": 1, # 我们只需要看第一张图
            "orientation": "landscape" # 横图
        }
        
        try:
            response = requests.get(url, params=params)
            
            if response.status_code == 200:
                data = response.json()
                total_results = data.get("total", 0)
                
                if total_results > 0:
                    image_url = data["results"][0]["urls"]["regular"]
                    description = data["results"][0].get("alt_description", "无描述")
                    print(f"  ✅ 搜到 {total_results} 张图!")
                    print(f"  🖼️ 第一张图的描述: {description}")
                    print(f"  🔗 图片链接: {image_url}\n")
                else:
                    print(f"  ❌ 结果为 0!Unsplash 根本不认识这个词!\n")
            else:
                print(f"  ❌ API 报错!状态码: {response.status_code}, 详情: {response.text}\n")
                
        except Exception as e:
            print(f"  ❌ 请求崩溃: {e}\n")

if __name__ == "__main__":
    test_unsplash_search()

测出来的结果是三个中文搜索词(水游城,夫子庙和玄武湖)都没有,而英文搜索词(Confucius Temple Nanjing)识别出来22张图片,然后我点了排名第一的图片进去看,如下:
(不是南京人,但我去旅游时候看见的夫子庙不长这样啊)

初步判断是调用unsplash有问题:1.可能是直接发了中文词去搜索(大概率),2.可能是unsplash图库里面中国的旅游景点图不多(存疑)

2.看是否是语言问题

找到backend\app\services\unsplash_service.py,去看他有没有翻译过:没发现翻译过。去全局搜索找get_unsplash_service,找到backend\app\api\routes\poi.py里面的调用:

没有翻译步骤。初步判断的确是翻译问题。

3.修改这个get_attraction_photo函数,加上翻译部分:
from app.services.llm_service import get_llm
#import放到最前面去

async def get_attraction_photo(name: str):
    """
    获取景点图片 (带极速 LLM 翻译)
    """
    try:
        unsplash_service = get_unsplash_service()

        print(f"\n📸 正在为前端获取图片: 【{name}】")

        # 1. 拦截并使用 LLM 进行极速翻译
        print(f"🌍 准备翻译: {name}")
        try:
            # 直接向中央分配中心申请一个 LLM 实例
            llm = get_llm()
            prompt = f"请将这个中国景点名称翻译成最地道、最容易在外国图库搜到的英文名。只输出英文,不要任何废话。景点名:{name}"
            
            messages = [{"role": "user", "content": prompt}]
            
            # 发送列表给大模型
            response_text = llm.invoke(messages)
            
            # 清理格式
            english_name = str(response_text).strip().replace('"', '').replace("'", "")
            print(f"✅ 极速翻译成功: {name} -> {english_name}")
            
        except Exception as e:
            print(f"⚠️ 翻译崩溃,降级使用拼音/原名: {e}")
            english_name = f"{name} China"

        # 2. 使用地道的英文名去 Unsplash 搜索
        # 加上 landmark 增加找景点的准确率
        search_query = f"{english_name} landmark"
        print(f"🔍 最终发给 Unsplash 的搜索词: {search_query}")
        
        photo_url = unsplash_service.get_photo_url(search_query)

        if not photo_url:
            print("⚠️ 没找到图,尝试放宽限制...")
            # 如果没找到,尝试只用英文原名搜索
            photo_url = unsplash_service.get_photo_url(english_name)

        if photo_url:
            print("✅ 成功获取到真实图片链接!")
        else:
            print("❌ 依然没有图 (Unsplash图库可能没收录)")

        return {
            "success": True,
            "message": "获取图片成功",
            "data": {
                "name": name,
                "photo_url": photo_url
            }
        }

    except Exception as e:
        print(f"❌ 获取景点图片失败: {str(e)}")
        raise HTTPException(
            status_code=500,
            detail=f"获取景点图片失败: {str(e)}"
        )

然后再在backend下面新建一个测试脚本test_photo_api.py测一下能不能跑的通,跑通了再去前端试试:

import asyncio
# 导入你刚刚修改过的带有极速翻译功能的那个函数
from app.api.routes.poi import get_attraction_photo

async def test_photos():
    print("====== 📸 景点搜图接口 (含极速翻译) 独立测试 ======\n")
    
    # 我们就测这三个之前死活搜不到图的“硬骨头”
    test_locations = ["夫子庙", "水游城", "玄武湖景区"]
    
    for location in test_locations:
        print("=" * 50)
        print(f"🎯 正在测试获取: 【{location}】 的图片...")
        
        try:
            # 直接调用你的接口,就像前端向后端发请求一样
            result = await get_attraction_photo(location)
            
            # 打印最终前端会收到的干净 JSON
            if result.get("success"):
                photo_url = result["data"]["photo_url"]
                if photo_url:
                    print(f"\n🎉 测试成功!前端成功拿到了图片链接!")
                    print(f"🔗 链接地址: {photo_url}")
                else:
                    print(f"\n⚠️ 接口跑通了,但是 Unsplash 图库里确实没有这张图 (返回了 None)")
            else:
                print(f"\n❌ 接口返回失败: {result}")
                
        except Exception as e:
            print(f"\n❌ 函数执行崩溃: {e}")
            
    print("\n" + "=" * 50)
    print("🏁 全部单点测试完毕!")

if __name__ == "__main__":
    # 因为 get_attraction_photo 是 async 函数,必须用 asyncio 跑
    asyncio.run(test_photos())

最后跑出来的结果是:

夫子庙和玄武湖都搜到了真实图片(看起来像是正确的图片),但是水游城它给了我一个水立方的图片(无语了)。

为了不让他放宽搜索条件从而搜出来奇怪东西,修改poi.py,找不到就说没有别硬找了

        if not photo_url:
             print("⚠️ Unsplash 没找到图,尝试把关键词放宽到純英文原名...")
             # 如果没找到,尝试只用英文原名搜索 (最后一次诚实的尝试)
             photo_url = unsplash_service.get_photo_url(english_name)

        if photo_url:
            print("✅ 成功获取到 Unsplash 真实图片链接!")
        else:
            # 保持 null (None)
            print("❌ Unsplash图库确实没有收录此景点 (返回空 None)")

        return {
            "success": True,
            "message": "获取图片成功",
            "data": {
                "name": name,
                "photo_url": photo_url # <--- 搜不到就是 None,没毛病
            }
        }
    except Exception as e:
        print(f"❌ 获取景点图片失败: {str(e)}")
        raise HTTPException(
            status_code=500,
            detail=f"获取景点图片失败: {str(e)}"
        )
4.解决显示地图空白的问题。

在前端按f12,发现是没填高德api key(-_-||)补上。

需要申请两个。

5.考虑用换个图库,试试Serper.dev。

首先去https://serper.dev/注册,新用户有2500 free credits。

写一个脚本测试一下能不能搜到令人满意的图片:

import requests
import json

def test_serper_image_search():
    print("====== 📸 Serper.dev (Google Images) API 裸调探针 ======\n")
    
    # 替换成你在 serper.dev 后台复制的 API Key
    api_key = "替换成你的_SERPER_API_KEY"
    
    if api_key == "替换成你的_SERPER_API_KEY":
        print("❌ 请先去 serper.dev 注册并填入你的 API Key!")
        return

    test_locations = ["南京水游城", "南京夫子庙", "南京玄武湖景区"]
    
    url = "https://google.serper.dev/images"
    headers = {
        'X-API-KEY': api_key,
        'Content-Type': 'application/json'
    }
    
    for location in test_locations:
        print(f"🔍 正在通过 Serper 搜索: 【{location}】...")
        
        # 构造请求参数,强制指定中国区域和中文语言
        payload = json.dumps({
            "q": location,
            "gl": "cn",     # 地区:中国
            "hl": "zh-cn",  # 语言:简体中文
            "num": 1        # 只要第一张图
        })
        
        try:
            response = requests.request("POST", url, headers=headers, data=payload)
            response.raise_for_status()
            
            data = response.json()
            images = data.get("images", [])
            
            if images:
                image_url = images[0].get("imageUrl")
                source = images[0].get("source", "未知来源")
                title = images[0].get("title", "无标题")
                
                print(f"  ✅ 成功搜到图!来源: {source} ({title})")
                print(f"  🔗 图片链接: {image_url}\n")
            else:
                print(f"  ❌ 结果为 0!\n")
                
        except Exception as e:
            print(f"  ❌ 请求崩溃: {e}\n")

if __name__ == "__main__":
    test_serper_image_search()

返回给我的水游城图片如下:

然后看下来源:

能用,去改原代码。

6.备份,然后去改使用的图库

(用git备份:新开一个cmd,cd到要保存的根目录-git init-git add .-git commit -m "初始备份:跑通高德地图与底层MCP"-git checkout -b feature/serper-images,如果要恢复到原始的就git checkout main-git restore .)

在backend/app/services下面新建serper_service.py:

"""
Serper.dev (Google Images) 图片搜索服务
[风格比照 unsplash_service.py 编写]
"""

import requests
import json
from typing import List, Optional
from ..config import get_settings # 比照作者:从配置文件引入设置

class SerperService:
    """Serper.dev图片服务类"""
    
    def __init__(self):
        """初始化服务 [比照作者风格]"""
        settings = get_settings()
        # 请确保你在下一节完成了 app/config.py 的修改
        # 如果报错 AttributeError,说明 config.py 没加字段
        self.api_key = getattr(settings, "serper_api_key", None)
        self.base_url = "https://google.serper.dev/images"
        
        # 预设请求头
        self.headers = {
            'X-API-KEY': self.api_key,
            'Content-Type': 'application/json'
        }
    
    def search_photos(self, query: str, count: int = 5) -> List[dict]:
        """
        搜索图片
        [比照作者的 unsplash_service.search_photos 结构]
        """
        if not self.api_key:
            print("❌ Serper搜索失败: 未配置 SERPER_API_KEY")
            return []

        try:
            # 构造 POST 请求体,强制指定中文本地化环境
            payload = json.dumps({
                "q": query,
                "gl": "cn",     # 地区:中国 [移植核心:解决幻觉]
                "hl": "zh-cn",  # 语言:简体中文
                "num": count    # 只要第一张图
            })
            
            # 比照作者使用 requests 发起请求
            response = requests.post(self.base_url, headers=self.headers, data=payload, timeout=10)
            response.raise_for_status()
            
            data = response.json()
            # Serper 返回的图片数组在 'images' 键下
            results = data.get("images", [])
            
            # 提取图片信息,并比照 Unsplash 的字典结构规范化输出
            photos = []
            for photo in results:
                photos.append({
                    "id": photo.get("title"), # 没有明确ID,用标题代替
                    # 关键图片URL
                    "url": photo.get("imageUrl"), 
                    # 缩略图URL
                    "thumb": photo.get("thumbnailUrl"),
                    # 描述
                    "description": photo.get("title"),
                    # 来源网站名
                    "photographer": photo.get("source") 
                })
            
            return photos
            
        except Exception as e:
            print(f"❌ Serper搜索失败: {str(e)}")
            return []
    
    def get_photo_url(self, query: str) -> Optional[str]:
        """
        获取单张图片URL
        [比照作者的 unsplash_service.get_photo_url 结构]
        """
        photos = self.search_photos(query, count=1)
        if photos:
            return photos[0].get("url")
        return None


# ==========================================
# 👇 全局服务实例 (单例模式) [完全比照作者风格]
# ==========================================
_serper_service = None

def get_serper_service() -> SerperService:
    """获取Serper服务实例"""
    global _serper_service
    
    if _serper_service is None:
        _serper_service = SerperService()
    
    return _serper_service

然后要改backend的env,加上SERPER_API_KEY=your_serper_key;

找到backend\app\config.py,在原本的    # Unsplash API配置下面加一行serper_api_key: str = "";

找到backend/app/api/routes/poi.py,改顶部import:from app.services.serper_service import get_serper_service,函数内调用改为:serper_service = get_serper_service(),搜图改为:photo_url = serper_service.get_photo_url(name)

最后规划结果:

地图√正确图片√

最后合并分支:git add .,git commit -m "改用serper代替unsplash,新增 serper_service.py"

3.数据模型设计(13.2)

1.pydantic模型

  • 保安(验证): 规定必须是数字,谁敢传字符串,直接挡在门外并报错,不让脏数据污染系统。
  • 清洁工(转换/解析): 比如 16℃ 变 16。把稍微有点乱的数据(比如外部 API 返回的格式不统一)自动清洗干净,变成需要的标准格式。
  • 翻译官(序列化/反序列化 ):反序列化(进门): 把大模型或者前端吐出来的一堆乱七八糟的 JSON 文本,自动组装成你写好的 Python 类(比如 Attraction)。序列化(出门): 等你后端处理完了,他又能一键把你的 Python 类,变成干净标准的 JSON 字典发给前端(比如用 model.model_dump())。不用手动去拼字典了。

字典是黏土: 想捏什么形状就捏什么形状,非常随意,但容易塌。

Pydantic 模型是模具: 规定了 Attraction 这个模型必须有一个字符串的 name,一个套娃的 Location。以后所有的数据,都必须倒进这个模具里,出来的都是标准件。

2.自底向上的数据模型设计

先定义最基础的模型,然后逐步组合成复杂的结构。

作为旅行计划助手,它的原子模型(基础模型)是位置,有位置才能规划。所以作者先设置了class Location,包含经纬度。

然后可以设置所有需要位置的:class Attraction(景点),class Meal(餐饮),class Hotel(酒店)。

接下来是不需要位置、和景点之类平级的:class Budget(预算),class WeatherInfo(天气信息,因为高德api给过来的数据类型不符还得额外处理一下)。

有了景点+餐饮+酒店+预算就可以规划一日的计划:class DayPlan

最后是整体的计划:class TripPlan,包含class DayPlan和class WeatherInfo

所以pydantic到底有什么用呢?

在前后端交界的大门处,立两块写满规则的铁牌子。进来的数据必须查,出去的数据也必须查。不合规的就地正法,绝不让脏数据在系统里乱跑。

4.多智能体协作设计(13.3)

为什么需要多智能体呢?

如果一个agent来干所有的活,他就会混乱,然后产生幻觉,而且它的prompt会非常非常长,难以维护。

再有一个原因,SimpleAgent 每次run()调用只能执行一个工具。如果考虑用ReactAgent,他的每一轮思考都需要调用 LLM,如果需要调用三个工具,就需要至少三轮思考,这意味着至少三次 LLM 调用。而且这些调用是串行的,必须等前一个完成才能开始下一个,总时间会很长。

所以最好的办法就是分工合作。

1.分工

旅行助手的分工是分了四个:景点搜索、天气查询、酒店推荐、行程规划。

分工原则:(自己总结版,以后有的话再补充吧)

  1. 买菜的不思考,做饭的不伸手:前3个Agent是买菜的,它们只要把菜(数据)买回来。最后一个行程规划是做饭的,它没有外部工具(手),就是要炒菜。
  2. 一个Agent只配发一把专属的武器:搜天气的和搜酒店的分开给,保证了工具调用 100% 不会出错。
  3. 不同数据的获取逻辑不同,分开设置获取:天气是确定性的(只要给个城市名就行);景点和酒店是偏好性的(需要根据用户的“历史文化”、“经济型”等偏好去搜索)。

下一步是详细设计每个 Agent 的角色和提示词:

设计提示词时,我们需要考虑几个关键问题:这个 Agent 需要什么输入?它应该产生什么输出?它需要调用什么工具?它可能遇到什么问题?

1:角色设定—— “你是谁”:你是景点搜索专家。

2:任务目标—— “你要干嘛”:根据用户偏好({preferences})搜索{city}的景点

3:工具使用指南—— “怎么买菜”(买菜专属):工具调用格式……正确示例……

4:输出格式—— “你怎么向我汇报” (炒菜专属):输出格式……

5:纪律/要求—— “绝对不能干什么/绝对要干什么”:**重要:**、**规划要求:**。用**大写加粗警告llm。

2.协作

现在让这四个agent一起工作:

__init__ :招募

plan_trip 方法:SOP,步骤

用 Python 写的先后顺序调用,

每个步骤的输出作为下一个步骤

的输入。

3.查询构建

PlannerAgent 需要整合所有信息,这个查询需要包含所有必要的信息,而且要组织得清晰有序,让 LLM 能够准确理解。

_build_planner_query :编写“呈送给厨师简报”:

“师傅,请根据以下信息生成计划:

客户要去:{request.city},预算是:{request.budget}。

这是景点查回来的真实资料:{attraction_response}

这是天气查回来的真实数据:{weather_response}

这是酒店查回来的真实空房:{hotel_response}

请师傅严格基于上述资料,不要自己瞎编,生成最终的 JSON 计划!”

信息的传递在这里完成了闭环:把多个小模型的输出,拼接成一个巨大无比的输入,喂给最终的大模型。

5.MCP工具集成(13.4)

别写requests.get,写service.py。

1.service.py

以高德地图的amap_service.py为例子:

_amap_mcp_tool = MCPTool(..., auto_expand=True)会自动跑去问外包(amap-mcp-server):“你能做什么?”外包公司就会交出一份标准说明书,里面写着功能和需求:比如maps_weather: 查天气,需要给我 city。

主程序拿到这份说明书,塞给llm看。

llm思考后决定:“我现在要查南京的天气”。它在聊天回复里吐出一个标准工单JSON:{"action": "call_tool", "tool_name": "maps_weather", "arguments": {"city": "南京"}}。这就像大模型写了一张纸条:“呼叫maps_weather的外包,参数是南京。”

这个时候,大堂经理(poi.py)拿到了大模型的纸条,把它递给了传达室(amap_service.py)。amap_service.py里面的result = self.mcp_tool.run({"action": "call_tool","tool_name":"maps_weather",……})就是:传达室大爷(amap_service.py)也不知道高德查天气的真实网址是什么,只是拿着一个对讲机(self.mcp_tool.run)喊外包执行 maps_weather,城市是南京。

最终干活的是@sugarforever/amap-mcp-server。

2.MCP共享

景点搜索、天气查询和路线规划三个agent都需要用到高德api,更好的做法是让所有 Agent 共享同一个MCPTool实例。这样只需要启动一个 MCP 服务器进程,所有的 API 调用都通过这个进程进行。这不仅节省资源,还可以更好地控制 API 调用频率。

相当于是整个后厨只买一个对讲机,谁用谁拿——amap_service.py 里的 global _amap_mcp_tool

这样,三个 Agent 都可以使用高德地图的 16 个工具,但底层只有一个 MCP 服务器进程在运行。三个 Agent 会依次调用工具,所有的请求都通过同一个 MCP 服务器发送到高德地图 API。

6.前端开发(13.5)

前后端分离。

后端只负责提供 API 接口,返回 JSON 格式的数据。

前端是一个独立的应用,通过 HTTP 请求调用后端 API,获取数据后渲染页面。

这种架构有几个明显的优势:

前端和后端可以独立开发、独立部署、独立测试;

前端可以是 Web 应用、移动应用或桌面应用,都使用同一套后端 API;

前端可以使用现代的框架和工具链,提供更好的用户体验。

在智能旅行助手项目中,后端是用 Python 和 FastAPI 实现的,提供了一个核心 API 接口POST /api/trip/plan,接收旅行需求,返回旅行计划。

前端是用 Vue 3 和 TypeScript 实现的,是一个单页应用(SPA),用户在浏览器中填写表单,点击"开始规划"按钮,前端发送 HTTP 请求到后端,等待响应,然后渲染结果页面。整个过程中,页面不会刷新,用户体验很流畅。

frontend的结构:

1.types 文件夹 (合同室) —— 包含index.ts

作用: 专门存放 TypeScript 的类型定义(interface)。

比喻: 餐厅的规章制度和菜谱标准。这里面只有规则,没有动作。它规定了“返回的数据必须包含城市名和天数”。

2.services 文件夹 (传达室) —— 包含api.ts

作用: 专门负责给后端(Python)发网络请求。

比喻: 餐厅的服务员。api.ts 里的代码会照着 types 里的合同标准,去敲后端的门,把后端的数据拿回来。这里通常写满了 axios.post('/api/trip/plan') 这种发请求的代码。

3.views 文件夹 (展示大堂) —— 包含Home.vue, Result.vue

作用: 真正给用户看和点的网页界面。

比喻: 餐厅的豪华包厢。

Home.vue:前台接待处(用户填城市的表单)。

Result.vue:菜品展示桌(拿到后端数据后,画出行程序列和地图)。

1.类型定义(13.5.2)

在前面3.数据模型设计里面,在后端使用 Pydantic 定义了数据模型,比如LocationAttractionDayPlanTripPlan等。在前端,我们需要定义对应的 TypeScript 类型。

写在frontend\src\types\index.ts

举例最基础的Location类型,表示经纬度坐标:

// frontend/src/types/index.ts
export interface Location {
  longitude: number
  latitude: number
}
后端 Python (Pydantic)  前端 TypeScript (界面)  说人话
class Location(BaseModel): export interface Location { 起名字(告诉别人我要定义一个新词了)
name: str name: string 文本(字母全小写)
price: int / score: float price: number 数字(不管整数小数,前端统叫 number)
is_free: bool is_free: boolean 真假(True/False 变成 true/false)
days: List[DayPlan] days: DayPlan[] 列表/数组(后面加个 [] 就代表一堆)
hotel: Optional[Hotel] hotel?: Hotel 字段名后面加个 ?,代表“可有可无

作用总结:

调用 API 时,TypeScript 会检查我们传递的数据是否符合TripPlanRequest类型。如果不对,TypeScript 会立即报错。

其次,当我们接收 API 响应时,TypeScript 会检查响应数据是否符合TripPlan类型。如果后端返回的数据结构发生变化,前端会立即发现。

最后,IDE 可以根据类型定义提供代码补全,我们输入tripPlan.时,IDE 会自动列出所有可用的字段。

2.api封装(13.5.3)

frontend\src\service\api.ts

使用 Axios 来发送 HTTP 请求,建一个 Axios 实例,配置了基础 URL、超时时间和请求头;

添加拦截器,拦截器可以在请求发送前和响应接收后执行一些通用逻辑,比如日志记录、错误处理、认证等;

定义 API 函数,这是前端调用后端的唯一入口。

如果说类型定义部分是签合同,api封装就是在前后端完成交易的快递员。

3.home表单设计(13.5.4)

表单就是餐厅前台递给顾客的那张点菜单。网页上所有能让用户打字(输入框)、打勾(复选框)、拉动(滑动条)的区域,组合起来就是一个表单。

设计好格子防止用户乱点菜+设计加载动画。

1 :准备空盒子

        代码: const formData = ref<TripPlanRequest>({ city: '', days: 3 ... })

        本质: 在前台准备一张空白的点菜单。

2 :定义提交动作

        代码: const handleSubmit = async () => { ... }

        本质: 规定前台服务员在顾客填完单子后该干嘛的SOP:

                1: 开启 Loading(告诉顾客后厨在做了)。

                2: 把 formData 交给上一节的快递员 generateTripPlan 送去后厨。

                3 (成功): 后厨把菜做好了,立刻跳转(router.push)到 Result 页面端给顾客看。

                4(失败): catch(error),弹出红字提示“生成失败”。

                5(兜底): finally,不管成功失败,把 Loading 动画关掉。

第 3 段:画 UI 界面

        代码:<template><a-form>, <a-input>。模板部分使用 Ant Design Vue 的组件

4.result展示(13.5.5)

设计展示页面

文字基础+高德美化+支持导出图片或pdf

代码见原文https://datawhalechina.github.io/hello-agents/#/./chapter13/%E7%AC%AC%E5%8D%81%E4%B8%89%E7%AB%A0%20%E6%99%BA%E8%83%BD%E6%97%85%E8%A1%8C%E5%8A%A9%E6%89%8B?id=_1355-result-%e9%a1%b5%e9%9d%a2%e5%b1%95%e7%a4%ba

7.功能实现详解(13.6)

这部分就直接去看原文吧https://datawhalechina.github.io/hello-agents/#/./chapter13/%E7%AC%AC%E5%8D%81%E4%B8%89%E7%AB%A0%20%E6%99%BA%E8%83%BD%E6%97%85%E8%A1%8C%E5%8A%A9%E6%89%8B?id=_136-%e5%8a%9f%e8%83%bd%e5%ae%9e%e7%8e%b0%e8%af%a6%e8%a7%a3

Logo

电影级数字人,免显卡端渲染SDK,十行代码即可调用,工业级demo免费开源下载!

更多推荐