深度实战：基于 LangGraph + MCP 的智能音乐推荐 Agent 全解析（从搭建到落地）

在 AI 与推荐系统结合的浪潮中，传统音乐推荐往往局限于“基于协同过滤”或“关键词匹配”，无法理解用户自然语言中的模糊需求（如“心情低落想听治愈系民谣”“运动时需要燃脂电子乐”）。而基于 LangGraph + MCP 构建的智能音乐推荐 Agent，恰好解决了这一痛点——它能听懂自然语言意图，通过标准化工具调用实现个性化推荐，还能输出可解释的推荐理由。

一、项目核心价值与技术选型深度解析

1. 项目定位与核心优势

自然语言意图理解：支持多维度需求（心情、场景、流派、相似歌曲、艺术家），无需用户手动筛选；
可解释推荐：每首歌附带推荐理由（如“节奏明快适合运动”“旋律治愈匹配低落心情”），提升用户信任度；
轻量化可扩展：基于 Streamlit 快速搭建 UI，支持本地数据与外部 API 切换，MCP 协议适配多工具；
技术栈灵活：LangGraph 编排工作流，支持替换 LLM（DeepSeek、Qwen、GPT 等），适配不同预算与场景。

2. 技术选型深层原因

技术组件	核心作用	选型深层原因
LangGraph	工作流编排与状态管理	支持复杂分支逻辑（多意图路由）、状态持久化，适合 Agent 任务拆解
LangChain	LLM 工具链与提示词管理	封装 LLM 调用、工具适配，降低与 MCP/外部 API 的集成成本
硅基流动（SiliconFlow）	LLM 服务提供	兼容 OpenAI API 格式，支持多模型切换，无需修改业务代码
Streamlit	轻量 Web UI 开发	快速迭代原型，支持组件化布局，无需前端框架（Vue/React）经验
MCP	工具调用标准化	统一对接音乐平台 API、本地工具，降低跨平台适配成本
Pydantic	数据校验与模型定义	确保 LLM 输出格式规范，避免推荐结果解析失败

3. 适用场景扩展

个人使用：本地部署打造专属音乐助手，支持自定义音乐库；
创业原型：快速验证 AI 推荐产品可行性，对接 Spotify/网易云 API 即可商业化；
企业内部工具：适配办公场景（如“会议室背景音乐推荐”“员工健身音乐合集”）；
教学/研究：学习 LangGraph 工作流编排、MCP 协议应用、意图识别技术。

二、快速上手：5 分钟跑通 + 常见问题速解

1. 环境准备

操作系统：Windows 10+/macOS 12+/Linux（Ubuntu 20.04+）；
Python 版本：3.9-3.11（推荐 3.10，避免版本兼容问题）；

核心依赖版本锁定（避免依赖冲突）：

langgraph==0.1.14
langchain==0.1.10
streamlit==1.32.2
openai==1.13.3
pydantic==2.6.1
aiohttp==3.9.3
python-dotenv==1.0.1

2. 完整部署步骤

步骤 1：克隆项目与安装依赖

# 克隆仓库（替换为实际仓库地址）
git clone https://github.com/imagist13/Muisc-Research.git
cd Muisc-Research

# 创建虚拟环境（推荐，避免依赖污染）
python -m venv venv
# 激活虚拟环境（Windows）
venv\Scripts\activate
# 激活虚拟环境（macOS/Linux）
source venv/bin/activate

# 安装依赖（指定版本避免冲突）
pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple

步骤 2：配置文件详解（setting.json 全字段说明）

{
  "SILICONFLOW_API_KEY": "你的硅基流动 API Key", // 从硅基流动控制台获取
  "SILICONFLOW_BASE_URL": "https://api.siliconflow.cn/v1", // 固定地址
  "SILICONFLOW_CHAT_MODEL": "Qwen/Qwen2.5-72B-Instruct", // 可选：DeepSeek/DeepSeek-MoE-16B-Instruct
  "SILICONFLOW_TEMPERATURE": 0.3, // 推荐 0.2-0.5，值越低推荐越稳定
  "TAILYAPI_API_KEY": "你的 Tavily 搜索 API Key", // 用于补充实时音乐数据（可选）
  "TAILYAPI_BASE_URL": "https://api.tavily.com",
  "APP_NAME": "智能音乐推荐助手", // 自定义应用名称
  "SPOTIFY_CLIENT_ID": "你的 Spotify Client ID", // 对接 Spotify 时必填
  "SPOTIFY_CLIENT_SECRET": "你的 Spotify Client Secret",
  "LOCAL_MUSIC_DB_PATH": "data/music_database.json", // 本地音乐数据库路径
  "CACHE_EXPIRE_SECONDS": 3600, // 推荐结果缓存时间（1小时）
  "MAX_RECOMMEND_NUM": 10, // 单次最大推荐歌曲数
  "INTENT_RECOGNITION_PROMPT_PATH": "prompts/intent_recognition.txt" // 意图识别提示词路径
}

步骤 3：启动应用与验证

# 启动 Streamlit 应用
python run_music_app.py
# 或直接启动（跳过检查脚本）
streamlit run music_app.py --server.port 8501

访问地址：http://localhost:8501
验证场景：
1. 输入“心情很好，推荐开心的流行乐”，查看推荐结果与理由；
2. 点击侧边栏“运动”快捷按钮，验证场景化推荐；
3. 输入“类似《晴天》的歌曲”，验证相似歌曲推荐。

3. 快速上手常见问题速解

问题现象	解决方案
启动报错“缺少 SILICONFLOW_API_KEY”	1. 硅基流动官网注册获取 API Key；2. 填入 setting.json
Streamlit 启动后无法访问	检查端口是否被占用，更换端口：`--server.port 8502`
推荐结果为空	1. 检查本地音乐数据库（data/music_database.json）是否有数据；2. 确保 LLM 模型配置正确
意图识别错误（如“运动”识别为“聊天”）	优化 prompts/intent_recognition.txt 中的提示词，增加意图示例

三、系统架构深度解析：模块交互与数据流转

1. 整体架构图（补充交互关系）

用户 → Streamlit UI（输入/展示） → MusicAgent 主入口 → LangGraph 工作流
                                      ↓
                          ┌─────────┬─────────┬─────────┬─────────┐
                          │  LLM 层  │ 工具层  │ 配置层  │ 数据层  │
                          │（意图识别）│（MCP/API）│（参数管理）│（本地/外部）│
                          └─────────┴─────────┴─────────┴─────────┘
                                      ↓
                          MCP Server → 外部工具（Spotify API/音乐搜索）

2. 核心模块职责与交互细节

（1）配置层（config/）：统一参数管理

核心文件：settings_loader.py
核心逻辑：
1. 优先级：setting.json > 环境变量 > 默认值；
2. 支持动态加载：修改配置后无需重启应用，下次请求自动生效；
3. 关键功能：参数校验（如 API Key 格式、模型名称合法性）。

代码解析：

def load_settings():
  # 加载 JSON 配置
  with open("setting.json", "r", encoding="utf-8") as f:
      json_config = json.load(f)
  # 合并环境变量（覆盖 JSON 配置）
  env_config = {
      key: os.getenv(key) for key in json_config if os.getenv(key) is not None
  }
  final_config = {**json_config, **env_config}
  # 参数校验
  validate_settings(final_config)
  return final_config

（2）LLM 层（llms/）：模型抽象与适配

核心文件：siliconflow_llm.py
设计思路：
1. 兼容 OpenAI API 格式，替换模型时无需修改业务代码；
2. 封装重试逻辑（API 超时、限流时自动重试）；
3. 支持模型调优参数（temperature、max_tokens、top_p）。

关键代码：

class SiliconFlowLLM:
  def __init__(self, api_key, base_url, model_name, temperature=0.3):
      self.client = OpenAI(
          api_key=api_key,
          base_url=base_url
      )
      self.model_name = model_name
      self.temperature = temperature

  def chat_completion(self, messages):
      # 重试逻辑（最多 3 次）
      for _ in range(3):
          try:
              response = self.client.chat.completions.create(
                  model=self.model_name,
                  messages=messages,
                  temperature=self.temperature,
                  response_format={"type": "json_object"}  # 强制 JSON 输出
              )
              return json.loads(response.choices[0].message.content)
          except Exception as e:
              log.error(f"LLM 调用失败：{e}")
              time.sleep(1)
      raise Exception("LLM 调用多次失败，请检查 API Key 或网络")

（3）工作流层（graphs/）：LangGraph 状态与路由

核心概念：
- 状态（State）：存储用户输入、意图类型、推荐结果、错误信息；
- 节点（Node）：封装具体逻辑（意图识别、搜索、推荐、结果生成）；
- 路由（Route）：根据意图类型分发到对应节点。

状态定义代码：

from langgraph.graph import StateGraph, MessagesState
from pydantic import BaseModel

# 自定义状态（扩展 MessagesState）
class MusicAgentState(MessagesState):
  intent: str = None  # 意图类型（recommend_by_mood/activity/genre 等）
  recommendations: list = []  # 推荐结果列表
  error: str = None  # 错误信息

核心节点实现：

def analyze_intent(state: MusicAgentState) -> MusicAgentState:
  """意图识别节点：从用户输入中提取意图"""
  user_input = state.messages[-1].content
  # 调用 LLM 识别意图
  llm = SiliconFlowLLM(**settings)
  prompt = load_prompt("prompts/intent_recognition.txt")
  messages = [
      {"role": "system", "content": prompt},
      {"role": "user", "content": user_input}
  ]
  result = llm.chat_completion(messages)
  state.intent = result["intent"]
  state.error = result.get("error")
  return state

def route_by_intent(state: MusicAgentState) -> str:
  """路由节点：根据意图分发到对应处理节点"""
  if state.error:
      return "handle_error"
  match state.intent:
      case "recommend_by_mood":
          return "recommend_by_mood"
      case "recommend_by_activity":
          return "recommend_by_activity"
      case "search_music":
          return "search_music"
      case "general_chat":
          return "general_chat"
      case _:
          return "handle_unknown_intent"

工作流构建：

def build_music_graph() -> StateGraph:
  graph = StateGraph(MusicAgentState)
  # 添加节点
  graph.add_node("analyze_intent", analyze_intent)
  graph.add_node("recommend_by_mood", recommend_by_mood)
  graph.add_node("recommend_by_activity", recommend_by_activity)
  graph.add_node("search_music", search_music)
  graph.add_node("generate_result", generate_result)
  graph.add_node("handle_error", handle_error)
  # 定义边（路由逻辑）
  graph.set_entry_point("analyze_intent")
  graph.add_conditional_edges("analyze_intent", route_by_intent)
  # 所有处理节点最终指向结果生成
  graph.add_edge("recommend_by_mood", "generate_result")
  graph.add_edge("recommend_by_activity", "generate_result")
  graph.add_edge("search_music", "generate_result")
  graph.add_edge("general_chat", "generate_result")
  graph.add_edge("handle_error", "generate_result")
  graph.add_edge("handle_unknown_intent", "generate_result")
  return graph.compile()

（4）工具层（tools/）：MCP 适配与外部调用

工具层是连接 Agent 与外部服务的核心：

工具分类：
- 本地工具：本地音乐搜索、相似歌曲匹配（基于 TF-IDF 或余弦相似度）；
- MCP 工具：对接外部音乐 API（Spotify、网易云）、音乐评分查询。

MCP Server 开发（音乐搜索工具示例）：

# mcp/music_search_server.py
from mcp.server import Server, Tool
from mcp.types import InputSchema, OutputSchema
from pydantic import BaseModel

# 输入/输出模型
class MusicSearchInput(BaseModel):
  keyword: str  # 搜索关键词（歌曲名/艺术家）
  genre: str = None  # 流派筛选（可选）

class MusicSearchOutput(BaseModel):
  songs: list[dict]  # 搜索结果：[{"title": "", "artist": "", "album": ""}]

# 工具实现
def search_music(input_data: MusicSearchInput) -> MusicSearchOutput:
  # 对接本地数据库或外部 API
  with open(settings["LOCAL_MUSIC_DB_PATH"], "r", encoding="utf-8") as f:
      music_db = json.load(f)
  # 筛选逻辑
  results = [
      song for song in music_db
      if input_data.keyword.lower() in song["title"].lower() 
      and (input_data.genre is None or song["genre"] == input_data.genre)
  ]
  return MusicSearchOutput(songs=results[:10])

# 注册 MCP Server
server = Server()
search_tool = Tool(
  name="music_search",
  description="根据关键词和流派搜索音乐",
  input_schema=InputSchema(model=MusicSearchInput),
  output_schema=OutputSchema(model=MusicSearchOutput),
  handler=search_music
)
server.register_tool(search_tool)

# 启动 MCP Server
if __name__ == "__main__":
  import uvicorn
  uvicorn.run(server.fastapi_app, host="0.0.0.0", port=8000)

Agent 调用 MCP 工具代码：

def search_music(state: MusicAgentState) -> MusicAgentState:
  """搜索节点：调用 MCP 工具搜索音乐"""
  user_input = state.messages[-1].content
  # 提取搜索关键词（通过 LLM 解析）
  llm = SiliconFlowLLM(**settings)
  prompt = "提取用户输入中的音乐搜索关键词和流派（如无则为 None），返回 JSON：{\"keyword\": \"\", \"genre\": \"\"}"
  messages = [{"role": "user", "content": f"{prompt}\n用户输入：{user_input}"}]
  search_params = llm.chat_completion(messages)
  # 调用 MCP 工具
  mcp_client = Client(base_url="http://localhost:8000")
  try:
      response = mcp_client.call_tool("music_search", search_params)
      state.recommendations = response.content[0].text["songs"]
  except Exception as e:
      state.error = f"搜索失败：{str(e)}"
  return state

（5）前端层（music_app.py）：交互体验优化

核心组件：
1. 侧边栏：快捷按钮（心情/场景/流派）、配置状态显示；
2. 主区域：用户输入框、推荐结果卡片（含歌曲信息、推荐理由）、历史记录；
3. 结果卡片：使用 Streamlit 的 card 组件，支持点击播放链接。

关键代码：

def render_recommendation_cards(recommendations):
  """渲染推荐结果卡片"""
  for idx, song in enumerate(recommendations, 1):
      with st.card(f"推荐歌曲 #{idx}"):
          col1, col2 = st.columns([1, 2])
          with col1:
              st.image(song.get("cover_url", "default_cover.png"), use_column_width=True)
          with col2:
              st.subheader(song["title"])
              st.text(f"艺术家：{song['artist']}")
              st.text(f"流派：{song['genre']}")
              st.text(f"推荐理由：{song['reason']}")
              # 播放链接（对接 Spotify 时生效）
              if song.get("spotify_url"):
                  st.link_button("Spotify 播放", song["spotify_url"])

四、核心流程：从意图识别到推荐结果的全链路拆解

1. 核心时序图（用户输入“运动时需要燃脂电子乐”）

用户 → 输入需求 → Streamlit UI → 提交请求 → MusicAgent
                          ↓
MusicAgent → 调用 LangGraph 工作流 → 分析意图节点（analyze_intent）
                          ↓
LLM → 识别意图为“recommend_by_activity” → 路由节点（route_by_intent）
                          ↓
推荐节点（recommend_by_activity） → 调用 MCP 工具搜索“燃脂电子乐”
                          ↓
外部 API/本地数据库 → 返回符合条件的歌曲列表 → 结果生成节点（generate_result）
                          ↓
LLM → 为每首歌生成推荐理由 → Streamlit UI → 渲染结果卡片

2. 关键环节详解

（1）意图识别：提示词优化技巧

意图识别的准确性直接影响推荐效果，推荐使用“示例引导+格式约束”的提示词：

# prompts/intent_recognition.txt
你是音乐推荐 Agent 的意图识别助手，需从用户输入中提取意图类型，返回 JSON 格式（必须包含 intent 字段）。
支持的意图类型：
1. recommend_by_mood：用户提到心情（如开心、悲伤、治愈）；
2. recommend_by_activity：用户提到场景/活动（如运动、工作、睡觉）；
3. recommend_by_genre：用户提到音乐流派（如民谣、电子乐、摇滚）；
4. recommend_by_artist：用户提到艺术家/歌手（如周杰伦、Taylor Swift）；
5. recommend_by_similar：用户提到“类似/像XX歌曲”；
6. search_music：用户明确搜索歌曲/专辑；
7. general_chat：不涉及音乐推荐的闲聊。

示例：
用户输入：“心情低落想听治愈系民谣” → {"intent": "recommend_by_mood", "mood": "治愈"}
用户输入：“运动时需要燃脂的电子乐” → {"intent": "recommend_by_activity", "activity": "运动", "genre": "电子乐"}
用户输入：“类似《晴天》的歌曲” → {"intent": "recommend_by_similar", "song": "晴天"}

注意：
- 若无法识别意图，intent 设为 "unknown"；
- 额外提取相关参数（如 mood、activity、genre），无则留空；
- 严格返回 JSON，不添加额外描述。

（2）推荐逻辑：结合用户需求与音乐特征

推荐节点的核心是“筛选符合条件的歌曲 + 个性化排序”，示例代码：

def recommend_by_activity(state: MusicAgentState) -> MusicAgentState:
    """根据活动场景推荐音乐"""
    # 提取活动类型和额外参数
    user_input = state.messages[-1].content
    llm = SiliconFlowLLM(**settings)
    prompt = "提取用户输入中的活动类型和音乐偏好（如流派、节奏），返回 JSON：{\"activity\": \"\", \"genre\": \"\", \"tempo\": \"\"}"
    params = llm.chat_completion([{"role": "user", "content": f"{prompt}\n{user_input}"}])
    activity = params["activity"]
    genre = params.get("genre")
    tempo = params.get("tempo", "fast")  # 默认快节奏（运动场景）

    # 从本地数据库/外部 API 筛选歌曲
    with open(settings["LOCAL_MUSIC_DB_PATH"], "r", encoding="utf-8") as f:
        music_db = json.load(f)
    # 筛选逻辑：活动匹配 + 流派匹配 + 节奏匹配
    filtered_songs = [
        song for song in music_db
        if song["activity"] == activity 
        and (genre is None or song["genre"] == genre)
        and song["tempo"] == tempo
    ]
    # 排序：按流行度降序
    filtered_songs.sort(key=lambda x: x["popularity"], reverse=True)
    state.recommendations = filtered_songs[:settings["MAX_RECOMMEND_NUM"]]
    return state

（3）结果生成：可解释性优化

推荐结果需包含“歌曲信息 + 推荐理由”，让用户理解“为什么推荐这首歌”：

def generate_result(state: MusicAgentState) -> MusicAgentState:
    """生成可解释的推荐结果"""
    if state.error:
        result = f"推荐失败：{state.error}"
    elif not state.recommendations:
        result = "未找到符合条件的歌曲，可尝试调整需求描述～"
    else:
        # 生成推荐理由
        llm = SiliconFlowLLM(**settings)
        prompt = load_prompt("prompts/generate_result.txt")
        songs_str = json.dumps([{"title": s["title"], "artist": s["artist"], "genre": s["genre"]} for s in state.recommendations])
        messages = [
            {"role": "system", "content": prompt},
            {"role": "user", "content": f"用户需求：{state.messages[-1].content}\n推荐歌曲：{songs_str}"}
        ]
        result = llm.chat_completion(messages)
        # 合并歌曲信息与推荐理由
        for song, reason in zip(state.recommendations, result["reasons"]):
            song["reason"] = reason

    # 添加到状态，用于前端展示
    state.messages.append({"role": "assistant", "content": json.dumps(state.recommendations)})
    return state

五、进阶实战：MCP + Spotify API 对接全流程

1. Spotify 开发者账号准备

步骤 1：创建应用

访问 Spotify for Developers，注册账号并创建应用；
记录 Client ID 和 Client Secret（填入 setting.json）；
在“Redirect URIs”中添加 http://localhost:8501/callback（用于授权）。

步骤 2：获取访问令牌（Authorization Code Flow）

Spotify API 需授权才能调用，核心流程：

前端引导用户授权，获取 Authorization Code；
后端用 Code 兑换 Access Token 和 Refresh Token；
用 Access Token 调用 API，过期后用 Refresh Token 刷新。

2. MCP Server 对接 Spotify API

步骤 1：开发 Spotify 音乐搜索 MCP Server

# mcp/spotify_search_server.py
from mcp.server import Server, Tool
from mcp.types import InputSchema, OutputSchema
from pydantic import BaseModel
import spotipy
from spotipy.oauth2 import SpotifyOAuth

# 输入/输出模型
class SpotifySearchInput(BaseModel):
    keyword: str
    genre: str = None
    limit: int = 10

class SpotifySearchOutput(BaseModel):
    songs: list[dict] = []

# Spotify 客户端初始化
def init_spotify_client():
    return spotipy.Spotify(
        auth_manager=SpotifyOAuth(
            client_id=settings["SPOTIFY_CLIENT_ID"],
            client_secret=settings["SPOTIFY_CLIENT_SECRET"],
            redirect_uri="http://localhost:8501/callback",
            scope="user-read-private user-read-email playlist-modify-private"
        )
    )

# 工具实现
def spotify_search(input_data: SpotifySearchInput) -> SpotifySearchOutput:
    try:
        sp = init_spotify_client()
        # 搜索歌曲
        query = f"{input_data.keyword} {'genre:' + input_data.genre if input_data.genre else ''}"
        results = sp.search(q=query, type="track", limit=input_data.limit)
        # 解析结果
        songs = []
        for track in results["tracks"]["items"]:
            songs.append({
                "title": track["name"],
                "artist": ", ".join([a["name"] for a in track["artists"]]),
                "album": track["album"]["name"],
                "cover_url": track["album"]["images"][0]["url"],
                "spotify_url": track["external_urls"]["spotify"],
                "popularity": track["popularity"],
                "tempo": "fast" if track["tempo"] > 120 else "slow"
            })
        return SpotifySearchOutput(songs=songs)
    except Exception as e:
        log.error(f"Spotify 搜索失败：{e}")
        raise Exception(f"Spotify API 调用失败：{str(e)}")

# 注册工具
server = Server()
spotify_tool = Tool(
    name="spotify_search",
    description="通过 Spotify API 搜索音乐，支持关键词和流派筛选",
    input_schema=InputSchema(model=SpotifySearchInput),
    output_schema=OutputSchema(model=SpotifySearchOutput),
    handler=spotify_search
)
server.register_tool(spotify_tool)

# 启动 MCP Server
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(server.fastapi_app, host="0.0.0.0", port=8001)

步骤 2：Agent 调用 Spotify MCP 工具

修改 tools/music_tools.py，添加 MCP 客户端调用：

def search_music(state: MusicAgentState) -> MusicAgentState:
    """调用 Spotify MCP 工具搜索音乐"""
    # 解析搜索参数
    user_input = state.messages[-1].content
    llm = SiliconFlowLLM(**settings)
    params = llm.chat_completion([{"role": "user", "content": f"提取搜索关键词和流派：{user_input}"}])
    # 调用 MCP Server
    mcp_client = Client(base_url="http://localhost:8001")
    try:
        response = mcp_client.call_tool("spotify_search", {
            "keyword": params["keyword"],
            "genre": params.get("genre"),
            "limit": settings["MAX_RECOMMEND_NUM"]
        })
        state.recommendations = response.content[0].text["songs"]
    except Exception as e:
        state.error = f"搜索失败：{str(e)}"
    return state

步骤 3：前端添加 Spotify 播放功能

修改 music_app.py，支持点击播放和歌单创建：

def render_spotify_features(song):
    """渲染 Spotify 专属功能"""
    col1, col2 = st.columns(2)
    with col1:
        st.link_button("立即播放", song["spotify_url"], type="primary")
    with col2:
        if st.button("添加到歌单"):
            # 调用 Spotify API 创建歌单
            try:
                sp = init_spotify_client()
                playlist = sp.user_playlist_create(
                    user=sp.current_user()["id"],
                    name="AI 推荐歌单",
                    public=False
                )
                sp.playlist_add_items(playlist["id"], [song["spotify_url"].split("/")[-1]])
                st.success("已添加到歌单！")
            except Exception as e:
                st.error(f"添加失败：{str(e)}")

六、性能优化与生产级部署

1. 性能优化技巧

（1）缓存策略：减少重复调用

对相同意图和参数的请求，缓存推荐结果（使用 Redis 或本地缓存）：

def add_cache_middleware(func):
    """缓存装饰器"""
    cache = {}
    @wraps(func)
    def wrapper(state: MusicAgentState):
        # 生成缓存键（用户输入 + 意图）
        cache_key = f"{state.messages[-1].content}_{state.intent}"
        if cache_key in cache and time.time() - cache[cache_key]["timestamp"] < settings["CACHE_EXPIRE_SECONDS"]:
            state.recommendations = cache[cache_key]["data"]
            return state
        # 无缓存则执行原函数
        result = func(state)
        # 存入缓存
        cache[cache_key] = {
            "data": state.recommendations,
            "timestamp": time.time()
        }
        return result
    return wrapper

# 为推荐节点添加缓存
@add_cache_middleware
def recommend_by_activity(state: MusicAgentState) -> MusicAgentState:
    # 原有逻辑不变
    pass

（2）模型调优：平衡速度与效果

降低模型参数量：测试环境用 Qwen2.5-7B，生产环境根据预算选择 14B/72B；
调整温度参数：推荐场景用 0.2-0.3，避免推荐结果波动；
限制上下文长度：只保留最近 3 轮对话，减少 LLM 推理负担。

（3）异步处理：提升并发能力

使用 asyncio 异步调用 LLM 和 MCP 工具，支持多用户同时请求：

async def async_recommend_by_mood(state: MusicAgentState) -> MusicAgentState:
    """异步推荐节点"""
    loop = asyncio.get_event_loop()
    # 异步调用 LLM
    llm_task = loop.run_in_executor(None, llm.chat_completion, messages)
    # 异步调用 MCP 工具
    mcp_task = loop.run_in_executor(None, mcp_client.call_tool, "music_search", params)
    # 等待结果
    llm_result, mcp_result = await asyncio.gather(llm_task, mcp_task)
    # 处理结果
    state.recommendations = mcp_result.content[0].text["songs"]
    return state

2. 生产级部署方案

（1）Docker 容器化部署

创建 Dockerfile 和 docker-compose.yml，简化部署流程：

# Dockerfile
FROM python:3.10-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple
COPY . .
EXPOSE 8501
CMD ["python", "run_music_app.py"]

# docker-compose.yml
version: "3"
services:
  music-agent:
    build: .
    ports:
      - "8501:8501"
    environment:
      - SILICONFLOW_API_KEY=your_key
      - SPOTIFY_CLIENT_ID=your_id
      - SPOTIFY_CLIENT_SECRET=your_secret
    volumes:
      - ./data:/app/data
    restart: always
  mcp-server:
    build: ./mcp
    ports:
      - "8001:8001"
    restart: always

（2）云服务部署（以阿里云为例）

创建 ECS 实例（推荐 2C4G 及以上）；
安装 Docker 和 Docker Compose；
克隆项目，配置 setting.json；
执行 docker-compose up -d 启动服务；
配置安全组，开放 8501 端口；
访问 http://ECS公网IP:8501 即可使用。

（3）监控与日志

添加 Prometheus + Grafana 监控，或使用 Streamlit 的内置日志功能：

def setup_logging():
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
        handlers=[
            logging.FileHandler("logs/music_agent.log"),
            logging.StreamHandler()
        ]
    )

七、扩展方向与未来规划

1. 功能扩展

多模态推荐：结合歌词情感分析、封面图风格识别，提升推荐精准度；
用户偏好学习：存储用户历史反馈（喜欢/不喜欢），用向量数据库（如 Pinecone）构建个性化模型；
多平台对接：除 Spotify 外，支持网易云音乐、QQ 音乐 API；
语音交互：集成 Whisper 实现语音输入，支持“语音说需求，语音听推荐”。

2. 技术升级

多 Agent 协同：拆分意图识别、推荐、结果生成到不同 Agent，提升灵活性；
模型微调：用音乐推荐数据集微调 LLM，提升意图识别和推荐理由生成的准确性；
边缘部署：适配嵌入式设备（如智能音箱），支持离线推荐。

八、总结：智能推荐 Agent 的核心价值与落地建议

基于 LangGraph + MCP 的智能音乐推荐 Agent，核心价值在于“用自然语言理解打破推荐壁垒，用标准化工具调用实现生态扩展”。它不仅是一个 Demo 级项目，更是一套可复用的 Agent 构建框架——你可以将其适配到其他推荐场景（如电影、书籍、商品），只需替换工具层和数据层。

落地建议：

初期用本地数据验证流程，再对接外部 API；
优先优化意图识别和推荐逻辑，再扩展 MCP 工具和多平台集成；
小流量测试后，逐步添加缓存、监控和部署优化；
关注用户反馈，通过“反馈闭环”持续优化推荐效果。

除非注明，否则均为李锋镝的博客原创文章，转载必须以链接形式标明本文链接
本文链接：https://www.lifengdi.com/ren-gong-zhi-neng/4569

本作品采用知识共享署名-非商业性使用-相同方式共享 4.0 国际许可协议进行许可