前言

在大模型应用开发中，LangGraph 作为 LangChain 生态内的工作流编排利器，可以轻松实现多步骤、有状态、可循环的智能体流程。本文将带你从零开发一个真实天气查询智能体：
通过智谱 GLM-4.5-Air 生成查询计划，利用 LangGraph 分步执行，调用高德天气官方 API获取实时气温，完成一套可落地、可扩展的 LLM 自动化流程。

---

一、技术栈介绍

• LangGraph：智能体工作流编排，状态管理、条件分支、循环执行
• LangChain：大模型调用、提示词模板、结构化输出
• 智谱AI ChatZhipuAI：使用 glm-4.5-air 生成结构化计划
• 高德地图天气 API：提供国内城市真实、稳定的实时天气数据
• Python：整体开发语言

二、整体流程设计

本项目采用计划-执行双节点智能体架构：

1. 生成计划节点：LLM 生成四个直辖市天气查询步骤
2. 执行查询节点：依次调用天气 API，更新状态
3. 条件路由：判断是否还有未完成计划，自动循环/结束

流程图：
START → 生成计划 → 执行单步查询 →（还有计划？）→ 继续执行 / END

三、环境准备

1. 安装依赖

pip install langgraph langchain langchain-core langchain-community pydantic requests

2. 获取 API Key

1. 智谱AI Key
   前往：https://open.bigmodel.cn/ 获取 API Key
2. 高德 Web 服务 Key
   前往：https://lbs.amap.com/ → 控制台 → 应用管理 → 创建应用 → 添加 Key（选择 Web 服务）

四、完整代码实现

1. 导入库与基础配置

import operator
import os
import requests
from langchain_core.messages import AnyMessage, HumanMessage, AIMessage
from langgraph.graph import START, END, StateGraph
from langchain_core.runnables import RunnableConfig
from langchain_community.chat_models import ChatZhipuAI
from langchain_core.prompts import ChatPromptTemplate
from typing import TypedDict, List, Annotated
from pydantic import BaseModel, Field

2. 配置 API 信息

# 智谱AI配置
os.environ["ZHIPUAI_API_KEY"] = "你的智谱API Key"
model = ChatZhipuAI(
    temperature=0.1,
    model="glm-4.5-air"
)

# 高德天气API配置
GAODE_KEY = "你的高德Web服务Key"

3. 定义状态与数据结构

# 智能体状态
class MessageState(TypedDict):
    title: str
    plans: List[str]           # 待执行计划
    finished: List[str]       # 已完成计划
    messages: Annotated[List[AnyMessage], operator.add]

# 结构化输出：计划列表
class Plans(BaseModel):
    plans: List[str] = Field(description="天气查询步骤")

4. 封装真实天气 API

def get_real_weather(city_name: str) -> str:
    """调用高德天气API，获取真实气温"""
    try:
        url = "https://restapi.amap.com/v3/weather/weatherInfo"
        params = {
            "city": city_name,
            "key": GAODE_KEY,
            "extensions": "base",
            "output": "json"
        }
        resp = requests.get(url, params=params, timeout=10)
        data = resp.json()
        
        if data.get("status") == "1" and data.get("lives"):
            live = data["lives"][0]
            return f"{live['weather']}，气温 {live['temperature']}℃"
    except Exception as e:
        print(f"查询{city_name}失败：{e}")
    return "天气获取失败"

5. 构建 LangGraph 节点

节点1：生成查询计划

def make_plan_node(state: MessageState):
    prompt = ChatPromptTemplate.from_messages([
        ("system", "只输出4个计划：查北京天气、查上海天气、查天津天气、查重庆天气"),
        ("user", "{title}")
    ])
    chain = prompt | model.with_structured_output(schema=Plans)
    resp = chain.invoke({"title": state["title"]})
    plans = resp.plans if resp else ["查北京天气","查上海天气","查天津天气","查重庆天气"]
    return {"plans": plans}

节点2：执行单步天气查询

def run_plan_node(state: MessageState):
    plans = state["plans"]
    if not plans:
        return {"plans": []}
    
    current = plans[0]
    city = current.replace("查","").replace("天气","").strip()
    weather = get_real_weather(city)
    
    return {
        "plans": plans[1:],
        "finished": state["finished"] + [current],
        "messages": [AIMessage(content=f"{city}：{weather}")]
    }

路由函数

def should_continue(state):
    return END if len(state["plans"]) == 0 else "run_plan_node"

6. 编译并运行智能体

# 构建流程图
builder = StateGraph(MessageState)
builder.add_node("make_plan_node", make_plan_node)
builder.add_node("run_plan_node", run_plan_node)

builder.add_edge(START, "make_plan_node")
builder.add_edge("make_plan_node", "run_plan_node")
builder.add_conditional_edges("run_plan_node", should_continue)

graph = builder.compile()

# 执行
if __name__ == "__main__":
    out = graph.invoke({
        "title": "查询四个直辖市天气",
        "plans": [],
        "finished": [],
        "messages": [HumanMessage(content="开始查询真实天气")],
    })

    print("\n===== 天气查询结果 =====")
    for msg in out["messages"]:
        print(msg.content)

五、运行效果

===== 天气查询结果 =====
开始查询真实天气
北京：晴，气温 14℃
上海：多云，气温 19℃
天津：晴，气温 12℃
重庆：阴，气温 17℃

六、常见问题与排坑

1. No module named ‘langgraph’
   pip install langgraph
2. jwt.encode 报错
   pip uninstall jwt && pip install pyjwt
3. 第三方天气接口不通
   优先使用高德、和风天气等官方正规 API，不要依赖免费小接口
4. 智能体死循环/消息重复
   节点只返回需要更新的字段，不要覆盖整个 state

七、扩展方向

• 支持任意城市天气查询
• 增加多轮重试、异常处理机制
• 接入工具调用（ToolCall）自动选择查询城市
• 构建多智能体协作：计划者 + 执行者 + 总结者
• 部署为 API 服务，提供前端调用

八、总结

本文通过LangGraph + 智谱AI + 高德天气API，完成了一个标准的企业级 LLM 自动化智能体开发流程。

• LangGraph 状态管理与条件路由
• 大模型结构化输出（Pydantic）
• 真实第三方 API 接入
• 智能体工作流设计思想

---

这套架构可直接迁移到：自动化报告、批量数据查询、RAG 流程、多步骤任务执行等场景。

• 博客园
• 公众号 行走之飞鱼