深度解析LangChain天气Agent空数据问题：从工具调用到API响应的全链路排查

当开发者尝试构建基于LangChain的天气查询Agent时，最令人沮丧的莫过于看到Agent返回空数据或无效响应。这种情况往往发生在工具调用链路的某个环节，但具体原因可能隐藏在从工具注册到API响应的任何一个步骤中。本文将带您深入LangChain Agent的内部工作机制，揭示导致天气查询失败的常见陷阱，并提供一套可立即落地的诊断方案。

1. 工具注册失败的典型场景与诊断

工具是LangChain Agent与外部世界交互的桥梁。一个天气查询工具若未正确注册，Agent将无法调用它获取数据。以下是工具注册环节最常见的三类问题：

1.1 工具描述不清晰导致LLM无法识别

LangChain中的工具通过description字段向LLM说明其用途。模糊或不准确的描述会导致LLM无法在适当的情境下选择该工具。对比以下两种描述方式：

# 不推荐的模糊描述
@tool
def weather_tool(query: str):
    """查询天气的工具"""
    pass

# 推荐的具体描述
@tool
def get_current_weather(location: str) -> str:
    """当用户询问某地当前或未来天气时使用此工具。输入应为明确的地理位置名称，如'北京'或'纽约'。"""
    pass

诊断方法：

1. 检查工具描述是否明确包含：
   • 使用场景（"当...时使用"）
   • 预期输入格式
   • 返回内容类型
2. 通过agent.tools属性验证工具是否已加载
3. 使用agent.run("工具列表")查看LLM对工具功能的理解

1.2 工具参数定义与LLM输出不匹配

LangChain Agent在调用工具时，LLM需要生成符合工具参数要求的JSON。常见问题包括：

• 参数名称不一致（工具定义city但LLM输出location）
• 参数类型不匹配（工具需要str但LLM提供int）
• 缺少必需参数

# 工具定义
@tool
def weather_search(city_name: str, date: str = None) -> str:
    pass

# LLM可能生成的错误调用
{
    "action": "weather_search",
    "action_input": {"location": "北京"}  # 参数名不匹配
}

解决方案：

1. 使用Pydantic模型明确定义参数：

from pydantic import BaseModel, Field

class WeatherInput(BaseModel):
    city_name: str = Field(..., description="城市中文名称，如'北京'")
    date: str = Field(None, description="可选日期，格式YYYY-MM-DD")

@tool(args_schema=WeatherInput)
def weather_search(city_name: str, date: str = None) -> str:
    pass

2. 在工具描述中明确参数要求

1.3 多工具冲突与优先级问题

当Agent加载多个工具时，LLM可能错误选择非天气工具来处理天气查询。例如，同时存在weather_api和web_search工具时，LLM可能选择后者。

优化策略：

1. 工具命名具有明确区分度（如get_current_weather vs search_web）
2. 使用return_direct=True让特定工具直接返回结果
3. 通过tool_choice参数强制指定工具（OpenAI函数调用风格）

from langchain.agents import Tool

tools = [
    Tool(
        name="get_current_weather",
        func=fetch_weather,
        description="专用于查询实时天气数据",
        return_direct=True
    ),
    # 其他工具...
]

2. API响应处理中的常见陷阱

即使工具被正确调用，API响应处理不当仍会导致空数据。以下是三个关键故障点：

2.1 城市编码映射失效

大多数天气API需要城市代码而非名称。本地映射文件（如city.json）问题会导致查询失败：

# 原始代码中的潜在风险
cities = json.load(open('city.json'))  # 文件路径？编码？键名？
city_code = cities.get(city)  # 键不存在时返回None

强化方案：

1. 使用更健壮的编码获取方式：

from typing import Dict

def load_city_codes() -> Dict[str, str]:
    try:
        with open('city.json', 'r', encoding='utf-8') as f:
            return json.load(f)
    except Exception as e:
        logger.error(f"加载城市编码失败: {e}")
        return {}

def get_city_code(city: str, city_db: Dict[str, str]) -> str:
    # 支持简繁体、常见别名
    aliases = {
        "北京": ["北京市", "京城", "北平"],
        "上海": ["上海市", "沪"]
    }
    city = city.strip()
    if city in city_db:
        return city_db[city]
    
    for code, names in aliases.items():
        if city in names:
            return city_db.get(code, "")
    return ""

2. 添加备用API（如同时支持和风天气与心知天气）

2.2 API响应格式验证不足

不同天气API返回格式差异很大，未做充分验证会导致解析失败：

API提供商	状态字段	成功值	数据路径
和风天气	code	"200"	data.now
心知天气	status	"ok"	results[0]
OpenWeather	cod	200	main

健壮性改造：

def parse_weather_response(response, api_type="hefeng"):
    if not response.ok:
        return f"API请求失败: {response.status_code}"
    
    data = response.json()
    
    # 验证不同API的成功状态
    if api_type == "hefeng" and data.get("code") != "200":
        return f"和风API错误: {data.get('message', '未知错误')}"
    elif api_type == "xinzhi" and data.get("status") != "ok":
        return f"心知API错误: {data.get('status', '未知错误')}"
    
    # 分API类型解析
    try:
        if api_type == "hefeng":
            return format_hefeng_data(data)
        elif api_type == "xinzhi":
            return format_xinzhi_data(data)
    except KeyError as e:
        logger.error(f"解析响应数据失败: {e}")
        return "天气数据解析失败"

2.3 网络请求缺乏容错机制

简单的requests.get()调用在面对网络波动或API限流时会直接抛出异常。

生产级改进：

import requests
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10),
    retry=retry_if_exception_type(
        (requests.exceptions.Timeout, 
         requests.exceptions.ConnectionError)
    )
)
def safe_api_call(url, params=None, timeout=5):
    try:
        response = requests.get(
            url,
            params=params,
            timeout=timeout,
            headers={"User-Agent": "MyWeatherAgent/1.0"}
        )
        response.raise_for_status()
        return response
    except requests.exceptions.RequestException as e:
        logger.warning(f"API请求异常: {e}")
        raise

3. LLM指令解析的优化策略

LLM在Agent工作流程中负责解析用户输入、选择工具并生成调用参数。这一环节的常见问题包括：

3.1 城市名称提取不准确

用户可能输入"北京天气怎么样？"或"我在上海，需要带伞吗？"等多样化表达。简单的提示词可能无法可靠提取位置。

增强型提示工程：

def build_location_extraction_prompt(query: str) -> str:
    return f"""请从以下用户输入中提取地理位置信息：
    
输入：{query}

请按以下规则处理：
1. 只返回明确的地理位置名称（如"北京市"、"上海"）
2. 忽略非地理相关词汇（如"天气"、"预报"）
3. 对简称和别称进行标准化（如"帝都"→"北京"）
4. 若无明确位置，返回空字符串

提取结果："""

多级位置校验：

1. 先用LLM提取候选位置
2. 与城市编码库比对
3. 如不匹配，让LLM进行确认：

def confirm_location(candidate, city_db):
    if candidate in city_db:
        return candidate
    
    prompt = f"""候选位置'{candidate}'未匹配到城市编码。请从以下选项中选择最接近的：
    
{list(city_db.keys())[:10]}...

或回复'无匹配'。你的选择："""
    # 调用LLM进行确认...

3.2 复杂查询的渐进式处理

对于"北京明天下午的降雨概率"这类复杂查询，需要分解为多个步骤：

1. 提取位置和时间
2. 确定是否需要特定时间段的预报
3. 调整API查询参数

示例流程：

def handle_complex_query(query):
    # 第一步：提取时空信息
    time_pattern = r"(今天|明天|后天|\d+月\d+日)"
    loc_pattern = r"(北京|上海|广州|深圳)"
    
    # 第二步：确定API查询范围
    if "降雨概率" in query:
        return get_precipitation_probability(location, date)
    elif "温度" in query:
        return get_temperature_range(location, date)
    else:
        return get_general_forecast(location, date)

3.3 记忆组件的合理运用

ConversationBufferMemory可以帮助Agent记住对话上下文，但也可能导致工具调用混乱。

最佳实践：

from langchain.memory import ConversationBufferWindowMemory

memory = ConversationBufferWindowMemory(
    k=3,  # 只保留最近3轮对话
    memory_key="chat_history",
    input_key="input",
    output_key="output"
)

agent = initialize_agent(
    tools,
    llm,
    agent=AgentType.CONVERSATIONAL_REACT_DESCRIPTION,
    memory=memory,
    verbose=True,
    handle_parsing_errors=True
)

4. 全链路诊断Checklist与实战调试

当天气Agent返回空数据时，按照以下步骤系统排查：

4.1 诊断流程图

[开始]
  │
  ▼
[检查工具是否加载] → 未加载 → 检查工具注册代码
  │ ✓
  ▼
[验证LLM能否选择正确工具] → 错误选择 → 优化工具描述
  │ ✓
  ▼
[检查工具参数生成] → 参数错误 → 调整提示词或使用Pydantic
  │ ✓
  ▼
[API请求是否发出] → 无请求 → 检查网络和API密钥
  │ ✓
  ▼
[API响应状态码] → 非200 → 检查API文档
  │ ✓
  ▼
[响应数据解析] → 解析失败 → 验证JSON路径
  │ ✓
  ▼
[最终输出格式] → 格式不符 → 调整输出模板
  │ ✓
  ▼
[结束]

4.2 日志增强方案

在关键节点添加详细日志：

import logging
from functools import wraps

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('agent_debug.log'),
        logging.StreamHandler()
    ]
)

def log_tool_call(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        logger.info(f"调用工具 {func.__name__}，参数: {kwargs}")
        try:
            result = func(*args, **kwargs)
            logger.info(f"工具返回: {result[:200]}...")  # 截断长响应
            return result
        except Exception as e:
            logger.error(f"工具调用异常: {e}", exc_info=True)
            raise
    return wrapper

@log_tool_call
@tool
def weather_search(city: str):
    pass

4.3 单元测试策略

为每个关键组件编写测试用例：

import pytest
from unittest.mock import patch

def test_city_code_mapping():
    assert get_city_code("北京") == "101010100"
    assert get_city_code("上海市") == "101020100"
    assert get_city_code("广州") == "101280101"

@patch('requests.get')
def test_weather_api_success(mock_get):
    mock_response = type('', (), {'ok': True, 'json': lambda: {"code": "200"}})()
    mock_get.return_value = mock_response
    
    result = weather_search("北京")
    assert "北京" in result
    mock_get.assert_called_once()

def test_agent_weather_flow():
    agent = create_test_agent()
    response = agent.run("北京今天天气如何？")
    assert "天气" in response
    assert "北京" in response
    assert not "抱歉" in response  # 确保没有错误信息

4.4 真实案例：和风天气API集成

完整示例代码：

from langchain.agents import AgentType, initialize_agent
from langchain.tools import Tool
from langchain_community.llms import Tongyi
import requests
from pydantic import BaseModel, Field
from typing import Optional

class WeatherInput(BaseModel):
    location: str = Field(..., description="城市中文名称")
    date: Optional[str] = Field(None, description="日期，格式YYYY-MM-DD")

def get_hefeng_weather(location: str, date: str = None) -> str:
    """调用和风天气API获取天气数据"""
    base_url = "https://devapi.qweather.com/v7/weather/now"
    params = {
        "key": "YOUR_API_KEY",
        "location": get_city_code(location),
        "lang": "zh"
    }
    
    try:
        response = requests.get(base_url, params=params, timeout=10)
        data = response.json()
        
        if data["code"] != "200":
            return f"天气查询失败: {data.get('message', '未知错误')}"
            
        current = data["now"]
        return (
            f"{location}当前天气: {current['text']}\n"
            f"温度: {current['temp']}°C (体感{current['feelsLike']}°C)\n"
            f"风向: {current['windDir']} {current['windScale']}级\n"
            f"湿度: {current['humidity']}%"
        )
    except Exception as e:
        return f"天气查询异常: {str(e)}"

weather_tool = Tool.from_function(
    func=get_hefeng_weather,
    name="get_current_weather",
    description="查询指定城市当前天气情况",
    args_schema=WeatherInput
)

llm = Tongyi(temperature=0)
agent = initialize_agent(
    [weather_tool],
    llm,
    agent=AgentType.OPENAI_FUNCTIONS,
    verbose=True
)

# 测试查询
response = agent.run("上海现在天气怎么样？")
print(response)

关键优化点：

1. 使用Pydantic严格定义输入模型
2. 清晰的工具描述引导LLM正确使用
3. 详细的错误处理和用户友好输出
4. 结构化响应提升可读性