地理空间 AI 智能体（Code Agent）中国区部署实践

# webmain.py
import json, os, sys, hmac, hashlib, time, tempfile
from pathlib import Path
from fastapi import FastAPI, HTTPException, Request, Depends
from fastapi.responses import StreamingResponse, FileResponse
from fastapi.staticfiles import StaticFiles
from starlette.middleware.gzip import GZipMiddleware
BASE_DIR = Path(__file__).parent
DOWNLOAD_DIR = Path(tempfile.gettempdir()) / "geo-agent-files"
DOWNLOAD_DIR.mkdir(parents=True, exist_ok=True)
def load_config():
    """config.json 只存非敏感配置，敏感凭证从环境变量读取"""
    config_path = Path(os.getenv("CONFIG_PATH", BASE_DIR / "config.json"))
    file_cfg = json.load(open(config_path)) if config_path.is_file() else {}
    return {
        "aws": {
            "access_key_id": os.getenv("AWS_ACCESS_KEY_ID", ""),
            "secret_access_key": os.getenv("AWS_SECRET_ACCESS_KEY", ""),
            "region_name": file_cfg.get("aws",{}).get("region_name","us-east-1"),
        },
        "bedrock": {"model_id": file_cfg.get("bedrock",{}).get("model_id","minimax.minimax-m2.5")},
        "tianditu": {"api_key": os.getenv("TIANDITU_API_KEY", "")},
        "auth": {"users": json.loads(os.getenv("AUTH_USERS", "{}"))},
    }
CONFIG = load_config()
os.environ["AWS_ACCESS_KEY_ID"] = CONFIG["aws"]["access_key_id"]
os.environ["AWS_SECRET_ACCESS_KEY"] = CONFIG["aws"]["secret_access_key"]
os.environ["AWS_DEFAULT_REGION"] = CONFIG["aws"]["region_name"]
os.environ["TIANDITU_API_KEY"] = CONFIG["tianditu"]["api_key"]
_TOKEN_SECRET = (CONFIG["aws"]["secret_access_key"] or "default-secret").encode()
_TOKEN_EXPIRY = 86400 * 7  # 7 天
def _sign_token(username):
    expire = int(time.time()) + _TOKEN_EXPIRY
    msg = f"{username}.{expire}".encode()
    sig = hmac.new(_TOKEN_SECRET, msg, hashlib.sha256).hexdigest()[:32]
    return f"{username}.{expire}.{sig}"
app = FastAPI()
app.add_middleware(GZipMiddleware, minimum_size=1000)
@app.post("/login")
async def login(request: Request):
    """简易登录（替代 Cognito）"""
    data = await request.json()
    username, password = data.get("username",""), data.get("password","")
    auth_users = CONFIG["auth"]["users"]
    if not auth_users:
        return {"token": _sign_token("anonymous")}
    if auth_users.get(username) == password:
        return {"token": _sign_token(username)}
    raise HTTPException(status_code=401, detail="用户名或密码错误")
@app.post("/invocations")
async def invoke_agent(request: Request):
    """SSE 流式响应，与 AgentCore 接口兼容"""
    payload = await request.json()
    agent = GeospatialAgent(
        payload.get("coordinates"),
        payload.get("sessionId", "local"),
        payload.get("history"),
        payload.get("model_id", CONFIG["bedrock"]["model_id"]),
    )
    async def stream():
        async for msg in agent.stream_async(payload["message"]):
            yield f"data: {json.dumps(msg, ensure_ascii=False)}\n\n"
    return StreamingResponse(stream(), media_type="text/event-stream")
@app.get("/download/{filename}")
async def download_file(filename: str):
    """本地文件下载（替代 S3 pre-signed URL）"""
    path = DOWNLOAD_DIR / filename
    if not path.exists():
        raise HTTPException(status_code=404)
    return FileResponse(path)
# 前端静态文件托管
FRONTEND = BASE_DIR / "user-interface" / "dist"
if FRONTEND.exists():
    app.mount("/", StaticFiles(directory=str(FRONTEND), html=True))

// 矢量底图（CGCS2000，无偏移）
http​://t{0-7}.tianditu.gov.cn/vec_w/wmts?...&tk=
// 影像底图
http​://t{0-7}.tianditu.gov.cn/img_w/wmts?...&tk=
// 中文标注
http​://t{0-7}.tianditu.gov.cn/cva_w/wmts?...&tk=

TIANDITU_KEY = os.getenv("TIANDITU_API_KEY", "")
def geocode(location_name: str):
    url = "http://api.tianditu.gov.cn/geocoder"
    params = {"ds": json.dumps({"keyWord": location_name}), "tk": TIANDITU_KEY}
    resp = requests.get(url, params=params).json()
    if resp.get("status") == "0":
        return (resp["location"]["lat"], resp["location"]["lon"])
    return None

{
    "aws": {"region_name": "us-east-1"},
    "bedrock": {"model_id": "minimax.minimax-m2.5"}
}