数据中心气象代理
一种LangGraph代理系统，使用模型上下文协议（MCP）确定数据中心位置的天气预报。
建筑：自定义状态图，具有显式状态管理、模块化节点和符合LangGraph 2026最佳实践的条件路由。
概述
该系统回答了以下问题： *“数据中心的天气预报是什么？”*
它自主地： 0.检查意图分类，避免回答不应该回答的问题，或对任何问题给出相同的答案
- 发现数据中心的公共IP地址
- 将IP解析为地理坐标
- 获取该位置的当前天气预报
- 合成自然语言响应
为什么是这种架构？
此实现使用LangGraph的 自定义状态图 方法而不是预构建的代理，因为：
- 完全透明：每个决策点都是明确的，可调试的
- 生产就绪：模块化设计支持测试、监控和扩展
- 最佳实践：遵循LangGraph 2026对LangGraph文档中获取的有状态代理的建议：https://docs.langchain.com/oss/python/langgraph/workflows-agents
- 可维护性：与类型化状态模式明确分离关注点
系统架构
┌─────────────────────────────────────────────────────────────────┐
│ Data Center Weather Agent │
│ │
│ ┌─────────────┐ ┌───────────────────────────────────┐ │
│ │ Agent │ HTTP │ MCP Server │ │
│ │ (LangGraph) │◄───────►│ ┌────────┐ ┌─────────┐ ┌───────┐ │ │
│ │ │ SSE │ │ ipify │ │ip_to_geo│ │weather│ │ |
│ │ 5 Nodes │ │ └───┬────┘ └────┬────┘ └──┬────┘ │ │
│ │ 3 Edges │ │ │ │ │ │ │
│ │ 9 State │ │ ▼ ▼ ▼ │ │
│ │ Fields │ │ ipify.org ip-api.com meteo │ │
│ └─────────────┘ └───────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘组件
1.MCP服务器(server/)
- 技术：Python、FastMCP、uvicorn
- 端口：8000（苏格兰和南方能源公司运输）
- 工具:
- ipify：获取公共IP地址 - ip_to_geo：将IP转换为lat/lon（IP-api.com） - weather_forecast：获取天气数据（Open Meteo）
- 特性：输入验证、错误处理、结构化日志记录
2.LangGraph代理(agent/)
- 技术：Python、LangGraph、LangChain、谷歌Gemini
- 建筑：自定义状态图
- 状态：跟踪工作流进度的9个键入字段
- 节点：5个模块化计算单元
- 边缘：3个条件路由函数
- 特性：LLM回退、详细跟踪、错误恢复
状态管理
所有数据都通过强类型 AgentState:
class AgentState(TypedDict):
question: str # User's input query
public_ip: str | None # From ipify tool
latitude: float | None # From ip_to_geo tool
longitude: float | None # From ip_to_geo tool
weather_data: str | None # From weather_forecast tool
answer: str | None # Final LLM response
messages: list # Conversation history
error: str | None # Error tracking
current_step: str # Progress indicator图形结构
START
↓
┌───────────────────────┐
│ intent classification│ → [custom logic]
└──────┬────────────────┘
↓
┌─────────────┐
│ get_ip │ → [ipify tool]
└──────┬──────┘
│ [conditional: success/error]
↓
┌──────────────────┐
│ resolve_location │ → [ip_to_geo tool]
└──────┬───────────┘
│ [conditional: success/error]
↓
┌──────────────┐
│fetch_weather │ → [weather_forecast tool]
└──────┬───────┘
│ [conditional: success/error]
↓
┌────────────────┐
│generate_answer │ → [LLM synthesis]
└───────┬────────┘
↓
END
[error] ← Any failure routes here节点职责
- get_ip_node：调用ipify工具，使用public_ip更新状态
- resolve_location_node：调用ip_to_geo，提取lat/lon
- fetch_weather_node：使用坐标调用weather_forecast
- generate_answer_node：LLM创建自然语言响应
- 错误节点：集中错误处理和用户消息传递
条件路由
在继续之前，每条边都会验证之前的操作：
def route_after_ip(state: AgentState) -> Literal["resolve_location", "error"]:
"""Route to next node or error based on IP fetch result"""
if state.get("error") or not state.get("public_ip"):
return "error"
return "resolve_location"设置
先决条件
- Python 3.10或更高版本
- 在3.14.2版本上进行了全面测试
- Google Gemini API密钥（提供免费层）
- 可选：LongCat API密钥（回退）
安装
- 克隆或下载 这个项目
- 安装依赖项:
pip install -r requirements.txt- 配置API密钥:
复制示例环境文件：
cp .env.example .env编辑 .env 并添加您的API密钥：
GOOGLE_API_KEY=your_gemini_api_key_here
LONGCAT_API_KEY=your_longcat_api_key_here # Optional获取API密钥：
- 双子座：https://ai.google.dev/ - 长猫：https://longcat.chat/（可选回退）
用法
启动系统
您需要两个终端窗口：
终端1-启动MCP服务器:
python3 -m server.main预期产量：
INFO: Started server process
INFO: Waiting for application startup.
INFO: Application startup complete.
INFO: Uvicorn running on http://0.0.0.0:8000终端2-运行代理:
python3 -m agent.main示例会话
============================================================
Data Center Weather Agent (Custom StateGraph)
============================================================
Connected to MCP Server
LLM configured with fallback: Gemini -> LongCat
Graph structure built successfully
Agent Ready! (Type 'quit' to exit)
------------------------------------------------------------
Enter your question: What is the weather forecast of the data center?
============================================================
EXECUTION TRACE
============================================================
[Step 1: IP Discovery]
Tool: ipify
Result: 174.162.142.78
[Step 2: Location Resolution]
Tool: ip_to_geo
Input: 174.162.142.78
Result: 40.3495, -111.8998
[Step 3: Weather Retrieval]
Tool: weather_forecast
Input: lat=40.3495, lon=-111.8998
Result: Temperature: 0.2 C, Windspeed: 2.2 km/h
[Step 4: Answer Generation]
Generated answer successfully
============================================================
FINAL ANSWER
============================================================
The data center, located at 40.3495°N, 111.8998°W, currently has a
temperature of 0.2°C and a wind speed of 2.2 km/h.
Enter your question: Where is the data center located?
[... processes IP and location lookup ...]
FINAL ANSWER
The data center is located at coordinates 40.3495°N, 111.8998°W
(approximately in Utah, United States).
Enter your question: quit文件结构
datacenter_weather_agent/
├── server/
│ ├── main.py # MCP server with FastMCP
│ ├── tools.py # Tool implementations
│ └── __init__.py
├── agent/
│ ├── main.py # Custom StateGraph agent (~570 lines)
│ ├── client.py # MCP client wrapper
│ └── __init__.py
├── README.md # This file
├── ARCHITECTURE.md # Deep technical documentation
├── .env.example # API key template
├── .env # Your API keys (create this)
└── requirements.txt # Python dependencies主要特点
1.稳健的输入验证
服务器端:
- IP地址格式验证（IPv4/IPv6）
- IPv4八位字节范围检查（0-255）
- 坐标范围验证（纬度：-90至90，经度：-180至180）
- 对所有参数进行类型检查
客户端:
- 响应结构验证
- 非空内容验证
- 运行时错误检测
2.全面的错误处理
每个节点错误处理:
async def get_ip_node(state, tools):
try:
# ... tool execution ...
return success_state
except Exception as e:
logger.error(f"Error in get_ip_node: {e}")
return error_state条件路由到错误节点:
- 任何节点故障都会路由到集中式错误处理程序
- 用户友好的错误消息
- 代理保持响应（不崩溃）
3.LLM回退策略###->额外功能\ str:
# Implementation return timezone_data
1. **注册于 `server/main.py`**:
@mcp.tool() async def timezone(latitude: float, longitude: float) -> str: return await get_timezone(latitude, longitude)
1. **添加节点 `agent/main.py`**:
async def get_timezone_node(state, tools): tool = next(t for t in tools if t.name == "timezone") result = await tool.ainvoke({ "latitude": state["latitude"], "longitude": state["longitude"] }) return {**state, "timezone": result}
workflow.add_node("get_timezone", get_timezone_node) workflow.add_edge("fetch_weather", "get_timezone") workflow.add_edge("get_timezone", "generate_answer")
### 修改路由逻辑
更改条件边以支持替代流：
def route_after_ip(state: AgentState): if state.get("error"): return "error"
# New: Check if IP is internal/private ip = state.get("public_ip", "") if ip.startswith("192.168.") or ip.startswith("10."): return "handle_private_ip" # New node
return "resolve_location"
## 测试
### 单元测试
测试单个节点：
import pytest from agent.main import get_ip_node
@pytest.mark.asyncio async def test_get_ip_node(): mock_tools = [MockIPifyTool()] state = {"question": "test", "messages": []}
result = await get_ip_node(state, mock_tools)
assert result["public_ip"] is not None assert result["error"] is None
### 集成测试
测试完整图形执行：
@pytest.mark.asyncio async def test_full_workflow(mcp_server_running): graph = await build_graph(client, llm) initial_state = {...}
final_state = await graph.ainvoke(initial_state)
assert final_state["answer"] is not None assert "temperature" in final_state["answer"].lower()
### LangSmith监控（来源于文档，最后访问时间为2026年1月30日）
启用生产调试跟踪：
export LANGCHAIN_TRACING_V2=true export LANGCHAIN_API_KEY=your_key export LANGCHAIN_PROJECT="datacenter-weather"
每次执行都会在LangSmith仪表板上进行跟踪。
## 安全考虑
1. **API密钥**：从不承诺 `.env` 到版本控制
1. **服务器暴露**：MCP服务器没有身份验证（仅限本地使用）
1. **速率限制**：外部API可能会禁止滥用
1. **输入验证**：服务器验证所有工具输入
1. **错误消息**：避免日志中的敏感数据泄露
## 依赖项
- **httpx**：异步HTTP客户端
- **主控程序**：模型上下文协议SDK
- **兰格拉夫**：基于图的代理框架
- **语言链**：LLM抽象层
- **郎链谷歌genai**：Gemini集成
- **兰开夏**：OpenAI兼容API（LongCat）
- **python dotenv**：环境变量管理
- **优维康**：ASGI服务器
看 `requirements.txt` 对于特定版本。
## 了解更多
- [LangGraph文档](https://langchain.com/docs/langgraph)
- [MCP规范](https://modelcontextprotocol.io/)
- [状态图教程](https://langchain.com/docs/langgraph/tutorials/state_graph)
- [Google Gemini API](https://ai.google.dev/gemini-api/docs)
## 许可证
这是一个用于个人目的的示范项目。不提供保修。
______________________________________________________________________
**建于**：LangGraph自定义状态图（生产最佳实践，2026年1月）