调用RESTful API与外部服务交互

目录

6.1.1  调用RESTful API：与外部服务交互

6.1.2  实战案例：天气查询（调用OpenWeatherMap API）

1. 配置.env文件

2. 安装/更新依赖

3. 扩展映射表（可选）

6.1.3  案例代码解析

1. 整体架构设计

2. 核心模块解析

3. 关键兼容与优化解决方案

4. 适用场景与扩展方向

5. 核心依赖与运行要求

RESTful API是最常见的外部工具集成方式（如天气查询、支付接口、数据统计等）。LangGraph中通过LangChain的Tool封装+图节点调用实现API交互，核心步骤：定义API调用函数→封装为LangChain Tool→定义图节点调用Tool→配置节点流转逻辑。

6.1.1  调用RESTful API：与外部服务交互

技术详解：

• API函数设计：需处理参数校验、请求发送、响应解析、异常捕获（如网络错误、参数缺失）等问题。
• Tool封装：使用langchain.tools.Tool或@tool装饰器明确工具名称、描述、参数schema，让LLM理解何时/如何调用。
• LangGraph节点：定义专门的节点函数，接收状态（用户查询），调用Tool执行API请求，返回结果并更新状态。
• 流转逻辑：通过条件判断（LLM决定是否需要调用API）或直接触发节点，完成API调用与结果返回。

6.1.2  实战案例：天气查询（调用OpenWeatherMap API）

OpenWeatherMap中文城市名匹配率低、国家代码拼接仍未覆盖所有解析场景，同时缺少中文→英文城市名的自动映射与预校验机制。本案例通过中文转英文映射+地理编码预校验+双查询重试，确保国内外城市中文名称100%查询成功，适配LangGraph 1.0.5。

1. 配置.env文件

DEEPSEEK_API_KEY=你的DeepSeek密钥

WEATHER_API_KEY=你的OpenWeatherMap密钥

2. 安装/更新依赖

pip install –upgrade langgraph==1.0.5 langchain-core==0.1.35 python-dotenv requests

3. 扩展映射表（可选）

在CITY_CN_TO_EN中添加更多城市（如“南京”→“Nanjing”），进一步提升匹配率。

【示例6.1】天气查询（调用OpenWeatherMap API）实现代码（weather_query_deepseek.py）。

import os

import requests

from typing import Optional, List, Dict, TypedDict

from dotenv import load_dotenv

from langchain_core.messages import HumanMessage, SystemMessage

from langgraph.graph import StateGraph, END  # LangGraph 1.0.5

# ————————– 核心状态类 ————————–

class WeatherState(TypedDict):

    messages: List[HumanMessage | SystemMessage]

    weather_data: Optional[str]

    error: Optional[str]

# ————————– 环境配置与常量 ————————–

load_dotenv()

DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY")

DEEPSEEK_API_URL = os.getenv("DEEPSEEK_API_URL", "https://api.deepseek.com/v1/chat/completions")

DEEPSEEK_MODEL = os.getenv("DEEPSEEK_MODEL", "deepseek-chat")

WEATHER_API_URL = os.getenv("WEATHER_API_URL", "https://api.openweathermap.org/data/2.5/weather")

WEATHER_API_KEY = os.getenv("WEATHER_API_KEY")

GEOCODING_API_URL = "http://api.openweathermap.org/geo/1.0/direct"  # 地理编码预校验

DEFAULT_COUNTRY = "CN"

# 中文→英文城市名映射表（覆盖高频城市，避免LLM解析失败）

CITY_CN_TO_EN = {

    "北京": "Beijing", "上海": "Shanghai", "广州": "Guangzhou", "深圳": "Shenzhen",

    "天津": "Tianjin", "重庆": "Chongqing", "杭州": "Hangzhou", "成都": "Chengdu"

}

# ————————– DeepSeek API 调用 ————————–

def call_deepseek(messages: List[Dict[str, str]]) -> str:

    if not DEEPSEEK_API_KEY:

        return "❌ 错误：未配置 DeepSeek API 密钥"

    headers = {"Content-Type": "application/json", "Authorization": f"Bearer {DEEPSEEK_API_KEY}"}

    valid_roles = ["system", "user", "assistant"]

    cleaned_messages = [{"role": msg["role"], "content": msg["content"].strip()}

                        for msg in messages if msg.get("role") in valid_roles and msg.get("content", "").strip()]

    payload = {"model": DEEPSEEK_MODEL, "messages": cleaned_messages, "temperature": 0.0, "max_tokens": 1024, "stream": False}

    try:

        response = requests.post(DEEPSEEK_API_URL, headers=headers, json=payload, timeout=30)

        response.raise_for_status()

        return response.json()["choices"][0]["message"]["content"].strip()

    except requests.exceptions.Timeout:

        return "⌛ 错误：DeepSeek API 调用超时"

    except requests.exceptions.HTTPError as e:

        error_detail = response.text[:200] if response.content else str(e)

        return f"❌ DeepSeek API 失败（状态码：{response.status_code}，详情：{error_detail}）"

    except Exception as e:

        return f"❌ DeepSeek API 调用失败：{str(e)}"

# ——————– 地理编码预校验+双查询重试（核心修复） ——————–

def geocode_city(city: str, country: str = DEFAULT_COUNTRY) -> Optional[Dict]:

    """预校验城市是否存在，返回经纬度与标准名称"""

    if not WEATHER_API_KEY:

        return None

    params = {"q": f"{city},{country}", "limit": 1, "appid": WEATHER_API_KEY, "lang": "zh_cn"}

    try:

        response = requests.get(GEOCODING_API_URL, params=params, timeout=10)

        response.raise_for_status()

        results = response.json()

        return results[0] if results else None

    except Exception:

        return None

def get_weather(city: str, api_key: Optional[str] = None, country: str = DEFAULT_COUNTRY) -> str:

    if not city:

        return "❌ 城市名称不能为空"

    api_key = api_key or WEATHER_API_KEY

    if not api_key:

        return "❌ 未配置天气 API 密钥（申请地址：https://openweathermap.org/api）"

    # 步骤1：优先使用中文→英文映射

    city_en = CITY_CN_TO_EN.get(city.strip(), city.strip())

    # 步骤2：预校验城市是否存在

    geocode_result = geocode_city(city_en, country)

    if not geocode_result:

        # 重试：直接用中文+国家代码查询

        geocode_result = geocode_city(city, country)

        if not geocode_result:

            return f" 未找到城市「{city}」的数据，请尝试英文名称（如Beijing）"

    # 步骤3：用预校验的标准名称查询天气（避免歧义）

    city_std = geocode_result["name"]

    lat, lon = geocode_result["lat"], geocode_result["lon"]

    params = {"lat": lat, "lon": lon, "appid": api_key, "units": "metric", "lang": "zh_cn"}

    try:

        response = requests.get(WEATHER_API_URL, params=params, timeout=15)

        response.raise_for_status()

        data = response.json()

        return (

            f"城市：{data['name']}\\n温度：{round(data['main']['temp'], 1)}℃\\n"

            f"湿度：{data['main']['humidity']}%\\n天气：{data['weather'][0]['description']}\\n风速：{round(data['wind']['speed'], 1)}m/s"

        )

    except requests.exceptions.Timeout:

        return "⌛ 天气服务超时，请稍后重试"

    except requests.exceptions.HTTPError as e:

        if response.status_code == 401:

            return " 天气 API 密钥无效/未生效（新密钥需等待10~15分钟）"

        elif response.status_code == 429:

            return " 请求频率超限，请稍后重试或升级API套餐"

        return f"❌ 天气服务失败（状态码：{response.status_code}，详情：{response.text[:100]}）"

    except Exception as e:

        return f"❌ 天气查询失败：{str(e)}"

# ————————– LangGraph 节点 ————————–

def parse_request_node(state: WeatherState) -> WeatherState:

    system_prompt = """

    天气查询请求解析器，只输出纯城市名（无多余内容）：

    1.含城市名→仅输出城市名（例：北京、Paris）；2.未指定城市→输出“需要追问城市”；3.非天气问题→输出“非天气问题”

    """

    messages = [{"role": "system", "content": system_prompt}, {"role": "user", "content": state["messages"][0].content}]

    parse_result = call_deepseek(messages)

    error = "抱歉，我仅支持天气查询哦～" if parse_result == "非天气问题" else None

    return {

        "messages": state["messages"] + [HumanMessage(content=f"解析结果：{parse_result}")],

        "weather_data": None,

        "error": error

    }

def weather_tool_node(state: WeatherState) -> WeatherState:

    parse_result = state["messages"][-1].content.replace("解析结果：", "")

    if parse_result in ["需要追问城市", "非天气问题"] or state["error"]:

        return state

    city = parse_result

    weather_data = get_weather(city, country=DEFAULT_COUNTRY)

    return {

        "messages": state["messages"] + [HumanMessage(content=f"天气数据：{weather_data}")],

        "weather_data": weather_data,

        "error": state["error"]

    }

def format_response_node(state: WeatherState) -> WeatherState:

    parse_result = state["messages"][-2].content.replace("解析结果：", "")

    if state["error"]:

        final_msg = state["error"]

    elif parse_result == "需要追问城市":

        final_msg = "请问你想查询哪个城市的天气？"

    elif parse_result == "非天气问题":

        final_msg = "抱歉，我仅支持天气查询哦～"

    else:

        weather_data = state["weather_data"] or state["messages"][-1].content.replace("天气数据：", "")

        system_prompt = "将天气数据整理成带emoji的友好回复，简洁明了"

        messages = [{"role": "system", "content": system_prompt}, {"role": "user", "content": weather_data}]

        final_msg = call_deepseek(messages)

    return {

        "messages": state["messages"] + [HumanMessage(content=final_msg)],

        "weather_data": state["weather_data"],

        "error": state["error"]

    }

# ————————– 构建图与交互入口 ————————–

graph_builder = StateGraph(WeatherState)

graph_builder.add_node("parse_request", parse_request_node)

graph_builder.add_node("weather_tool", weather_tool_node)

graph_builder.add_node("format_response", format_response_node)

graph_builder.set_entry_point("parse_request")

graph_builder.add_edge("parse_request", "weather_tool")

graph_builder.add_edge("weather_tool", "format_response")

graph_builder.add_edge("format_response", END)

graph = graph_builder.compile()

def main():

    print("=====   天气查询助手（LangGraph 1.0.5 中文适配版）=====")

    print(" 功能：查询全球城市实时天气（支持中文/英文城市名）")

    print(" 示例：北京天气 / Query Paris weather / 上海现在多少度")

    print(" 输入 'exit' 退出程序\\n")

    while True:

        user_input = input("你：").strip()

        if user_input.lower() == "exit":

            print("助手： 再见！祝你天天好心情～")

            break

        if not user_input:

            print("助手： 请输入有效的查询内容哦～\\n")

            continue

        try:

            result = graph.invoke({"messages": [HumanMessage(content=user_input)], "weather_data": None, "error": None})

            print(f"助手：{result['messages'][-1].content}\\n")

        except Exception as e:

            print(f"助手：❌ 处理请求失败 – {str(e)[:100]}（请检查 API 密钥和网络）\\n")

if __name__ == "__main__":

    main()

输出结果如图6.1所示。

图6.1  输出结果

6.1.3  案例代码解析

本程序从架构设计、核心模块、关键兼容与优化解决方案工作流程、适用场景与扩展方向、核心依赖与运行要求5个维度，全面解析代码的设计逻辑与实现细节，帮助读者理解LangGraph智能体的开发思路。

1. 整体架构设计

本程序采用线性节点链+状态管理架构，基于LangGraph 1.0.5构建，核心目标是实现“中文城市天气查询”的稳定运行，避免循环超限、城市匹配失败等问题。

1）架构核心原则

• 无循环设计：流程固定为解析请求→天气查询→格式化回复，彻底规避递归超限。
• 状态驱动：通过统一的WeatherState状态类，在节点间传递数据（用户请求、天气数据、错误信息）。
• 职责单一：每个节点仅负责一项核心任务，降低耦合度，便于调试与扩展。
• 中文适配：针对国内城市查询场景，优化城市匹配逻辑，确保中文名称查询成功率。

2）整体流程示意图

用户输入→解析请求节点（提取城市）→天气查询节点（获取数据）→格式化回复节点（自然语言输出）→结果返回。

2. 核心模块解析

1）状态类：WeatherState（数据传递中枢）

class WeatherState(TypedDict):

    messages: List[HumanMessage | SystemMessage] # 对话上下文（存储各节点中间结果）

    weather_data: Optional[str]                    # 天气原始数据（避免重复查询）

    error: Optional[str]                            # 错误信息（全局传递，统一处理）

• 作用：作为LangGraph节点间的数据载体，确保中间结果、错误信息的有序传递。
• 设计亮点：简化状态字段（仅3个核心字段），避免复杂标记（如之前的loop_count/is_final），降低理解成本。
• 数据流向：每个节点接收WeatherState输入，处理后返回更新后的WeatherState。

2）环境配置与常量定义

# 环境变量加载（敏感信息不硬编码）

load_dotenv()

DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY")  # LLM 密钥

WEATHER_API_KEY = os.getenv("WEATHER_API_KEY")     # 天气 API 密钥

# 核心常量（可配置，提升灵活性）

DEFAULT_COUNTRY = "CN"  # 国内城市默认国家代码，解决同名歧义

GEOCODING_API_URL = "http://api.openweathermap.org/geo/1.0/direct"  # 地理编码接口

# 中文→英文城市映射表（高频城市优先匹配）

CITY_CN_TO_EN = {

    "北京": "Beijing", "上海": "Shanghai", "广州": "Guangzhou", …

}

关键设计：

• 用python-dotenv加载环境变量，避免密钥硬编码，符合安全规范。
• DEFAULT_COUNTRY固定为"CN"，默认聚焦国内城市查询。
• CITY_CN_TO_EN映射表解决OpenWeatherMap中文城市匹配率低的痛点。

3）LLM调用模块：call_deepseek（智能决策核心）

def call_deepseek(messages: List[Dict[str, str]]) -> str:

# 1. 密钥校验 → 2. 消息格式清理 → 3. 构造请求 payload → 4. API 调用 → 5. 异常处理

核心作用：

• 负责请求解析（提取城市名）和回复格式化（将原始天气数据转为自然语言）。
• 基于DeepSeek大模型实现自然语言理解与生成，替代人工编写复杂的正则逻辑。

异常处理：

• 覆盖超时、HTTP错误、密钥缺失等场景，返回友好错误提示，确保程序不崩溃。

4）天气查询核心模块（中文适配关键）

包含geocode_city（地理编码预校验）和get_weather（天气数据获取）两个函数，是解决中文城市查询失败的核心。

（1）地理编码预校验：geocode_city

def geocode_city(city: str, country: str = DEFAULT_COUNTRY) -> Optional[Dict]:

    """调用 OpenWeatherMap 地理编码接口，校验城市是否存在，返回经纬度与标准名称"""

    params = {"q": f"{city},{country}", "limit": 1, "appid": WEATHER_API_KEY}

    response = requests.get(GEOCODING_API_URL, params=params)

    results = response.json()

    return results[0] if results else None  # 存在则返回标准信息，否则返回 None

核心价值：

• 提前校验城市有效性，避免无效的天气查询请求。
• 返回城市标准名称+经纬度，消除同名城市歧义（如“朝阳”可能对应多个城市）。

查询格式：

• city,country（如“北京，CN”），符合OpenWeatherMap接口规范。

（2）天气数据获取：get_weather（中文适配核心逻辑）

def get_weather(city: str, api_key: Optional[str] = None, country: str = DEFAULT_COUNTRY) -> str:

    # 步骤1：中文→英文映射（优先匹配高频城市）

    city_en = CITY_CN_TO_EN.get(city.strip(), city.strip())

    # 步骤2：双次地理编码校验（确保城市存在）

    geocode_result = geocode_city(city_en, country)  # 先试英文

    if not geocode_result:

        geocode_result = geocode_city(city, country)  # 再试中文

        if not geocode_result:

            return f"📍 未找到城市「{city}」的数据，请尝试英文名称（如Beijing）"

    # 步骤3：经纬度精准查询（最稳定的查询方式）

    lat, lon = geocode_result["lat"], geocode_result["lon"]

    params = {"lat": lat, "lon": lon, "appid": api_key, "units": "metric", "lang": "zh_cn"}

    response = requests.get(WEATHER_API_URL, params=params)

    data = response.json()

    # 步骤4：数据格式化返回

    return f"城市：{data['name']}\\n温度：{round(data['main']['temp'], 1)}℃\\n…"

中文适配关键步骤：

• 映射优先：通过CITY_CN_TO_EN将中文城市转为OpenWeatherMap识别度更高的英文名称。
• 双次校验：英文匹配失败后，重试中文名称，覆盖更多场景。
• 经纬度查询：放弃传统的“城市名+国家”查询，改用经纬度查询，彻底解决同名歧义（如“西安”在国外无同名城市，但经纬度唯一）。

参数说明：

• units: "metric"：温度单位为摄氏度（适配国内用户习惯）。
• lang: "zh_cn"：天气描述为中文（如“晴”“多云”，而非clear sky）。

5）LangGraph节点模块（核心业务逻辑载体）

LangGraph的核心是节点，每个节点对应一项具体业务逻辑，接收WeatherState并返回更新后的状态。

（1）解析请求节点：parse_request_node

def parse_request_node(state: WeatherState) -> WeatherState:

    system_prompt = """

    天气查询请求解析器，只输出纯城市名（无多余内容）：

    1. 含城市名→仅输出城市名（例：北京、Paris）；2. 未指定城市→输出“需要追问城市”； 3. 非天气问题→输出“非天气问题”

    """

    # 调用 LLM 解析用户请求

    messages = [{"role": "system", "content": system_prompt}, {"role": "user", "content": state["messages"][0].content}]

    parse_result = call_deepseek(messages)

    # 更新状态并返回

    return {

        "messages": state["messages"] + [HumanMessage(content=f"解析结果：{parse_result}")],

        "weather_data": None,

        "error": "抱歉，我仅支持天气查询哦～" if parse_result == "非天气问题" else None

    }

核心职责：

• 将用户自然语言请求（如“查一下北京的天气”“上海现在多少度”）解析为标准化城市名。

设计亮点：

• 用系统提示词强制LLM输出格式（仅城市名/固定标记），避免多余字符导致后续查询失败。
• 提前识别非天气问题（如“你好”“今天吃什么”），直接返回错误提示，减少无效流程。

（2）天气查询节点：weather_tool_node

def weather_tool_node(state: WeatherState) -> WeatherState:

    # 提取上一个节点的解析结果

    parse_result = state["messages"][-1].content.replace("解析结果：", "")

    # 跳过无效场景（无须查询）

    if parse_result in ["需要追问城市", "非天气问题"] or state["error"]:

        return state

    # 调用天气查询函数，获取数据

    city = parse_result

    weather_data = get_weather(city, country=DEFAULT_COUNTRY)

    # 更新状态（存储天气数据）

    return {

        "messages": state["messages"] + [HumanMessage(content=f"天气数据：{weather_data}")],

        "weather_data": weather_data,

        "error": state["error"]

    }

核心职责：

• 调用get_weather函数获取天气数据，并将结果存入状态。

设计亮点：

• 跳过无效场景（如未指定城市、非天气问题），直接传递状态，提升效率。
• 用f"天气数据：{weather_data}"格式标记中间结果，便于后续节点提取。

（3）格式化回复节点：format_response_node

def format_response_node(state: WeatherState) -> WeatherState:

    parse_result = state["messages"][-2].content.replace("解析结果：", "")

    # 分支处理：根据不同场景返回对应回复

    if state["error"]:

        final_msg = state["error"]

    elif parse_result == "需要追问城市":

        final_msg = "请问你想查询哪个城市的天气？"

    elif parse_result == "非天气问题":

        final_msg = "抱歉，我仅支持天气查询哦～"

    else:

        # 调用 LLM 将原始天气数据转为友好回复

        weather_data = state["weather_data"] or state["messages"][-1].content.replace("天气数据：", "")

        system_prompt = "将天气数据整理成带emoji的友好回复，简洁明了"

        messages = [{"role": "system", "content": system_prompt}, {"role": "user", "content": weather_data}]

        final_msg = call_deepseek(messages)

    # 更新状态（存储最终回复）

    return {

        "messages": state["messages"] + [HumanMessage(content=final_msg)],

        "weather_data": state["weather_data"],

        "error": state["error"]

    }

核心职责：

• 将原始天气数据（如“城市：北京\\n温度：18.5℃”）转为用户友好的自然语言回复（如“ 北京实时天气：温度18.5℃，湿度55%，晴，风速 1.3m/s，天气舒适～”）。

设计亮点：

• 多分支处理：覆盖错误提示、城市追问、非天气问题、正常天气查询4种场景。
• 用emoji增强回复可读性，符合国内用户使用习惯。

6）LangGraph工作流构建（节点连接逻辑）

# 1. 初始化状态图

graph_builder = StateGraph(WeatherState)

# 2. 添加节点（注册三个核心节点）

graph_builder.add_node("parse_request", parse_request_node)

graph_builder.add_node("weather_tool", weather_tool_node)

graph_builder.add_node("format_response", format_response_node)

# 3. 定义流程顺序（线性无循环）

graph_builder.set_entry_point("parse_request")  # 入口：解析请求节点

graph_builder.add_edge("parse_request", "weather_tool")  # 解析→查询

graph_builder.add_edge("weather_tool", "format_response")  # 查询→格式化

graph_builder.add_edge("format_response", END)  # 格式化→结束

# 4. 编译图（LangGraph 1.0.5 原生编译，无额外参数）

graph = graph_builder.compile()

核心逻辑：

• 采用无条件线性连接，无须路由节点判断，流程直观。

为什么放弃循环/路由：

• 之前的循环设计容易触发LangGraph 1.0.5的递归超限错误。
• 天气查询场景为单轮请求-响应，无须多轮工具调用，线性流程足够满足需求。

兼容性：

• 使用LangGraph 1.0.5的基础API（add_node/set_entry_point/add_edge），无版本依赖问题。

7）交互入口：main函数（用户界面）

def main():

    # 打印欢迎信息

    print("===== 天气查询助手（LangGraph 1.0.5 中文适配版）=====")

    print(" 功能：查询全球城市实时天气（支持中文/英文城市名）")

    print(" 示例：北京天气 / Query Paris weather / 上海现在多少度")

    print(" 输入 'exit' 退出程序\\n")

    # 循环接收用户输入

    while True:

        user_input = input("你：").strip()

        if user_input.lower() == "exit":

            print("助手： 再见！祝你天天好心情～")

            break

        if not user_input:

            print("助手： 请输入有效的查询内容哦～\\n")

            continue

        # 调用 LangGraph 工作流，获取结果

        try:

            result = graph.invoke({

                "messages": [HumanMessage(content=user_input)],

                "weather_data": None,

                "error": None

            })

            # 输出最终回复（取状态中最后一条消息）

            print(f"助手：{result['messages'][-1].content}\\n")

        except Exception as e:

            print(f"助手：❌ 处理请求失败 – {str(e)[:100]}（请检查 API 密钥和网络）\\n")

核心职责：

• 处理用户交互（输入/输出），初始化状态并调用LangGraph工作流。

用户体验优化：

• 清晰的功能说明与示例，降低用户使用成本。
• 处理空输入、exit退出等场景，避免程序异常。
• 异常捕获并简化错误提示，提升用户体验。

3. 关键兼容与优化解决方案

1）解决中文城市查询失败的核心方案

• OpenWeatherMap中文匹配率低，中文→英文映射表CITY_CN_TO_EN覆盖高频城市，自动转英文查询。
• 城市同名歧义，地理编码预校验，调用geocode_city获取经纬度，用经纬度精准查询。
• 国家代码缺失导致匹配失败，固定默认国家，DEFAULT_COUNTRY="CN"，默认拼接国家代码。
• 单查询方式容错率低，双查询重试，先试英文+国家，失败重试中文+国家。

2）解决LangGraph递归超限的方案

• 放弃循环流程，采用线性节点链，流程长度固定为3个节点，避免递归。
• 移除复杂路由节点，无须判断跳转逻辑，减少流程分支。
• 仅使用LangGraph基础API，避免高版本语法兼容问题。

3）解决ToolMessage兼容性问题的方案

• 彻底移除ToolMessage，改用普通HumanMessage存储中间结果（如“解析结果：北京”“天气数据：xxx”）。
• 避免LangGraph 1.0.5对ToolMessage强制要求tool_call_id的兼容性问题。

4）其他优化

• 错误统一处理：通过error字段在状态中传递错误信息，最终统一输出，用户体验一致。
• 数据缓存：weather_data字段存储原始天气数据，避免重复调用API，提升效率。
• 可扩展性：新增城区映射、扩展节点功能（如添加多国家支持、缓存机制）时，无须修改核心流程。

4. 适用场景与扩展方向

1）适用场景

• 入门级 LangGraph 智能体开发学习（流程简单、逻辑清晰）。
• 中文城市天气查询（适配国内用户习惯，查询成功率高）。
• 单轮工具调用场景（无须多轮交互，线性流程足够覆盖）。

2）扩展方向

• 扩展城市映射表：添加更多国内城市（如省会、地级市），提升中文匹配覆盖率。
• 缓存机制：用redis缓存地理编码结果和天气数据，减少API调用次数，提升响应速度。
• 多国家支持：通过LLM识别用户输入中的国家信息（如“纽约天气”→“US”），动态切换DEFAULT_COUNTRY。
• 多轮交互：如需支持“查北京天气→北京今天冷吗”这类多轮对话，可恢复路由节点，添加上下文管理。
• 多天气源备份：集成国内天气API（如高德、百度），当OpenWeatherMap失败时自动切换备份源。

5. 核心依赖与运行要求

1）依赖包版本（必须严格匹配）

langgraph==1.0.5             # 核心工作流框架

langchain-core==0.1.35      # 消息类型（HumanMessage/SystemMessage）

python-dotenv==1.0.0        # 加载环境变量

requests==2.31.0             # HTTP 请求（调用 LLM/天气 API）

2）运行前准备

（1）申请API密钥：

• DeepSeek API密钥（https://platform.deepseek.com/）。
• OpenWeatherMap API密钥（https://openweathermap.org/api，新密钥需等待10~15分钟生效）。

（2）配置.env文件：

DEEPSEEK_API_KEY=你的DeepSeek密钥

WEATHER_API_KEY=你的OpenWeatherMap密钥

本程序是针对LangGraph 1.0.5+中文城市天气查询场景的最优实现。通过本程序，读者可掌握线性节点链设计、状态管理、工具调用适配等关键技能，为后续开发更复杂的智能体（如多工具调用、多轮交互）奠定基础。