第04章 - 一张图「智能体」实战

# route_v1.py —— 路线①：关键词路由，先跑起来，别急着上模型
from re import compile as re_compile

ROUTES = [
    (re_compile(r"(放大|定位|飞到|缩放到)"), "map.zoom_to"),
    (re_compile(r"(打开|显示|叠加).*(图层|专题)"), "map.toggle_layer"),
    (re_compile(r"(多少|几个|统计|分布).*(个|所|家|处)?"), "analysis.stat"),
    (re_compile(r"(量算|测距|面积)"), "analysis.measure"),
]

def route(text: str) -> str:
    for pattern, tool in ROUTES:
        if pattern.search(text):
            return tool
    return "fallback.chat"

if __name__ == "__main__":
    samples = ["放大到余杭区", "把教育设施图层打开", "这个街道有多少所学校"]
    for q in samples:
        print(q, "->", route(q))

# blackboard.py —— 任务黑板的最小结构，直接用 dataclass
from dataclasses import dataclass, field
from typing import Any

@dataclass
class Blackboard:
    goal: str = ""
    entities: dict[str, Any] = field(default_factory=dict)      # {"adcode": "330110"}
    artifacts: dict[str, Any] = field(default_factory=dict)     # {"stat_1": {"count": 186}}
    last_action: dict[str, Any] = field(default_factory=dict)   # {"tool": "map.zoom_to", "layer": "school"}
    steps: list[dict] = field(default_factory=list)             # 审计与回放

    def snapshot(self, max_chars: int = 1200) -> str:
        """给模型看的紧凑状态，宁可字段少也不要超长。"""
        text = f"goal={self.goal}; entities={self.entities}; last={self.last_action}"
        return text[:max_chars]

    def record(self, tool: str, args: dict, ok: bool, summary: str) -> None:
        self.steps.append({"tool": tool, "args": args, "ok": ok, "summary": summary})
        self.last_action = {"tool": tool, **args}

# recovery.py —— 三类失败的分派，别用一个 except 打天下
from enum import Enum

class FailKind(Enum):
    PARAM = "param"
    TRANSIENT = "transient"
    BUSINESS = "business"

def classify(code: str) -> FailKind:
    if code in ("INVALID_PARAM", "NOT_FOUND_LAYER"):
        return FailKind.PARAM
    if code in ("TIMEOUT", "UPSTREAM_ERROR", "RATE_LIMITED"):
        return FailKind.TRANSIENT
    return FailKind.BUSINESS            # FORBIDDEN / EMPTY_RESULT / UNSUPPORTED

def handle(code: str, attempt: int, readonly: bool) -> dict:
    kind = classify(code)
    if kind is FailKind.PARAM and attempt < 1:
        return {"action": "retry_with_feedback", "attempt": attempt + 1}
    if kind is FailKind.TRANSIENT and readonly and attempt < 2:
        return {"action": "retry_backoff", "delay_s": 2 ** attempt}
    return {"action": "degrade", "ask_user": True}

# minimal_agent.py —— Qwen-Agent 最小闭环：一个工具 + 一个 Assistant
from qwen_agent.agents import Assistant

def stat_features(adcode: str, facility_type: str = "school") -> dict:
    """演示用：真实实现应查库并带白名单。"""
    demo = {"330110": {"school": 186, "hospital": 63}}
    return {"ok": True, "count": demo.get(adcode, {}).get(facility_type, 0)}

# 工具既可以写成本地函数，也可以用 tools 里的 Function 子类声明 schema
TOOLS = [{
    "name": "stat_features",
    "description": "按行政区代码与设施类型统计要素数量。adcode 为 6 位行政区代码，facility_type 取 school 或 hospital。",
    "parameters": {
        "type": "object",
        "properties": {
            "adcode": {"type": "string", "description": "6 位行政区代码，如 330110"},
            "facility_type": {"type": "string", "enum": ["school", "hospital"]}
        },
        "required": ["adcode", "facility_type"]
    }
}]

bot = Assistant(
    llm={
        "model": "qwen3-32b",
        "model_server": "http://127.0.0.1:8000/v1",   # vLLM / DashScope 兼容端点
        "api_key": "EMPTY",
        "generate_cfg": {"temperature": 0.1, "max_tokens": 1024},
    },
    function_list=TOOLS,
    system_message="你是地图助手。只能调用给定工具，不要编造数据；缺参数时先追问。",
)

messages = [{"role": "user", "content": "余杭区有多少所学校？"}]
for chunk in bot.run(messages=messages):      # 流式输出，最后一项是完整消息列表
    final = chunk

print(final[-1]["content"])

# route_then_plan.py —— 先路由后编排，两个档位分开配置与限流
from openai import OpenAI

client = OpenAI(api_key="从环境变量读取", base_url="https://api.deepseek.com")
ROUTER_MODEL = "deepseek-v4-flash"    # 高频、便宜、只做判断
PLANNER_MODEL = "deepseek-v4-pro"     # 低频、贵、对准确率负责
INTENTS = {"map", "query", "analysis", "render", "write"}

def route(text: str) -> str:
    """路由档：短提示、固定枚举、temperature=0，只让它输出一个词。"""
    resp = client.chat.completions.create(
        model=ROUTER_MODEL,
        temperature=0,
        max_tokens=8,
        messages=[
            {"role": "system",
             "content": "把请求归到 map/query/analysis/render/write 之一，只输出类别名，不要解释。"},
            {"role": "user", "content": text},
        ],
    )
    label = (resp.choices[0].message.content or "").strip()
    return label if label in INTENTS else "write"    # 判不出来时按最谨慎的档处理

def plan(text: str, tool_schemas: list[dict]) -> object:
    """编排档：带完整工具定义，温度 0.1，允许长上下文。"""
    return client.chat.completions.create(
        model=PLANNER_MODEL,
        temperature=0.1,
        tools=tool_schemas,
        tool_choice="auto",
        messages=[
            {"role": "system", "content": "你是地图编排器，只能调用给定工具。"},
            {"role": "user", "content": text},
        ],
    )

if __name__ == "__main__":
    text = "把这三个街道的学校按密度分色，再叠加医院"
    intent = route(text)                       # 先花一次便宜调用
    print("intent:", intent)
    if intent != "write":                      # 写操作不进自动编排，走 4.5.4 的确认流程
        resp = plan(text, TOOL_SCHEMAS)        # TOOL_SCHEMAS 按 4.4.1 的规范生成
        print(resp.choices[0].message.tool_calls)

# onemap_mcp/server.py —— 最小 GIS MCP server（stdio）
from mcp.server.fastmcp import FastMCP
from pydantic import BaseModel, Field

mcp = FastMCP("onemap-gis")

# 演示数据；真实实现接图层目录库与空间数据库
LAYERS = {
    "school_2023": {"name": "教育设施（2023）", "geom": "point", "count": 186},
    "hospital_2023": {"name": "医疗机构（2023）", "geom": "point", "count": 63},
}

class LayerQuery(BaseModel):
    keyword: str = Field(description="图层名称关键字，如 教育、医院")
    limit: int = Field(default=5, ge=1, le=20, description="最多返回条数")

class BufferReq(BaseModel):
    layer_id: str = Field(description="图层 id，须先通过 search_layers 获得")
    distance_m: int = Field(ge=1, le=5000, description="缓冲距离，单位米，上限 5000")

@mcp.tool()
def search_layers(q: LayerQuery) -> dict:
    """按关键字检索图层目录。找不到时返回 EMPTY_RESULT，不要据此臆造图层 id。"""
    hits = [{"layer_id": k, **v} for k, v in LAYERS.items() if q.keyword in v["name"]]
    if not hits:
        return {"ok": False, "code": "EMPTY_RESULT", "hint": f"没有名称含「{q.keyword}」的图层"}
    return {"ok": True, "data": hits[: q.limit], "total": len(hits)}

@mcp.tool()
def buffer_layer(req: BufferReq) -> dict:
    """对指定图层做缓冲区分析，返回临时图层 id 与要素数。只读工具，不修改源数据。"""
    if req.layer_id not in LAYERS:
        return {"ok": False, "code": "NOT_FOUND", "hint": "先调用 search_layers 获取合法 layer_id"}
    return {"ok": True, "data": {"result_layer": f"{req.layer_id}__buf{req.distance_m}",
                                 "distance_m": req.distance_m, "count": LAYERS[req.layer_id]["count"]}}

if __name__ == "__main__":
    mcp.run(transport="stdio")     # 远程部署时改用 streamable HTTP

# onemap_mcp/http_server.py —— 单端点 Streamable HTTP，按协议版本头分流
from mcp.server.fastmcp import FastMCP
from starlette.applications import Starlette
from starlette.responses import JSONResponse

mcp = FastMCP("onemap-gis")
SUPPORTED = {"2026-07-28"}          # 以官方规范文档的版本号为准，不支持就往上报错

# 工具注册与 4.4.4 一致，此处省略

async def version_guard(request, call_next):
    """协议版本头缺失或不受支持时明确拒绝，不要让客户端猜。"""
    version = request.headers.get("mcp-protocol-version", "")
    if version and version not in SUPPORTED:
        return JSONResponse(
            {"error": "UNSUPPORTED_PROTOCOL_VERSION",
             "supported": sorted(SUPPORTED),
             "hint": "升级客户端或 server，不要静默降级"},
            status_code=400,
        )
    return await call_next(request)

app = mcp.streamable_http_app()      # 单端点；具体方法名以当前 Python SDK 版本为准
app.add_middleware(version_guard)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8765)   # 生产必须放在反代后面加鉴权

# result.py —— 统一返回体与错误码，所有工具都返回这个结构
from typing import Any

CODES = {
    "INVALID_PARAM": "参数不合法，按 hint 修正后重试",
    "NOT_FOUND": "对象不存在，先检索再调用",
    "FORBIDDEN": "越权访问，不要重试，换用可用范围",
    "TIMEOUT": "查询超时，缩小范围或降低 limit",
    "UPSTREAM_ERROR": "上游服务异常，可重试一次",
    "EMPTY_RESULT": "查询成功但无数据，属正常结果",
}

def fail(code: str, hint: str = "", retryable: bool = False) -> dict:
    return {"ok": False, "code": code, "message": CODES.get(code, code),
            "hint": hint, "retryable": retryable}

def ok(data: Any, message: str = "") -> dict:
    return {"ok": True, "data": data, "message": message}

# mongodb_tool.py —— 字段白名单 + 结构化过滤 + 分页 + 超时
import re
from pymongo import MongoClient
from pymongo.errors import PyMongoError, ExecutionTimeout
from result import ok, fail

ALLOWED_COLLECTIONS = {"edu_facility", "medical_facility", "admin_boundary"}
FIELD_WHITELIST = {
    "edu_facility":    {"name", "adcode", "type", "students", "geom"},
    "medical_facility": {"name", "adcode", "level", "beds", "geom"},
    "admin_boundary":  {"name", "adcode", "area_km2", "geom"},
}
OPS = {"eq": "$eq", "ne": "$ne", "gt": "$gt", "gte": "$gte", "lt": "$lt", "lte": "$lte", "in": "$in"}
MAX_LIMIT = 200
PROJECTION_ALWAYS = {"_id": 0}

class MongoDBTool:
    def __init__(self, uri: str, db: str = "onemap") -> None:
        self.client = MongoClient(uri, serverSelectionTimeoutMS=3000, readPreference="secondaryPreferred")
        self.db = self.client[db]

    def _check(self, collection: str, field: str | None = None) -> str | None:
        if collection not in ALLOWED_COLLECTIONS:
            return f"集合 {collection} 不在白名单，可选：{sorted(ALLOWED_COLLECTIONS)}"
        if field is not None and field not in FIELD_WHITELIST[collection]:
            return f"字段 {field} 不可查，可选：{sorted(FIELD_WHITELIST[collection])}"
        return None

    def _build_filter(self, collection: str, conds: list[dict]) -> dict:
        query: dict = {}
        for c in conds:
            field, op, value = c.get("field"), c.get("op", "eq"), c.get("value")
            if (err := self._check(collection, field)) is not None:
                raise ValueError(err)
            query.setdefault(field, {}).setdefault(OPS[op], value)
        return query

    def query_features(self, collection: str, conds: list[dict] | None = None,
                       fields: list[str] | None = None, limit: int = 20) -> dict:
        if (err := self._check(collection)) is not None:
            return fail("INVALID_PARAM", err)
        limit = max(1, min(int(limit), MAX_LIMIT))
        projection = dict(PROJECTION_ALWAYS)
        for f in (fields or []):
            if self._check(collection, f) is not None:
                return fail("INVALID_PARAM", f"字段 {f} 不可返回")
            projection[f] = 1
        try:
            query = self._build_filter(collection, conds or [])
            cursor = (self.db[collection].find(query, projection)
                      .limit(limit).max_time_ms(2000))
            rows = list(cursor)
        except ExecutionTimeout:
            return fail("TIMEOUT", "缩小 bbox 或减少条件后重试", retryable=True)
        except (ValueError, PyMongoError) as exc:
            return fail("UPSTREAM_ERROR", f"查询失败：{exc}", retryable=True)
        if not rows:
            return ok({"items": [], "total": 0}, "无匹配要素，属正常空结果")
        geom_free = [{k: v for k, v in r.items() if k != "geom"} for r in rows]
        return ok({"items": geom_free[:5], "total": len(rows),
                   "result_ref": f"mongo:{collection}:{abs(hash(str(query))) % 10**8}",
                   "truncated": len(rows) == limit})

# map_tool.py —— 地图状态指令，返回"前端该做什么"而不是"图已经变了"
import uuid
from result import ok, fail

class MapTool:
    def __init__(self, layer_catalog: dict[str, dict]) -> None:
        self.catalog = layer_catalog
        self.temp_layers: dict[str, dict] = {}

    def toggle_layer(self, layer_id: str, visible: bool) -> dict:
        if layer_id not in self.catalog:
            return fail("NOT_FOUND", f"图层 {layer_id} 不存在，先调 search_layers")
        return ok({"layer_id": layer_id, "visible": bool(visible)})

    def zoom_to(self, bbox: list[float] | None = None, layer_id: str | None = None,
                padding: int = 40) -> dict:
        if bbox is None and layer_id is None:
            return fail("INVALID_PARAM", "bbox 与 layer_id 至少给一个")
        if bbox is not None:
            if len(bbox) != 4 or bbox[0] >= bbox[2] or bbox[1] >= bbox[3]:
                return fail("INVALID_PARAM", "bbox 须为 [minx,miny,maxx,maxy] 且 min < max")
        return ok({"bbox": bbox, "layer_id": layer_id, "padding": padding})

    def add_temp_layer(self, geojson_ref: str, name: str, style: dict | None = None) -> dict:
        lid = f"tmp_{uuid.uuid4().hex[:8]}"
        self.temp_layers[lid] = {"name": name[:40], "style": style or {}, "ref": geojson_ref}
        return ok({"layer_id": lid, "ttl_minutes": 30})

# confirm.py —— 两阶段写操作，token 有 TTL 且一次性
import secrets, time

_PENDING: dict[str, dict] = {}

def prepare(action: str, payload: dict, ttl_s: int = 120) -> dict:
    token = secrets.token_urlsafe(16)
    _PENDING[token] = {"action": action, "payload": payload, "exp": time.time() + ttl_s, "used": False}
    return {"ok": True, "data": {"confirm_token": token,
                                 "summary": f"将执行 {action}，参数 {payload}",
                                 "expires_in": ttl_s}}

def commit(token: str) -> dict:
    rec = _PENDING.get(token)
    if not rec or rec["used"] or rec["exp"] < time.time():
        return {"ok": False, "code": "INVALID_PARAM", "hint": "确认令牌已失效，请重新发起"}
    rec["used"] = True
    return {"ok": True, "data": {"action": rec["action"], "payload": rec["payload"]}}

# result_budget.py —— 4 KB 硬预算：先裁样例，再换句柄，最后分页
import json, uuid

MAX_BYTES = 4096          # 与 4.4.2 的阈值保持一致，CI 里断言同一个常量
_STORE: dict[str, dict] = {}   # 生产环境换成 Redis，带 TTL

def size_of(obj) -> int:
    return len(json.dumps(obj, ensure_ascii=False).encode("utf-8"))

def make_handle(payload: dict, ttl_s: int = 600) -> str:
    ref = f"ref_{uuid.uuid4().hex[:12]}"      # 随机且不可猜，不要用可枚举自增 id
    _STORE[ref] = {"payload": payload, "ttl_s": ttl_s}
    return ref

def budgeted(rows: list[dict], sample: int = 5) -> dict:
    """返回体超 4 KB 就换句柄；未超就返回摘要 + 总数。"""
    summary = {"items": rows[:sample], "total": len(rows), "truncated": len(rows) > sample}
    if size_of(summary) <= MAX_BYTES:
        return {"ok": True, "data": summary}
    ref = make_handle({"rows": rows})
    slim = {"sample": rows[:2], "total": len(rows), "result_ref": ref,
            "hint": "明细超过 4 KB，需要完整数据请带 result_ref 调 fetch_result，按 next_cursor 翻页"}
    return {"ok": True, "data": slim}

def fetch_result(result_ref: str, cursor: int = 0, page_size: int = 50) -> dict:
    """句柄取数：一次只返回一页，模型自己决定还要不要下一页。"""
    rec = _STORE.get(result_ref)
    if rec is None:
        return {"ok": False, "code": "NOT_FOUND", "hint": "句柄已过期，请重新查询"}
    rows = rec["payload"]["rows"]
    page = rows[cursor: cursor + page_size]
    next_cursor = cursor + page_size if cursor + page_size < len(rows) else None
    return {"ok": True, "data": {"items": page, "next_cursor": next_cursor,
                                    "total": len(rows), "page_size": page_size}}

# resolve_place.py —— 歧义就返回候选，让模型/用户选，不要猜
CANDIDATES = {
    "城关镇": [{"adcode": "330110001", "full": "杭州市余杭区城关镇"},
               {"adcode": "330127001", "full": "杭州市淳安县城关镇"}],
}

def resolve_place(text: str, use_selection: bool = False) -> dict:
    hits = CANDIDATES.get(text, [])
    if len(hits) == 1:
        return {"ok": True, "data": {"adcodes": [hits[0]["adcode"]], "name": hits[0]["full"]}}
    if len(hits) > 1:
        return {"ok": False, "code": "INVALID_PARAM", "retryable": True,
                "hint": "地名有歧义，请让用户从候选中选择，不要自行假设",
                "data": {"options": hits}}
    return {"ok": False, "code": "NOT_FOUND", "hint": "检索无结果，试更完整的行政区名"}

# theme_tool.py —— 渲染规则的服务端校验，结构非法直接打回给模型
from result import ok, fail

METHODS = {"graduated": 3, "categorical": 2, "heatmap": 1, "cluster": 0, "single": 0}
HEX = set("0123456789abcdefABCDEF")

def set_render_rule(layer_id: str, method: str, field: str | None = None,
                    breaks: list[float] | None = None, colors: list[str] | None = None) -> dict:
    if method not in METHODS:
        return fail("INVALID_PARAM", f"method 只能是 {sorted(METHODS)}")
    if method == "graduated":
        if not breaks or len(breaks) < 1 or breaks != sorted(breaks):
            return fail("INVALID_PARAM", "breaks 必须是升序且至少 1 个")
        if not colors or len(colors) != len(breaks) + 1:
            return fail("INVALID_PARAM", f"graduated 需要 {len(breaks) + 1} 个颜色")
    for c in colors or []:
        if not (len(c) == 7 and c[0] == "#" and set(c[1:]) <= HEX):
            return fail("INVALID_PARAM", f"颜色 {c} 不是 7 位十六进制")
    return ok({"layer_id": layer_id, "method": method, "field": field,
               "breaks": breaks, "colors": colors, "rule_id": "rr_auto"})

# eval_agent.py —— 逐条回放评测集，输出 CSV；模型与工具版本固定后结果才可比
import csv, json, sys
from agent import build_agent          # 你的 Qwen-Agent 组装函数
from tools.registry import ToolRegistry

def score_one(case: dict, trace: dict) -> dict:
    used = [s["tool"] for s in trace["steps"]]
    expect = case["expect_tools"]
    it = iter(used)
    seq_ok = all(any(t == u for u in it) for t in expect)   # 子序列判定
    params_ok = all(
        any(s["tool"] == expect[0] and all(str(s["args"].get(k)) == str(v) for k, v in case.get("expect_params", {}).items())
            for s in trace["steps"])
    )
    end_ok = trace.get("final", {}).get("ok", False)
    return {"id": case["id"], "seq_ok": seq_ok, "params_ok": params_ok,
            "end_ok": end_ok, "steps": len(used), "tags": ";".join(case.get("tags", []))}

def main(set_path: str, out_path: str) -> None:
    cases = [json.loads(line) for line in open(set_path, encoding="utf-8") if line.strip()]
    agent, registry = build_agent()              # 注入受控工具
    rows = []
    for case in cases:
        trace = agent.run(case["text"])           # 内部走 4.2 的循环，记录每一步
        rows.append(score_one(case, trace))
    with open(out_path, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=list(rows[0].keys()))
        writer.writeheader(); writer.writerows(rows)
    n = len(rows)
    print(f"seq={sum(r['seq_ok'] for r in rows)/n:.3f} "
          f"params={sum(r['params_ok'] for r in rows)/n:.3f} "
          f"end2end={sum(r['end_ok'] for r in rows)/n:.3f} "
          f"avg_steps={sum(r['steps'] for r in rows)/n:.2f}")

if __name__ == "__main__":
    main(sys.argv[1], sys.argv[2])

# budget.py —— 单请求预算，超了就降级，不要"再试一次"
from dataclasses import dataclass

@dataclass
class Budget:
    max_steps: int = 6
    max_tokens: int = 20000
    max_tool_ms: int = 15000
    used_steps: int = 0
    used_tokens: int = 0
    used_ms: int = 0

    def allow_step(self) -> bool:
        return self.used_steps < self.max_steps and self.used_tokens < self.max_tokens

    def spend(self, tokens: int, ms: int) -> None:
        self.used_steps += 1
        self.used_tokens += tokens
        self.used_ms += ms