智能天气助手实战：基于 openJiuwen Agent Core 的 ReAct Agent 开发全流程

一、项目背景与价值

在数字化时代，天气信息已成为人们日常决策的重要依据。传统的天气应用多停留在数据展示层面，而智能天气助手则通过 AI 技术实现了从数据提供到智能服务的跃迁。本项目基于 openJiuwen Agent Core 框架，构建了一个能够理解自然语言、提供个性化建议、具备智能决策能力的天气助手。

项目地址：https://atomgit.com/openJiuwen/agent-core

二、项目环境搭建

环境要求

• Python 3.11+
• pip 包管理器

安装依赖

# 克隆项目
git clone https://atomgit.com/openJiuwen/agent-core
cd agent-core

# 安装依赖
pip install -e .

# 额外依赖
pip install python-dotenv requests

配置环境变量

在项目根目录创建 .env 文件：

API_BASE=https://api.deepseek.com/v1
API_KEY=your_api_key_here
MODEL_NAME=deepseek-v4-flash
MODEL_PROVIDER=openai

三、核心技术栈与架构

openJiuwen 框架选择

选择 openJiuwen Agent Core 作为开发框架，主要基于以下考虑：

• ReAct 架构支持：支持推理与行动的智能体模式
• 工具集成能力：灵活的外部 API 集成机制
• 对话管理：强大的上下文理解与管理能力
• 可扩展性：模块化设计便于功能扩展

ReAct 架构详解

ReAct（Reasoning and Acting）架构是本项目的核心，它允许智能体在推理（制定计划、进行逻辑思考）和行动（与外部工具交互）之间交替进行。这种架构特别适合需要外部知识和工具的复杂任务。

在智能天气助手中，ReAct 架构的工作流程如下：

1. 观察（Observation）：接收用户输入，如"明天杭州天气如何？"
2. 推理（Thought）：分析用户意图，确定需要调用天气 API
3. 行动（Action）：调用 weather_reporter 工具查询天气
4. 观察（Observation）：获取天气 API 返回的结果
5. 推理（Thought）：分析天气数据，准备响应
6. 最终响应（Final Response）：向用户提供天气信息及建议

# ReAct 智能体的核心工作流程示意图
def invoke(self, input_data):
    # 初始化对话历史（由 Session + Checkpointer 自动管理）
    # 循环直到任务完成
    for iteration in range(MAX_ITERATIONS):
        # 使用 LLM 生成下一步动作
        response = self.model.generate(history)

        if self._is_tool_call(response):
            # 执行工具调用
            tool_result = self._execute_tool(response)
            # 将工具结果添加到历史
            history.append({"role": "tool", "content": tool_result})
        else:
            # 返回最终结果
            return {"output": response}

在实际的 ReAct 循环中，智能体能够：

• 分析用户查询的意图
• 决定是否需要调用外部工具
• 解析工具返回的结果
• 生成自然语言响应
• 处理错误和异常情况

系统架构设计

┌─────────────────┐    ┌──────────────────┐    ┌─────────────────┐
│   用户输入      │───▶│  LegacyReActAgent│───▶│   响应生成      │
│   (自然语言)    │    │   (决策引擎)     │    │   (格式化)      │
└─────────────────┘    └──────────────────┘    └─────────────────┘
                        │                      ▲
                        ▼                      │
                 ┌──────────────────┐          │
                 │   Session +      │──────────┘
                 │   Checkpointer   │ (对话历史恢复/保存)
                 │   (inmemory)     │
                 └──────────────────┘
                        │
                        ▼
                 ┌──────────────────┐
                 │   天气API工具    │
                 │   (7日预报)      │
                 └──────────────────┘
                        │
               ┌────────┴────────┐
               ▼                  ▼
        ┌────────────┐   ┌──────────────┐
        │ Open-Meteo │   │  ip-api.com  │
        │ 气象API    │   │  IP定位服务  │
        └────────────┘   └──────────────┘

对话记忆机制

LegacyReActAgent 通过 Session + Checkpointer 机制实现多轮对话的记忆：

第一次 invoke:
  create_agent_session()
    → checkpointer.pre_agent_execute()   [恢复历史，首次无数据]
    → ContextEngine.create_context()      [创建消息容器]
    → ReAct 循环：消息存入 ModelContext
    → session.commit()
    → checkpointer.post_agent_execute()   [保存状态到 InMemory dict]

第二次 invoke:
  → checkpointer.pre_agent_execute()      [从 dict 恢复历史消息]
  → 之前所有对话已就绪，作为 LLM 请求的 messages 参数传入
  → 继续追加新消息 → ReAct 循环

默认使用 in_memory 存储类型，数据保存在进程内存中，会话内支持连续追问（如"那后天呢"会自动继承之前提到的城市）。

四、模块化实现详解

1. 天气数据源

本项目使用 Open-Meteo 作为天气数据源，这是一款免费开源的气象 API，无需 API Key：

2. 工具模块

工具模块负责与外部 API 交互，提供天气数据查询功能。

天气查询工具实现：

def get_weather_info(location: str = "", date: str = ""):
    """通过 Open-Meteo API 获取天气数据（免费，无需API Key）"""
    try:
        if not date:
            date = datetime.now().strftime("%Y-%m-%d")

        if location:
            # 指定城市：通过 Geocoding API 获取坐标
            geo_url = "https://geocoding-api.open-meteo.com/v1/search"
            params = {"name": location, "count": 1, "language": "zh", "format": "json"}
            resp = requests.get(geo_url, params=params, timeout=10)
            geo_data = resp.json()
            if "results" not in geo_data or not geo_data["results"]:
                return {"error": f"未找到城市: {location}"}
            lat = geo_data["results"][0]["latitude"]
            lon = geo_data["results"][0]["longitude"]
            city_name = geo_data["results"][0].get("name", location)
        else:
            # 未指定城市：通过 IP 自动定位
            ip_resp = requests.get("http://ip-api.com/json/?fields=city,lat,lon",
                                    headers={"User-Agent": "curl/8.0"}, timeout=10)
            ip_data = ip_resp.json()
            lat = ip_data.get("lat", 39.9)
            lon = ip_data.get("lon", 116.4)
            city_name = ip_data.get("city", "当前位置")

        # 查询天气预报
        weather_url = "https://api.open-meteo.com/v1/forecast"
        weather_params = {
            "latitude": lat, "longitude": lon,
            "daily": "weathercode,temperature_2m_max,temperature_2m_min,windspeed_10m_max,winddirection_10m_dominant",
            "timezone": "auto",
            "forecast_days": 7,
        }
        w_resp = requests.get(weather_url, params=weather_params, timeout=10)
        weather_data = w_resp.json()
        daily = weather_data.get("daily", {})

        # 查找指定日期
        dates = daily.get("time", [])
        target_idx = None
        for i, d in enumerate(dates):
            if d == date:
                target_idx = i
                break

        if target_idx is not None:
            # 找到指定日期
            weather_code = daily["weathercode"][target_idx]
            weather_desc = _code_to_desc(weather_code)
            info = {
                "city": city_name,
                "date": date,
                "weather": weather_desc,
                "temperature": f"最高{daily['temperature_2m_max'][target_idx]}°C, 最低{daily['temperature_2m_min'][target_idx]}°C",
                "wind_speed": f"{daily['windspeed_10m_max'][target_idx]} km/h",
                "wind_direction": _wind_dir(daily.get("winddirection_10m_dominant", [None])[target_idx]),
            }
            return info
        else:
            # 超出预报范围，返回完整预报列表
            forecast_list = []
            for i, d in enumerate(dates):
                forecast_list.append({
                    "date": d,
                    "weather": _code_to_desc(daily["weathercode"][i]),
                    "temperature": f"最高{daily['temperature_2m_max'][i]}°C, 最低{daily['temperature_2m_min'][i]}°C",
                })
            return {
                "city": city_name,
                "date": date,
                "weather": "未找到指定日期数据",
                "forecast": forecast_list,
            }
    except Exception as e:
        return {"error": f"获取天气失败: {str(e)}"}

天气码映射：Open-Meteo 返回 WMO 天气码（整数），需要转换为中文描述：

def _code_to_desc(code: int) -> str:
    wmo_codes = {
        0: "晴朗", 1: "大部晴朗", 2: "多云", 3: "阴天",
        45: "雾", 48: "雾凇",
        51: "小毛毛雨", 53: "中毛毛雨", 55: "大毛毛雨",
        56: "冻毛毛雨", 57: "冻毛毛雨",
        61: "小雨", 63: "中雨", 65: "大雨",
        66: "冻雨", 67: "冻雨",
        71: "小雪", 73: "中雪", 75: "大雪", 77: "雪粒",
        80: "小阵雨", 81: "中阵雨", 82: "大阵雨",
        85: "小阵雪", 86: "大阵雪",
        95: "雷暴", 96: "雷暴+冰雹", 99: "雷暴+冰雹",
    }
    return wmo_codes.get(code, f"未知({code})")

def _wind_dir(deg):
    if deg is None:
        return "未知"
    dirs = ["北", "东北", "东", "东南", "南", "西南", "西", "西北"]
    idx = round(deg / 45) % 8
    return dirs[idx]

工具模块特点：

• 实现了完整的错误处理机制，网络异常时返回友好错误信息
• 支持指定城市查询和自动 IP 定位两种模式
• 当查询日期超出 7 日预报范围时，自动返回完整预报列表供 LLM 参考
• 包含超时处理机制（10s）

工具注册：将函数包装为 Agent 可调用的 Tool：

def build_weather_tool():
    from openjiuwen.core.foundation.tool.function.function import LocalFunction
    from openjiuwen.core.foundation.tool.base import ToolCard

    card = ToolCard(
        id="weather_reporter",
        name="weather_reporter",
        description="天气查询工具，查询指定地点和日期的天气信息",
        input_params={
            "type": "object",
            "properties": {
                "location": {
                    "type": "string",
                    "description": "城市名称，如北京、上海，为空则自动定位",
                },
                "date": {
                    "type": "string",
                    "description": "日期，格式YYYY-MM-DD，为空则查询今天",
                },
            },
        },
    )

    def weather_reporter(location: str = "", date: str = "") -> str:
        result = get_weather_info(location=location, date=date)
        return json.dumps(result, ensure_ascii=True)

    return LocalFunction(card=card, func=weather_reporter)

3. 模型配置模块

模型配置模块负责设置 LLM 参数和 API 连接信息。

模型配置实现：

def build_model():
    from openjiuwen.core.foundation.llm import ModelConfig, BaseModelInfo

    return ModelConfig(
        model_provider=MODEL_PROVIDER,
        model_info=BaseModelInfo(
            model=MODEL_NAME,
            api_base=API_BASE,
            api_key=API_KEY,
            temperature=0.7,
            top_p=0.9,
            timeout=30,
        ),
    )

模型配置是智能体性能的关键因素。参数说明：

• temperature (0.7)：在创造性和一致性之间取得平衡，既需要一定的创造性来生成自然的响应，又需要保持信息的准确性
• top_p (0.9)：核采样参数，确保输出在保持自然性的同时不会过于随机
• timeout (30)：考虑到天气 API 响应和 LLM 推理的时间，设置 30 秒超时

4. 提示词工程模块

提示词工程模块定义了智能体的行为规则和交互逻辑。

系统提示词设计：

def build_prompt():
    today = datetime.now().strftime("%Y-%m-%d")
    return [
        {
            "role": "system",
            "content": (
                f"你是智能天气助手，负责回答天气查询及相关生活建议。\n"
                f"1. 默认查今天（{today}）的天气；\n"
                "2. 调用工具时直接使用中文城市名，如北京、上海；\n"
                "3. 城市参数为中文城市名，不明确指定城市时留空，API自动定位；\n"
                "4. 用户未指定日期时，默认使用今天；\n"
                "5. 穿衣/出行建议应先查天气再给出建议；\n"
                "6. 与天气无关的问题，不回答，介绍自己的能力。"
            ),
        }
    ]

提示词工程的核心原则：

• 定义了明确的工具调用规则
• 规定了自动定位逻辑
• 设置了智能决策条件
• 确定了超出范围问题的处理方式

5. 智能体核心模块

智能体核心模块整合了所有组件，创建完整的智能体实例。

智能体创建：

def create_weather_agent():
    from openjiuwen.core.single_agent.legacy import create_react_agent_config, LegacyReActAgent

    agent_config = create_react_agent_config(
        agent_id="weather_assistant",
        agent_version="0.1.0",
        description="智能天气助手 - 天气查询、穿衣/出行建议",
        model=build_model(),
        prompt_template=build_prompt(),
    )

    agent = LegacyReActAgent(agent_config)
    tool_func = build_weather_tool()
    agent.add_tools([tool_func])
    return agent

完整主程序：

import asyncio

async def main():
    print("=" * 60)
    print("  智能天气助手 (openJiuwen ReAct Agent)")
    print("=" * 60)
    print("  支持的查询：上海明天的天气、明天穿什么衣服、周末适合出行吗")
    print("  输入 'exit' 或 'quit' 退出\n")

    agent = create_weather_agent()

    while True:
        try:
            query = input("\n>>> ")
            if query.lower() in ("exit", "quit", "q"):
                break
            if not query.strip():
                continue

            result = await agent.invoke({"query": query})
            output = result.get("output") or result.get("result", "")
            print(f"\n{output}")

        except KeyboardInterrupt:
            break
        except Exception as e:
            print(f"\n错误: {e}")

if __name__ == "__main__":
    asyncio.run(main())

五、效果展示

智能体运行起来，有一个提示信息展示和简单的交互界面，等待输入问题。

天气查询

指定城市查询

>>> 南京天气怎么样

📍 城市： 南京
⛈️ 天气： 雷暴+冰雹
🌡️ 温度： 26.7°C ~ 32.0°C
💨 风向风速： 南风，16.2 km/h

自动定位查询

>>> 今天天气怎么样

📍 城市： 广州
🌩️ 天气： 雷暴
🌡️ 温度： 25.7°C ~ 31.3°C
💨 风向风速： 东南风，12.6 km/h

多轮连续对话

>>> 南京天气怎么样          ← 查询南京
>>> 下周的天气怎么样         ← 自动继承南京，批量查7天
>>> 那后天呢                ← 自动推断7月11日，继续查南京

穿衣建议

>>> 明天穿什么衣服

👕 穿衣建议
- 短袖T恤、薄衬衫，透气为主
- 建议穿防水的凉鞋或运动鞋，路面湿滑
- 必备品：带伞！

出行建议

>>> 周末适合户外活动吗

🚶 出行建议
- 虽然是毛毛雨，但风力不小，雨伞可能会被吹歪
- 气温较高，体感闷热潮湿，记得多补充水分
- 适合在家放松或去室内场所逛逛

六、项目总结

智能天气助手项目展示了 AI 技术在传统服务领域的应用价值。通过 openJiuwen Agent Core 框架，成功构建了一个具备智能决策、自然语言理解、个性化服务等能力的智能助手。

项目的核心价值在于：

• 智能化升级：从数据查询到智能服务的跃迁
• 用户体验优化：自然语言交互，个性化服务
• 技术架构先进：ReAct 架构，模块化设计
• 扩展性强：支持多种应用场景和功能扩展

通过本项目，验证了 ReAct 架构在构建智能助手方面的有效性，展示了如何将免费的天气 API 数据转化为对用户有价值的智能服务。随着 AI 技术的不断发展，智能天气助手将为用户提供更加智能、便捷、个性化的服务体验。