Deploy RouteOpt Agent

.env.example

@@ -0,0 +1,28 @@
+# 推荐：自动选择已配置 key 的提供商；没有 key 时退回 Pollinations。
+LLM_PROVIDER=auto
+
+# 如果本机代理导致 Python requests 报 ProxyError，保持 1。
+# 如果你的网络必须走代理才能访问外网，改成 0。
+LLM_IGNORE_PROXY=1
+
+# 方案 A：Groq，国际网络通常比较稳，OpenAI-compatible。
+# GROQ_API_KEY=gsk_xxx
+# GROQ_MODEL=openai/gpt-oss-20b
+
+# 方案 B：SiliconFlow，国内网络通常更友好，OpenAI-compatible。
+# SILICONFLOW_API_KEY=sk-xxx
+# SILICONFLOW_MODEL=THUDM/GLM-Z1-9B-0414
+
+# 方案 C：Google Gemini，官方免费额度，但国内网络可能需要代理。
+# GEMINI_API_KEY=AIza...
+# GEMINI_MODEL=gemini-2.5-flash
+
+# 方案 D：OpenRouter，可选免费模型，但不同模型 tool calling 支持不完全一致。
+# OPENROUTER_API_KEY=sk-or-v1-xxx
+# OPENROUTER_MODEL=qwen/qwen3-32b:free
+
+# 方案 E：任意 OpenAI-compatible 服务。
+# LLM_PROVIDER=custom
+# OPENAI_COMPATIBLE_BASE_URL=https://example.com/v1/chat/completions
+# LLM_API_KEY=sk-xxx
+# LLM_MODEL=your-model-name

.gitignore

@@ -0,0 +1,8 @@
+__pycache__/
+*.py[cod]
+.DS_Store
+.env
+cache/
+outputs/*.pdf
+outputs/*.png
+!outputs/.gitkeep

README.md

@@ -1,14 +1,162 @@
 ---
-title: Alg Agent
-emoji: 🏢
-colorFrom: indigo
+title: RouteOpt Agent
+emoji: 🧭
+colorFrom: blue
 colorTo: green
 sdk: gradio
-sdk_version: 6.19.0
-python_version: '3.13'
+python_version: 3.11
 app_file: app.py
 pinned: false
-short_description: 通用问题智能体-算法分析与设计课程作业
+short_description: Tool-calling route optimizer
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# RouteOpt Agent
+
+基于公开大模型 API + 轻量工具调用的小规模路线优化智能体。它把用户自然语言转成路线优化任务，调用地理编码和路径矩阵工具，再用本地算法求解访问顺序，并自动生成 PDF 报告。
+
+## 一键运行
+
+```bash
+chmod +x run.sh
+./run.sh
+```
+
+启动后打开：
+
+```text
+http://localhost:7860
+```
+
+首次运行会创建 conda 环境 `routeopt-agent` 并安装依赖。脚本会先尝试使用 `environment.yml` 创建标准环境；如果 conda 镜像下载不稳定，会自动克隆本机 base 环境，再用 pip 安装 `requirements.txt`。
+
+## LLM 配置
+
+默认配置不需要你注册任何网站，也不需要提供 API token：
+
+- LLM：Pollinations OpenAI-compatible API，无需 token。失败时系统自动切换到本地解析和本地报告模板。
+- 地理编码：OpenStreetMap Nominatim 公共接口，内置缓存并限制请求频率。
+- 路径矩阵：OSRM public demo server。失败时自动切换到直线距离近似矩阵。
+- 部署：Hugging Face Spaces 免费 Gradio Space。
+
+如果本机出现 `ProxyError('Unable to connect to proxy')`，说明 Python 请求走了不可用代理。默认 `LLM_IGNORE_PROXY=1` 会让 LLM 请求忽略系统代理；如果你的网络必须使用代理访问外网，可在 `.env` 中改成：
+
+```bash
+LLM_IGNORE_PROXY=0
+```
+
+更稳定的做法是使用正规 API key 服务，并把 key 放到 `.env` 或 Hugging Face Secrets。老师访问公开页面时不需要知道 token。
+
+复制配置模板：
+
+```bash
+cp .env.example .env
+```
+
+推荐方案：
+
+```bash
+# Groq：国际网络通常较稳，OpenAI-compatible
+LLM_PROVIDER=groq
+GROQ_API_KEY=gsk_xxx
+GROQ_MODEL=openai/gpt-oss-20b
+```
+
+```bash
+# SiliconFlow：国内网络通常更友好，OpenAI-compatible
+LLM_PROVIDER=siliconflow
+SILICONFLOW_API_KEY=sk-xxx
+SILICONFLOW_MODEL=THUDM/GLM-Z1-9B-0414
+```
+
+```bash
+# Gemini：Google 官方免费额度，国内网络可能需要代理
+LLM_PROVIDER=gemini
+GEMINI_API_KEY=AIza...
+GEMINI_MODEL=gemini-2.5-flash
+```
+
+也可以用 `LLM_PROVIDER=auto`，程序会按已配置的 key 自动选择：Groq -> SiliconFlow -> Gemini -> OpenRouter；没有 key 时退回 Pollinations。
+
+## 推荐演示输入
+
+自然语言需求：
+
+```text
+我从上海交通大学闵行校区出发，想一天内逛完徐家汇、人民广场、外滩、陆家嘴，最后回到起点，尽量总时间短。
+```
+
+起点：
+
+```text
+上海交通大学闵行校区
+```
+
+目的地：
+
+```text
+徐家汇
+人民广场
+外滩
+陆家嘴
+```
+
+这个样例的上海地点内置了离线坐标兜底，所以即使 Nominatim 临时限流，也能完成演示。
+
+也可以使用北京样例：
+
+```text
+我从北京邮电大学出发，想一天内逛完国家大剧院、鼓楼、北海、国家植物园，最后回到起点，尽量总时间短。
+```
+
+## 输入失败时如何处理
+
+页面不会直接抛出 Python 错误；如果任务无法完成，会说明失败发生在哪个工具，以及应该如何修改输入。
+
+常见原因：
+
+- 起点或目的地为空。
+- 目的地里只有起点，或者重复地点过多。
+- 地点不在内置演示坐标中，同时在线地理编码服务超时、限流或找不到结果。
+- 地名太泛，例如只写“鼓楼”“植物园”，在某些城市可能有多个匹配。
+
+当前内置演示城市：上海、北京。
+
+如果要演示其他城市，建议把地点写得更完整，例如“南京夫子庙”“广州塔”“深圳大学粤海校区”。如果在线地理编码仍然失败，可以直接输入坐标：
+
+```text
+某景点@39.9042,116.4074
+```
+
+## Agent Workflow
+
+1. `LLM tool_call: submit_route_task`：把自然语言和表单提示抽取成结构化任务。
+2. `geocode_places`：把地点转换为经纬度。
+3. `build_route_matrix`：调用 OSRM 获取距离/时间矩阵。
+4. `solve_route`：用 Held-Karp 动态规划精确求解小规模 TSP。
+5. `LLM compose_summary`：生成中文解释。
+6. `generate_pdf_report`：输出包含问题定义、工具调用轨迹、算法和结果的 PDF。
+
+## 公开部署到 Hugging Face Spaces
+
+1. 打开 Hugging Face，新建一个 Space。
+2. Space SDK 选择 `Gradio`。
+3. Visibility 可选 `Public`。
+4. 上传本仓库所有文件，或把仓库推送到 Space。
+5. Space 会根据 `requirements.txt` 自动安装依赖，入口文件是 `app.py`。
+
+老师访问 Space 链接即可运行，不需要知道任何 token。
+
+## 文件结构
+
+```text
+app.py                     # Gradio 页面入口
+routeopt_agent/agent.py    # Agent 编排和工具调用轨迹
+routeopt_agent/geo_tools.py# 地理编码、OSRM 矩阵、兜底矩阵
+routeopt_agent/solver.py   # Held-Karp 路线优化
+routeopt_agent/report.py   # PDF 报告生成
+routeopt_agent/parsing.py  # 本地自然语言解析兜底
+routeopt_agent/viz.py      # SVG 路线示意图
+environment.yml            # 本地 conda 环境
+requirements.txt           # Hugging Face Spaces 依赖
+run.sh                     # 一键运行脚本
+```

app.py

@@ -0,0 +1,254 @@
+from __future__ import annotations
+
+import os
+import socket
+import traceback
+from typing import Any
+
+import gradio as gr
+
+from routeopt_agent import RouteOptAgent
+from routeopt_agent.geo_tools import builtin_city_summary, builtin_place_examples
+from routeopt_agent.llm_client import get_llm_config
+from routeopt_agent.solver import format_km, format_minutes
+
+
+agent = RouteOptAgent()
+
+
+SAMPLE_REQUEST = "我从上海交通大学闵行校区出发，想一天内逛完徐家汇、人民广场、外滩、陆家嘴，最后回到起点，尽量总时间短。"
+SAMPLE_DESTINATIONS = "徐家汇\n人民广场\n外滩\n陆家嘴"
+
+
+def run_route_agent(
+    raw_request: str,
+    start_place: str,
+    destination_places: str,
+    objective_label: str,
+    return_to_start: bool,
+    fixed_end_place: str,
+    city_hint: str,
+    use_llm: bool,
+) -> tuple[Any, ...]:
+    objective = "distance" if "距离" in objective_label else "time"
+    try:
+        result = agent.run(
+            raw_request=raw_request,
+            start_hint=start_place,
+            destinations_hint=destination_places,
+            objective_hint=objective,
+            return_to_start_hint=return_to_start,
+            fixed_end_hint=fixed_end_place,
+            city_hint=city_hint,
+            use_llm=use_llm,
+        )
+    except Exception as exc:
+        error_md = build_user_error_message(exc, city_hint)
+        return error_md, "", [], [], [], trace_to_rows(agent.trace), None
+
+    solution = result.solution
+    warning_md = ""
+    if result.warnings:
+        warning_md = "\n\n**运行提示**\n" + "\n".join(f"- {item}" for item in result.warnings)
+
+    summary = (
+        f"**最优访问顺序**：{' → '.join(solution.route_names)}\n\n"
+        f"**总距离**：{format_km(solution.total_distance_meters)}  \n"
+        f"**预计驾驶时间**：{format_minutes(solution.total_duration_seconds)}  \n"
+        f"**优化算法**：{solution.algorithm}\n\n"
+        f"{result.summary_markdown}"
+        f"{warning_md}"
+    )
+
+    route_rows = [[idx + 1, name] for idx, name in enumerate(solution.route_names)]
+    point_rows = [
+        [idx, point.name, f"{point.lat:.6f}", f"{point.lon:.6f}", point.source, point.display_name]
+        for idx, point in enumerate(result.points)
+    ]
+    trace_rows = [
+        [
+            event.step,
+            event.tool,
+            event.status,
+            compact_for_ui(event.arguments),
+            event.result,
+        ]
+        for event in result.trace
+    ]
+
+    return (
+        summary,
+        result.route_svg,
+        route_rows,
+        solution.leg_rows,
+        point_rows,
+        trace_rows,
+        result.pdf_path,
+    )
+
+
+def compact_for_ui(value: Any) -> str:
+    text = str(value)
+    if len(text) > 260:
+        return text[:260] + "..."
+    return text
+
+
+def trace_to_rows(trace: list[Any]) -> list[list[Any]]:
+    return [
+        [
+            event.step,
+            event.tool,
+            event.status,
+            compact_for_ui(event.arguments),
+            event.result,
+        ]
+        for event in trace
+    ]
+
+
+def build_user_error_message(exc: Exception, city_hint: str) -> str:
+    message = str(exc).strip() or exc.__class__.__name__
+    suggestions = [
+        "确认起点和目的地都已填写，且目的地至少有一个不同于起点的地点。",
+        "地点名尽量写完整，例如加上城市、区县、校区或景区全称。",
+        f"当前内置演示坐标覆盖城市：{builtin_city_summary()}。如果输入其他城市，系统会依赖在线地理编码，可能受网络、限流或地名歧义影响。",
+        "如果在线地理编码不稳定，可以把地点写成 `地点名@纬度,经度`，例如 `某景点@39.9042,116.4074`。",
+    ]
+    if not city_hint.strip():
+        suggestions.insert(1, "城市/区域提示为空。建议填写类似 `北京，中国` 或 `上海，中国`，能显著降低地点歧义。")
+
+    return (
+        "**本次没有完成路线优化，但系统已定位到失败原因。**\n\n"
+        f"**失败原因**：\n\n{message}\n\n"
+        "**可以这样处理**：\n"
+        + "\n".join(f"- {item}" for item in suggestions)
+        + "\n\n"
+        "<details><summary>技术调试信息</summary>\n\n"
+        f"```text\n{traceback.format_exc()[-1800:]}\n```\n</details>"
+    )
+
+
+def runtime_status_text() -> str:
+    try:
+        config = get_llm_config()
+        key_text = "已配置 key" if config.api_key else "无 key"
+        llm_text = f"{config.provider} / {config.model}（{key_text}）"
+    except Exception as exc:
+        llm_text = f"LLM 配置异常：{exc}"
+    return (
+        f"当前 LLM：`{llm_text}`  \n"
+        f"内置演示城市：`{builtin_city_summary()}`  \n"
+        f"内置地点示例：{builtin_place_examples()}  \n"
+        "未知地点可用手动坐标格式：`地点名@纬度,经度`。"
+    )
+
+
+with gr.Blocks(title="RouteOpt Agent") as demo:
+    gr.Markdown(
+        "# RouteOpt Agent\n"
+        "基于公开大模型 API + 轻量工具调用的小规模路线优化智能体。"
+    )
+    gr.Markdown(runtime_status_text())
+    with gr.Row():
+        with gr.Column(scale=4):
+            raw_request = gr.Textbox(
+                label="自然语言需求",
+                value=SAMPLE_REQUEST,
+                lines=4,
+                placeholder="例如：我从交大闵行出发，想逛完外滩、陆家嘴、人民广场，最后回到起点，尽量总时间短。",
+            )
+            start_place = gr.Textbox(label="起点", value="上海交通大学闵行校区")
+            destination_places = gr.Textbox(
+                label="目的地（每行一个，建议 3-8 个）",
+                value=SAMPLE_DESTINATIONS,
+                lines=6,
+            )
+            with gr.Row():
+                objective = gr.Radio(
+                    label="优化目标",
+                    choices=["最短时间", "最短距离"],
+                    value="最短时间",
+                )
+                return_to_start = gr.Checkbox(label="最后回到起点", value=True)
+            with gr.Row():
+                fixed_end_place = gr.Textbox(label="固定终点（不回起点时可填）", value="")
+                city_hint = gr.Textbox(label="城市/区域提示", value="上海，中国")
+            use_llm = gr.Checkbox(label="启用 LLM 工具调用解析与总结", value=True)
+            run_button = gr.Button("开始优化并生成 PDF", variant="primary")
+
+        with gr.Column(scale=6):
+            summary = gr.Markdown(label="求解结果")
+            route_svg = gr.HTML(label="路线示意图")
+
+    with gr.Tab("路线"):
+        route_table = gr.Dataframe(
+            headers=["顺序", "地点"],
+            datatype=["number", "str"],
+            interactive=False,
+        )
+        legs_table = gr.Dataframe(
+            headers=["段", "从", "到", "距离", "时间"],
+            datatype=["number", "str", "str", "str", "str"],
+            interactive=False,
+        )
+    with gr.Tab("工具调用"):
+        trace_table = gr.Dataframe(
+            headers=["步骤", "工具", "状态", "参数", "结果摘要"],
+            datatype=["number", "str", "str", "str", "str"],
+            interactive=False,
+            wrap=True,
+        )
+    with gr.Tab("地理编码"):
+        points_table = gr.Dataframe(
+            headers=["序号", "地点", "纬度", "经度", "数据源", "匹配名称"],
+            datatype=["number", "str", "str", "str", "str", "str"],
+            interactive=False,
+            wrap=True,
+        )
+    with gr.Tab("报告"):
+        pdf_file = gr.File(label="PDF 报告")
+
+    run_button.click(
+        run_route_agent,
+        inputs=[
+            raw_request,
+            start_place,
+            destination_places,
+            objective,
+            return_to_start,
+            fixed_end_place,
+            city_hint,
+            use_llm,
+        ],
+        outputs=[
+            summary,
+            route_svg,
+            route_table,
+            legs_table,
+            points_table,
+            trace_table,
+            pdf_file,
+        ],
+    )
+
+
+def find_available_port(start_port: int, attempts: int = 20) -> int:
+    for port in range(start_port, start_port + attempts):
+        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+            try:
+                sock.bind(("0.0.0.0", port))
+                return port
+            except OSError:
+                continue
+    return start_port
+
+
+if __name__ == "__main__":
+    requested_port = int(os.getenv("GRADIO_SERVER_PORT") or os.getenv("PORT") or "7860")
+    server_name = os.getenv("GRADIO_SERVER_NAME")
+    if not server_name:
+        server_name = "0.0.0.0" if os.getenv("SPACE_ID") else "127.0.0.1"
+    port = find_available_port(requested_port)
+    print(f"RouteOpt Agent is starting on http://localhost:{port}")
+    demo.queue().launch(server_name=server_name, server_port=port)

docs/作业要求.txt

@@ -0,0 +1,13 @@
+基于公开可访问的大模型 API / 网页端（如豆包 、GPT、Claude、通义千问等，禁止使用内部未公开模型），通过提示词工程 + 轻量工具调用等方式，训练一个通用问题优化智能体
+
+
+
+1、智能体要可公开访问；
+
+2、智能体要解决一个问题，解决过程要录制视频MP4；
+
+3、智能体求解过程要自己输出一个详细文档PDF；
+
+
+
+提交：可访问链接+MP4视频+PDF文档。

environment.yml

@@ -0,0 +1,10 @@
+name: routeopt-agent
+channels:
+  - conda-forge
+dependencies:
+  - python=3.11
+  - pip
+  - pip:
+      - gradio>=4.44.0
+      - requests>=2.31.0
+      - reportlab>=4.2.0

outputs/.gitkeep

@@ -0,0 +1 @@
+

requirements.txt

@@ -0,0 +1,3 @@
+gradio>=4.44.0
+requests>=2.31.0
+reportlab>=4.2.0

routeopt_agent/__init__.py

@@ -0,0 +1,5 @@
+"""RouteOpt Agent package."""
+
+from .agent import RouteOptAgent
+
+__all__ = ["RouteOptAgent"]

routeopt_agent/agent.py

@@ -0,0 +1,438 @@
+from __future__ import annotations
+
+import json
+from typing import Any, Callable
+
+from .geo_tools import build_route_matrix, geocode_places
+from .llm_client import LLMError, chat_completion, extract_json_from_text, extract_tool_arguments
+from .models import AgentResult, GeoPoint, Objective, RouteMatrix, RouteSolution, RouteTask, ToolEvent
+from .parsing import dedupe_preserve_order, heuristic_extract_task, normalize_place_lines
+from .report import generate_pdf_report
+from .solver import format_km, format_minutes, solve_route
+from .viz import build_route_svg
+
+
+TASK_TOOL_SCHEMA: dict[str, Any] = {
+    "type": "function",
+    "function": {
+        "name": "submit_route_task",
+        "description": "Extract a route optimization task from user text and structured hints.",
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "start_place": {
+                    "type": "string",
+                    "description": "The route start place. Keep it specific enough for geocoding.",
+                },
+                "destination_places": {
+                    "type": "array",
+                    "items": {"type": "string"},
+                    "description": "Places that must be visited after the start.",
+                },
+                "objective": {
+                    "type": "string",
+                    "enum": ["time", "distance"],
+                    "description": "Optimization objective: time or distance.",
+                },
+                "return_to_start": {
+                    "type": "boolean",
+                    "description": "Whether the route should return to the start place.",
+                },
+                "fixed_end_place": {
+                    "type": "string",
+                    "description": "Optional fixed final destination when not returning to start.",
+                },
+                "constraints": {
+                    "type": "array",
+                    "items": {"type": "string"},
+                    "description": "Soft constraints or notes mentioned by the user.",
+                },
+            },
+            "required": ["start_place", "destination_places", "objective", "return_to_start"],
+        },
+    },
+}
+
+
+class RouteOptAgent:
+    def __init__(self) -> None:
+        self.trace: list[ToolEvent] = []
+
+    def run(
+        self,
+        raw_request: str,
+        start_hint: str = "",
+        destinations_hint: str = "",
+        objective_hint: str = "time",
+        return_to_start_hint: bool = True,
+        fixed_end_hint: str = "",
+        city_hint: str = "上海，中国",
+        use_llm: bool = True,
+    ) -> AgentResult:
+        self.trace = []
+        warnings: list[str] = []
+
+        task = self._extract_task(
+            raw_request=raw_request,
+            start_hint=start_hint,
+            destinations_hint=destinations_hint,
+            objective_hint=objective_hint,
+            return_to_start_hint=return_to_start_hint,
+            fixed_end_hint=fixed_end_hint,
+            use_llm=use_llm,
+            warnings=warnings,
+        )
+        self._validate_task(task, warnings)
+
+        all_places = [task.start_place] + task.destination_places
+        points = self._record_tool(
+            "geocode_places",
+            {"places": all_places, "city_hint": city_hint},
+            lambda: geocode_places(all_places, city_hint),
+            lambda result: f"解析 {len(result)} 个地点：{', '.join(point.name for point in result)}",
+        )
+        matrix = self._record_tool(
+            "build_route_matrix",
+            {"points": [point.name for point in points]},
+            lambda: build_route_matrix(points),
+            lambda result: f"获得 {len(result.points)}x{len(result.points)} 距离/时间矩阵，来源：{result.source}",
+        )
+        solution = self._record_tool(
+            "solve_route",
+            {
+                "objective": task.objective,
+                "return_to_start": task.return_to_start,
+                "fixed_end_place": task.fixed_end_place,
+            },
+            lambda: solve_route(matrix, task.objective, task.return_to_start, task.fixed_end_place),
+            lambda result: f"路线：{' -> '.join(result.route_names)}；距离 {format_km(result.total_distance_meters)}；时间 {format_minutes(result.total_duration_seconds)}",
+        )
+
+        summary = self._compose_summary(task, points, matrix, solution, use_llm, warnings)
+        route_svg = build_route_svg(points, solution.route_indices)
+        pdf_path = self._record_tool(
+            "generate_pdf_report",
+            {"route": solution.route_names},
+            lambda: generate_pdf_report(task, points, solution, self.trace, summary),
+            lambda result: f"PDF 已生成：{result}",
+        )
+
+        return AgentResult(
+            task=task,
+            points=points,
+            matrix=matrix,
+            solution=solution,
+            summary_markdown=summary,
+            trace=self.trace,
+            pdf_path=pdf_path,
+            route_svg=route_svg,
+            warnings=warnings,
+        )
+
+    def _extract_task(
+        self,
+        raw_request: str,
+        start_hint: str,
+        destinations_hint: str,
+        objective_hint: str,
+        return_to_start_hint: bool,
+        fixed_end_hint: str,
+        use_llm: bool,
+        warnings: list[str],
+    ) -> RouteTask:
+        if use_llm:
+            try:
+                task = self._extract_task_with_llm(
+                    raw_request,
+                    start_hint,
+                    destinations_hint,
+                    objective_hint,
+                    return_to_start_hint,
+                    fixed_end_hint,
+                )
+                self._append_event(
+                    "LLM tool_call: submit_route_task",
+                    {
+                        "raw_request": raw_request,
+                        "start_hint": start_hint,
+                        "destinations_hint": destinations_hint,
+                    },
+                    "ok",
+                    f"抽取任务：起点={task.start_place}；目的地={len(task.destination_places)} 个；目标={task.objective}",
+                )
+                return task
+            except Exception as exc:
+                warnings.append(f"LLM 工具调用抽取失败，已切换本地解析：{exc}")
+                self._append_event(
+                    "LLM tool_call: submit_route_task",
+                    {"raw_request": raw_request},
+                    "fallback",
+                    str(exc),
+                )
+
+        task = heuristic_extract_task(
+            raw_request,
+            start_hint=start_hint,
+            destinations_hint=destinations_hint,
+            objective_hint=objective_hint,
+            return_to_start_hint=return_to_start_hint,
+            fixed_end_hint=fixed_end_hint,
+        )
+        self._append_event(
+            "local_parse_route_task",
+            {
+                "raw_request": raw_request,
+                "start_hint": start_hint,
+                "destinations_hint": destinations_hint,
+            },
+            "ok",
+            f"本地解析任务：起点={task.start_place}；目的地={len(task.destination_places)} 个；目标={task.objective}",
+        )
+        return task
+
+    def _extract_task_with_llm(
+        self,
+        raw_request: str,
+        start_hint: str,
+        destinations_hint: str,
+        objective_hint: str,
+        return_to_start_hint: bool,
+        fixed_end_hint: str,
+    ) -> RouteTask:
+        messages = [
+            {
+                "role": "system",
+                "content": (
+                    "You are a route optimization agent controller. "
+                    "Use the submit_route_task function to return one structured task. "
+                    "Prefer structured hints over ambiguous natural language. "
+                    "Keep Chinese place names specific for geocoding. "
+                    "If the user says 最快/时间最短 use objective=time; if 距离/少走路 use objective=distance."
+                ),
+            },
+            {
+                "role": "user",
+                "content": json.dumps(
+                    {
+                        "raw_request": raw_request,
+                        "start_hint": start_hint,
+                        "destinations_hint": normalize_place_lines(destinations_hint),
+                        "objective_hint": objective_hint,
+                        "return_to_start_hint": return_to_start_hint,
+                        "fixed_end_hint": fixed_end_hint,
+                    },
+                    ensure_ascii=False,
+                ),
+            },
+        ]
+        message = chat_completion(messages, tools=[TASK_TOOL_SCHEMA], temperature=0.1)
+        args = extract_tool_arguments(message, "submit_route_task")
+        if args is None:
+            args = extract_json_from_text(message.get("content") or "")
+        if args is None:
+            raise LLMError("模型没有返回 submit_route_task 工具参数。")
+        return self._task_from_arguments(
+            args,
+            raw_request,
+            start_hint,
+            destinations_hint,
+            objective_hint,
+            return_to_start_hint,
+            fixed_end_hint,
+        )
+
+    def _task_from_arguments(
+        self,
+        args: dict[str, Any],
+        raw_request: str,
+        start_hint: str,
+        destinations_hint: str,
+        objective_hint: str,
+        return_to_start_hint: bool,
+        fixed_end_hint: str,
+    ) -> RouteTask:
+        local_task = heuristic_extract_task(
+            raw_request,
+            start_hint=start_hint,
+            destinations_hint=destinations_hint,
+            objective_hint=objective_hint,
+            return_to_start_hint=return_to_start_hint,
+            fixed_end_hint=fixed_end_hint,
+        )
+
+        destinations = args.get("destination_places") or local_task.destination_places
+        if isinstance(destinations, str):
+            destinations = normalize_place_lines(destinations)
+        destinations = dedupe_preserve_order([str(item).strip() for item in destinations if str(item).strip()])
+
+        start = start_hint.strip() or str(args.get("start_place") or local_task.start_place).strip()
+        objective = normalize_objective(str(args.get("objective") or local_task.objective))
+        fixed_end = fixed_end_hint.strip() or str(args.get("fixed_end_place") or local_task.fixed_end_place or "").strip()
+
+        return RouteTask(
+            raw_request=raw_request,
+            start_place=start,
+            destination_places=destinations,
+            objective=objective,
+            return_to_start=return_to_start_hint if return_to_start_hint is not None else bool(args.get("return_to_start", local_task.return_to_start)),
+            fixed_end_place=fixed_end or None,
+            constraints=[str(item) for item in args.get("constraints") or local_task.constraints],
+        )
+
+    def _validate_task(self, task: RouteTask, warnings: list[str]) -> None:
+        task.start_place = task.start_place.strip()
+        task.destination_places = [place.strip() for place in task.destination_places if place.strip()]
+        original_count = len(task.destination_places)
+        task.destination_places = dedupe_preserve_order(task.destination_places)
+        if len(task.destination_places) < original_count:
+            warnings.append("目的地中存在重复项，系统已自动去重。")
+
+        if not task.start_place:
+            raise ValueError("缺少起点。请在起点输入框填写一个地点，或在自然语言需求中写明“从哪里出发”。")
+        if not task.destination_places:
+            raise ValueError("缺少目的地。请至少填写 1 个目的地。建议每行写一个地点，例如：人民广场、外滩、陆家嘴。")
+
+        start_key = task.start_place.lower()
+        without_start = [
+            place
+            for place in task.destination_places
+            if place.lower() != start_key and start_key not in place.lower() and place.lower() not in start_key
+        ]
+        if len(without_start) < len(task.destination_places):
+            warnings.append("目的地中包含起点，系统已自动移除该重复访问点。")
+            task.destination_places = without_start
+        if not task.destination_places:
+            raise ValueError("目的地里只有起点本身。请至少增加一个不同于起点的目的地。")
+
+        if len(task.destination_places) > 10:
+            raise ValueError(
+                "目的地数量过多。当前演示版最多支持 10 个目的地；作业视频推荐 3 到 8 个，"
+                "这样 Held-Karp 精确算法和公开路线 API 都更稳定。"
+            )
+
+        if task.fixed_end_place and not task.return_to_start:
+            fixed = task.fixed_end_place.strip().lower()
+            if all(fixed not in item.lower() and item.lower() not in fixed for item in task.destination_places):
+                task.destination_places.append(task.fixed_end_place)
+
+    def _compose_summary(
+        self,
+        task: RouteTask,
+        points: list[GeoPoint],
+        matrix: RouteMatrix,
+        solution: RouteSolution,
+        use_llm: bool,
+        warnings: list[str],
+    ) -> str:
+        if use_llm:
+            try:
+                messages = [
+                    {
+                        "role": "system",
+                        "content": (
+                            "你是算法课程作业里的路线优化智能体。"
+                            "请用中文写一段简洁但不敷衍的求解说明，强调："
+                            "大模型负责理解和解释，工具负责地理编码/路径矩阵，算法负责最优化。"
+                            "不要编造没有出现的数据。"
+                        ),
+                    },
+                    {
+                        "role": "user",
+                        "content": json.dumps(
+                            {
+                                "task": task.__dict__,
+                                "points": [point.__dict__ for point in points],
+                                "matrix_source": matrix.source,
+                                "solution": {
+                                    "route": solution.route_names,
+                                    "total_distance_km": round(solution.total_distance_meters / 1000, 2),
+                                    "total_minutes": round(solution.total_duration_seconds / 60, 1),
+                                    "algorithm": solution.algorithm,
+                                },
+                                "tool_trace": [event.__dict__ for event in self.trace],
+                                "warnings": warnings,
+                            },
+                            ensure_ascii=False,
+                        ),
+                    },
+                ]
+                message = chat_completion(messages, temperature=0.35)
+                content = (message.get("content") or "").strip()
+                if content:
+                    self._append_event(
+                        "LLM compose_summary",
+                        {"route": solution.route_names},
+                        "ok",
+                        "已生成中文解释总结。",
+                    )
+                    return content
+            except Exception as exc:
+                warnings.append(f"LLM 总结失败，已切换本地报告模板：{exc}")
+                self._append_event("LLM compose_summary", {}, "fallback", str(exc))
+
+        return deterministic_summary(task, matrix, solution, warnings)
+
+    def _record_tool(
+        self,
+        name: str,
+        arguments: dict[str, Any],
+        func: Callable[[], Any],
+        summarize: Callable[[Any], str],
+    ) -> Any:
+        try:
+            result = func()
+            self._append_event(name, arguments, "ok", summarize(result))
+            return result
+        except Exception as exc:
+            self._append_event(name, arguments, "error", str(exc))
+            raise
+
+    def _append_event(self, tool: str, arguments: dict[str, Any], status: str, result: str) -> None:
+        self.trace.append(
+            ToolEvent(
+                step=len(self.trace) + 1,
+                tool=tool,
+                arguments=compact_arguments(arguments),
+                status=status,
+                result=result,
+            )
+        )
+
+
+def normalize_objective(value: str) -> Objective:
+    value = value.lower().strip()
+    if value in {"distance", "最短距离", "距离"}:
+        return "distance"
+    return "time"
+
+
+def compact_arguments(arguments: dict[str, Any]) -> dict[str, Any]:
+    compacted: dict[str, Any] = {}
+    for key, value in arguments.items():
+        if isinstance(value, list) and len(value) > 8:
+            compacted[key] = value[:8] + ["..."]
+        else:
+            compacted[key] = value
+    return compacted
+
+
+def deterministic_summary(
+    task: RouteTask,
+    matrix: RouteMatrix,
+    solution: RouteSolution,
+    warnings: list[str],
+) -> str:
+    objective_text = "预计驾驶时间" if task.objective == "time" else "路线距离"
+    warning_text = ""
+    if warnings:
+        warning_text = "\n\n**运行提示**：\n" + "\n".join(f"- {item}" for item in warnings)
+    return (
+        "本次求解把用户需求先转成结构化路线优化任务，然后调用地理编码工具获得经纬度，"
+        f"再通过 `{matrix.source}` 得到点对点距离/时间矩阵。最后，本地优化器以“{objective_text}”为目标，"
+        f"使用 `{solution.algorithm}` 搜索访问顺序。\n\n"
+        f"最终路线为：**{' → '.join(solution.route_names)}**。"
+        f"总距离约 **{format_km(solution.total_distance_meters)}**，预计驾驶时间约 **{format_minutes(solution.total_duration_seconds)}**。"
+        "\n\n这个设计中，大模型不直接猜最优路线，而是负责理解自然语言、组织工具调用并解释结果；"
+        "确定性工具负责拿真实数据和完成可验证的算法计算。"
+        f"{warning_text}"
+    )

routeopt_agent/geo_tools.py

@@ -0,0 +1,300 @@
+from __future__ import annotations
+
+import json
+import math
+import os
+import re
+import time
+from pathlib import Path
+from typing import Any
+from urllib.parse import quote
+
+import requests
+
+from .models import GeoPoint, RouteMatrix
+
+
+NOMINATIM_URL = "https://nominatim.openstreetmap.org/search"
+OSRM_TABLE_URL = "https://router.project-osrm.org/table/v1/driving"
+CACHE_DIR = Path("cache")
+GEO_CACHE_PATH = CACHE_DIR / "geo_cache.json"
+BUILTIN_DEMO_CITIES = ("上海", "北京")
+
+
+DEMO_POINTS: dict[str, tuple[float, float, str]] = {
+    "上海交通大学闵行校区": (31.0256, 121.4332, "上海交通大学闵行校区"),
+    "交大闵行": (31.0256, 121.4332, "上海交通大学闵行校区"),
+    "上海交大闵行校区": (31.0256, 121.4332, "上海交通大学闵行校区"),
+    "外滩": (31.2400, 121.4900, "外滩"),
+    "陆家嘴": (31.2381, 121.4970, "陆家嘴"),
+    "人民广场": (31.2304, 121.4737, "人民广场"),
+    "徐家汇": (31.1838, 121.4328, "徐家汇"),
+    "豫园": (31.2272, 121.4922, "豫园"),
+    "静安寺": (31.2230, 121.4454, "静安寺"),
+    "南京东路": (31.2366, 121.4842, "南京东路"),
+    "东方明珠": (31.2397, 121.4998, "东方明珠广播电视塔"),
+    "上海虹桥站": (31.1945, 121.3188, "上海虹桥站"),
+    "上海南站": (31.1546, 121.4296, "上海南站"),
+    "复旦大学": (31.2989, 121.5038, "复旦大学邯郸校区"),
+    "同济大学": (31.2820, 121.5062, "同济大学四平路校区"),
+    "华东师范大学": (31.2282, 121.4038, "华东师范大学普陀校区"),
+    "北京邮电大学": (39.9620, 116.3585, "北京邮电大学西土城路校区"),
+    "北邮": (39.9620, 116.3585, "北京邮电大学西土城路校区"),
+    "北京邮电大学西土城路校区": (39.9620, 116.3585, "北京邮电大学西土城路校区"),
+    "国家大剧院": (39.9038, 116.3838, "国家大剧院"),
+    "北京国家大剧院": (39.9038, 116.3838, "国家大剧院"),
+    "鼓楼": (39.9404, 116.3972, "北京鼓楼"),
+    "北京鼓楼": (39.9404, 116.3972, "北京鼓楼"),
+    "鼓楼大街": (39.9470, 116.3938, "鼓楼大街"),
+    "北海": (39.9255, 116.3895, "北海公园"),
+    "北海公园": (39.9255, 116.3895, "北海公园"),
+    "国家植物园": (39.9974, 116.2072, "国家植物园"),
+    "北京国家植物园": (39.9974, 116.2072, "国家植物园"),
+    "北京植物园": (39.9974, 116.2072, "国家植物园"),
+}
+
+
+class GeoToolError(RuntimeError):
+    pass
+
+
+def geocode_places(places: list[str], city_hint: str = "", timeout: int = 12) -> list[GeoPoint]:
+    cache = load_json_cache(GEO_CACHE_PATH)
+    results: list[GeoPoint] = []
+    last_live_query_at = 0.0
+
+    for place in places:
+        place = place.strip()
+        if not place:
+            raise GeoToolError("地点列表中存在空项。请删除空行，确保起点和每个目的地都有具体名称。")
+
+        manual = parse_manual_point(place)
+        if manual:
+            point = manual
+            results.append(point)
+            continue
+
+        demo = lookup_demo_point(place)
+        if demo:
+            lat, lon, display_name = demo
+            point = GeoPoint(place, place, lat, lon, display_name, "built-in-demo")
+            cache_key = normalize_key(place, city_hint)
+            cache[cache_key] = point.__dict__
+            results.append(point)
+            continue
+
+        cache_key = normalize_key(place, city_hint)
+        if cache_key in cache:
+            cached = cache[cache_key]
+            results.append(GeoPoint(**cached))
+            continue
+
+        query = f"{place}, {city_hint}".strip(" ,")
+        sleep_for_nominatim(last_live_query_at)
+        last_live_query_at = time.time()
+        try:
+            point = geocode_with_nominatim(place, query, timeout)
+        except GeoToolError:
+            raise
+        except requests.exceptions.Timeout as exc:
+            raise GeoToolError(build_geocode_failure_message(place, query, "在线地理编码服务响应超时")) from exc
+        except requests.exceptions.ConnectionError as exc:
+            raise GeoToolError(build_geocode_failure_message(place, query, "无法连接在线地理编码服务")) from exc
+        except requests.exceptions.HTTPError as exc:
+            raise GeoToolError(build_geocode_failure_message(place, query, f"在线地理编码 HTTP 错误：{exc.response.status_code if exc.response else 'unknown'}")) from exc
+        except requests.exceptions.RequestException as exc:
+            raise GeoToolError(build_geocode_failure_message(place, query, f"在线地理编码请求失败：{exc}")) from exc
+        cache[cache_key] = point.__dict__
+        results.append(point)
+
+    save_json_cache(GEO_CACHE_PATH, cache)
+    return results
+
+
+def build_route_matrix(points: list[GeoPoint], timeout: int = 15) -> RouteMatrix:
+    if len(points) < 2:
+        raise GeoToolError("至少需要起点和 1 个目的地。")
+
+    coords = ";".join(f"{point.lon:.6f},{point.lat:.6f}" for point in points)
+    url = f"{OSRM_TABLE_URL}/{coords}"
+    params = {"annotations": "duration,distance"}
+
+    try:
+        response = requests.get(url, params=params, timeout=timeout)
+        response.raise_for_status()
+        payload = response.json()
+        durations = payload.get("durations")
+        distances = payload.get("distances")
+        if not matrix_is_complete(durations) or not matrix_is_complete(distances):
+            raise GeoToolError("OSRM 返回的矩阵不完整。")
+        return RouteMatrix(
+            points=points,
+            durations=durations,
+            distances=distances,
+            source="OSRM public demo server",
+        )
+    except Exception:
+        durations, distances = approximate_city_matrix(points)
+        return RouteMatrix(
+            points=points,
+            durations=durations,
+            distances=distances,
+            source="haversine fallback, speed=35km/h, road factor=1.30",
+        )
+
+
+def geocode_with_nominatim(place: str, query: str, timeout: int) -> GeoPoint:
+    headers = {
+        "User-Agent": "routeopt-agent-homework/1.0 (public classroom demo)",
+        "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.6",
+    }
+    params = {
+        "q": query,
+        "format": "jsonv2",
+        "limit": 1,
+        "addressdetails": 1,
+    }
+    response = requests.get(NOMINATIM_URL, params=params, headers=headers, timeout=timeout)
+    response.raise_for_status()
+    items = response.json()
+    if not items:
+        raise GeoToolError(build_geocode_failure_message(place, query, "在线地理编码没有找到匹配地点"))
+    item = items[0]
+    return GeoPoint(
+        name=place,
+        query=query,
+        lat=float(item["lat"]),
+        lon=float(item["lon"]),
+        display_name=item.get("display_name", query),
+        source="OpenStreetMap Nominatim",
+    )
+
+
+def lookup_demo_point(place: str) -> tuple[float, float, str] | None:
+    normalized = place.strip().lower()
+    for key, value in DEMO_POINTS.items():
+        key_norm = key.lower()
+        if normalized == key_norm or normalized in key_norm or key_norm in normalized:
+            return value
+    return None
+
+
+def parse_manual_point(place: str) -> GeoPoint | None:
+    patterns = [
+        r"^(?P<name>.+?)[@|]\s*(?P<lat>-?\d+(?:\.\d+)?)\s*[,，]\s*(?P<lon>-?\d+(?:\.\d+)?)$",
+        r"^(?P<name>.+?)[（(]\s*(?P<lat>-?\d+(?:\.\d+)?)\s*[,，]\s*(?P<lon>-?\d+(?:\.\d+)?)\s*[)）]$",
+    ]
+    for pattern in patterns:
+        match = re.match(pattern, place.strip())
+        if not match:
+            continue
+        name = match.group("name").strip()
+        lat = float(match.group("lat"))
+        lon = float(match.group("lon"))
+        if not name:
+            raise GeoToolError("手动坐标格式缺少地点名称。正确格式示例：未知景点@39.90,116.40")
+        if not (-90 <= lat <= 90 and -180 <= lon <= 180):
+            raise GeoToolError(f"手动坐标超出范围：{place}。纬度应在 -90 到 90，经度应在 -180 到 180。")
+        return GeoPoint(
+            name=name,
+            query=place,
+            lat=lat,
+            lon=lon,
+            display_name=f"{name}（手动坐标）",
+            source="manual-coordinate",
+        )
+    return None
+
+
+def build_geocode_failure_message(place: str, query: str, reason: str) -> str:
+    return (
+        f"地点解析失败：`{place}`。\n\n"
+        f"失败原因：{reason}。\n\n"
+        f"本次查询语句：`{query}`。\n\n"
+        f"当前内置演示坐标覆盖城市：{builtin_city_summary()}。如果你输入的是其他城市或较冷门地点，"
+        "系统会尝试在线地理编码；当在线服务超时、限流或找不到地点时，就需要你把地名写得更具体，"
+        "例如加上城市、区县、校区/景区全称。\n\n"
+        "也可以直接使用手动坐标格式绕过在线地理编码：`地点名@纬度,经度`，例如 `某景点@39.9042,116.4074`。"
+    )
+
+
+def builtin_city_summary() -> str:
+    return "、".join(BUILTIN_DEMO_CITIES)
+
+
+def builtin_place_examples(limit: int = 12) -> str:
+    names = list(DEMO_POINTS.keys())[:limit]
+    return "、".join(names)
+
+
+def approximate_city_matrix(points: list[GeoPoint]) -> tuple[list[list[float]], list[list[float]]]:
+    distances: list[list[float]] = []
+    durations: list[list[float]] = []
+    road_factor = 1.30
+    speed_mps = 35_000 / 3600
+    for origin in points:
+        distance_row: list[float] = []
+        duration_row: list[float] = []
+        for dest in points:
+            if origin is dest:
+                distance = 0.0
+            else:
+                distance = haversine_meters(origin.lat, origin.lon, dest.lat, dest.lon) * road_factor
+            distance_row.append(distance)
+            duration_row.append(distance / speed_mps if speed_mps else 0.0)
+        distances.append(distance_row)
+        durations.append(duration_row)
+    return durations, distances
+
+
+def haversine_meters(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
+    radius = 6_371_000
+    phi1 = math.radians(lat1)
+    phi2 = math.radians(lat2)
+    d_phi = math.radians(lat2 - lat1)
+    d_lambda = math.radians(lon2 - lon1)
+    a = (
+        math.sin(d_phi / 2) ** 2
+        + math.cos(phi1) * math.cos(phi2) * math.sin(d_lambda / 2) ** 2
+    )
+    return 2 * radius * math.atan2(math.sqrt(a), math.sqrt(1 - a))
+
+
+def matrix_is_complete(matrix: Any) -> bool:
+    if not isinstance(matrix, list) or not matrix:
+        return False
+    size = len(matrix)
+    return all(isinstance(row, list) and len(row) == size and all(value is not None for value in row) for row in matrix)
+
+
+def format_osm_link(point: GeoPoint) -> str:
+    return f"https://www.openstreetmap.org/?mlat={point.lat:.6f}&mlon={point.lon:.6f}#map=14/{point.lat:.6f}/{point.lon:.6f}"
+
+
+def make_osrm_route_link(points: list[GeoPoint], route_indices: list[int]) -> str:
+    coords = ";".join(f"{points[idx].lon:.6f},{points[idx].lat:.6f}" for idx in route_indices)
+    return f"https://router.project-osrm.org/route/v1/driving/{quote(coords, safe=';,')}"
+
+
+def sleep_for_nominatim(last_live_query_at: float) -> None:
+    elapsed = time.time() - last_live_query_at
+    if last_live_query_at and elapsed < 1.05:
+        time.sleep(1.05 - elapsed)
+
+
+def load_json_cache(path: Path) -> dict[str, Any]:
+    if not path.exists():
+        return {}
+    try:
+        return json.loads(path.read_text(encoding="utf-8"))
+    except Exception:
+        return {}
+
+
+def save_json_cache(path: Path, data: dict[str, Any]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
+
+
+def normalize_key(place: str, city_hint: str) -> str:
+    return f"{place.strip().lower()}|{city_hint.strip().lower()}"

routeopt_agent/llm_client.py

@@ -0,0 +1,187 @@
+from __future__ import annotations
+
+import json
+import os
+from dataclasses import dataclass
+from typing import Any
+
+import requests
+
+
+POLLINATIONS_OPENAI_URL = "https://text.pollinations.ai/openai"
+
+
+@dataclass(frozen=True)
+class LLMConfig:
+    provider: str
+    url: str
+    model: str
+    api_key: str | None = None
+    ignore_proxy: bool = True
+
+
+DEFAULT_MODELS = {
+    "pollinations": "openai-fast",
+    "groq": "openai/gpt-oss-20b",
+    "siliconflow": "THUDM/GLM-Z1-9B-0414",
+    "gemini": "gemini-2.5-flash",
+    "openrouter": "qwen/qwen3-32b:free",
+}
+
+
+OPENAI_COMPATIBLE_URLS = {
+    "pollinations": POLLINATIONS_OPENAI_URL,
+    "groq": "https://api.groq.com/openai/v1/chat/completions",
+    "siliconflow": "https://api.siliconflow.cn/v1/chat/completions",
+    "gemini": "https://generativelanguage.googleapis.com/v1beta/openai/chat/completions",
+    "openrouter": "https://openrouter.ai/api/v1/chat/completions",
+}
+
+
+API_KEY_ENVS = {
+    "groq": "GROQ_API_KEY",
+    "siliconflow": "SILICONFLOW_API_KEY",
+    "gemini": "GEMINI_API_KEY",
+    "openrouter": "OPENROUTER_API_KEY",
+}
+
+
+class LLMError(RuntimeError):
+    pass
+
+
+def chat_completion(
+    messages: list[dict[str, Any]],
+    tools: list[dict[str, Any]] | None = None,
+    temperature: float = 0.2,
+    timeout: int = 45,
+) -> dict[str, Any]:
+    config = get_llm_config()
+    payload: dict[str, Any] = {
+        "model": config.model,
+        "messages": messages,
+        "temperature": temperature,
+        "stream": False,
+    }
+    if tools:
+        payload["tools"] = tools
+        payload["tool_choice"] = "auto"
+
+    headers = {"Content-Type": "application/json"}
+    if config.api_key:
+        headers["Authorization"] = f"Bearer {config.api_key}"
+    if config.provider == "openrouter":
+        headers["HTTP-Referer"] = os.getenv("OPENROUTER_SITE_URL", "http://localhost:7860")
+        headers["X-Title"] = os.getenv("OPENROUTER_APP_NAME", "RouteOpt Agent")
+
+    session = requests.Session()
+    session.trust_env = not config.ignore_proxy
+
+    try:
+        response = session.post(config.url, json=payload, headers=headers, timeout=timeout)
+        response.raise_for_status()
+        data = response.json()
+    except Exception as exc:
+        raise LLMError(f"LLM 请求失败（provider={config.provider}, model={config.model}）：{exc}") from exc
+
+    try:
+        return data["choices"][0]["message"]
+    except Exception as exc:
+        raise LLMError(f"LLM 返回格式异常：{json.dumps(data, ensure_ascii=False)[:300]}") from exc
+
+
+def extract_tool_arguments(message: dict[str, Any], tool_name: str) -> dict[str, Any] | None:
+    for tool_call in message.get("tool_calls") or []:
+        function = tool_call.get("function") or {}
+        if function.get("name") != tool_name:
+            continue
+        raw_arguments = function.get("arguments") or "{}"
+        if isinstance(raw_arguments, dict):
+            return raw_arguments
+        try:
+            return json.loads(raw_arguments)
+        except json.JSONDecodeError:
+            return None
+    return None
+
+
+def get_llm_config() -> LLMConfig:
+    provider = normalize_provider(os.getenv("LLM_PROVIDER", "auto"))
+    if provider == "auto":
+        provider = first_configured_provider() or "pollinations"
+
+    if provider == "custom":
+        url = required_env("OPENAI_COMPATIBLE_BASE_URL")
+        api_key = os.getenv("LLM_API_KEY") or os.getenv("OPENAI_API_KEY")
+        model = os.getenv("LLM_MODEL") or os.getenv("OPENAI_COMPATIBLE_MODEL") or required_env("LLM_MODEL")
+    else:
+        url = os.getenv(f"{provider.upper()}_BASE_URL") or OPENAI_COMPATIBLE_URLS[provider]
+        api_key_env = API_KEY_ENVS.get(provider)
+        api_key = os.getenv(api_key_env) if api_key_env else None
+        provider_model_env = os.getenv(f"{provider.upper()}_MODEL")
+        legacy_pollinations_model = os.getenv("POLLINATIONS_MODEL") if provider == "pollinations" else None
+        model = (
+            os.getenv("LLM_MODEL")
+            or provider_model_env
+            or legacy_pollinations_model
+            or DEFAULT_MODELS[provider]
+        )
+
+    ignore_proxy = os.getenv("LLM_IGNORE_PROXY", "1").strip().lower() not in {"0", "false", "no"}
+    return LLMConfig(provider=provider, url=url, model=model, api_key=api_key, ignore_proxy=ignore_proxy)
+
+
+def normalize_provider(provider: str) -> str:
+    provider = provider.strip().lower()
+    aliases = {
+        "pollinations.ai": "pollinations",
+        "pollination": "pollinations",
+        "silicon": "siliconflow",
+        "sf": "siliconflow",
+        "google": "gemini",
+        "google-gemini": "gemini",
+        "openai-compatible": "custom",
+    }
+    provider = aliases.get(provider, provider)
+    allowed = {"auto", "pollinations", "groq", "siliconflow", "gemini", "openrouter", "custom"}
+    if provider not in allowed:
+        raise LLMError(f"未知 LLM_PROVIDER={provider}，可选：{', '.join(sorted(allowed))}")
+    return provider
+
+
+def first_configured_provider() -> str | None:
+    for provider, env_name in API_KEY_ENVS.items():
+        if os.getenv(env_name):
+            return provider
+    if os.getenv("OPENAI_COMPATIBLE_BASE_URL"):
+        return "custom"
+    return None
+
+
+def required_env(name: str) -> str:
+    value = os.getenv(name)
+    if not value:
+        raise LLMError(f"缺少环境变量 {name}")
+    return value
+
+
+def extract_json_from_text(text: str) -> dict[str, Any] | None:
+    text = text.strip()
+    if not text:
+        return None
+    if text.startswith("```"):
+        text = text.strip("`")
+        text = text.removeprefix("json").strip()
+    try:
+        return json.loads(text)
+    except json.JSONDecodeError:
+        pass
+
+    start = text.find("{")
+    end = text.rfind("}")
+    if start >= 0 and end > start:
+        try:
+            return json.loads(text[start : end + 1])
+        except json.JSONDecodeError:
+            return None
+    return None

routeopt_agent/models.py

@@ -0,0 +1,69 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any, Literal
+
+
+Objective = Literal["time", "distance"]
+
+
+@dataclass
+class RouteTask:
+    raw_request: str
+    start_place: str
+    destination_places: list[str]
+    objective: Objective = "time"
+    return_to_start: bool = True
+    fixed_end_place: str | None = None
+    constraints: list[str] = field(default_factory=list)
+
+
+@dataclass
+class GeoPoint:
+    name: str
+    query: str
+    lat: float
+    lon: float
+    display_name: str
+    source: str
+
+
+@dataclass
+class RouteMatrix:
+    points: list[GeoPoint]
+    durations: list[list[float]]
+    distances: list[list[float]]
+    source: str
+
+
+@dataclass
+class RouteSolution:
+    route_indices: list[int]
+    route_names: list[str]
+    total_duration_seconds: float
+    total_distance_meters: float
+    objective: Objective
+    algorithm: str
+    leg_rows: list[list[Any]]
+
+
+@dataclass
+class ToolEvent:
+    step: int
+    tool: str
+    arguments: dict[str, Any]
+    status: str
+    result: str
+
+
+@dataclass
+class AgentResult:
+    task: RouteTask
+    points: list[GeoPoint]
+    matrix: RouteMatrix
+    solution: RouteSolution
+    summary_markdown: str
+    trace: list[ToolEvent]
+    pdf_path: str
+    route_svg: str
+    warnings: list[str] = field(default_factory=list)

routeopt_agent/parsing.py

@@ -0,0 +1,130 @@
+from __future__ import annotations
+
+import re
+
+from .models import Objective, RouteTask
+
+
+def normalize_place_lines(text: str) -> list[str]:
+    places: list[str] = []
+    for line in text.replace("；", "\n").replace(";", "\n").splitlines():
+        item = line.strip(" \t\r\n-•、,，")
+        if item:
+            places.append(item)
+    return dedupe_preserve_order(places)
+
+
+def dedupe_preserve_order(items: list[str]) -> list[str]:
+    seen: set[str] = set()
+    result: list[str] = []
+    for item in items:
+        key = item.strip().lower()
+        if key and key not in seen:
+            seen.add(key)
+            result.append(item.strip())
+    return result
+
+
+def infer_objective(raw_text: str, objective_hint: str | None) -> Objective:
+    hint = (objective_hint or "").strip().lower()
+    text = raw_text.lower()
+    if hint in {"distance", "最短距离", "距离最短"}:
+        return "distance"
+    if hint in {"time", "最短时间", "最快路线"}:
+        return "time"
+    if any(word in text for word in ["距离", "路程", "公里", "少走路", "最短路"]):
+        return "distance"
+    return "time"
+
+
+def infer_return_to_start(raw_text: str, return_hint: bool | None) -> bool:
+    if return_hint is not None:
+        return bool(return_hint)
+    return any(word in raw_text for word in ["返回", "回到", "回起点", "最后回", "闭环"])
+
+
+def heuristic_extract_task(
+    raw_text: str,
+    start_hint: str = "",
+    destinations_hint: str = "",
+    objective_hint: str = "",
+    return_to_start_hint: bool | None = None,
+    fixed_end_hint: str = "",
+) -> RouteTask:
+    raw_text = (raw_text or "").strip()
+    start = start_hint.strip() or extract_start(raw_text)
+    destinations = normalize_place_lines(destinations_hint)
+    if not destinations:
+        destinations = extract_destinations(raw_text)
+
+    fixed_end = fixed_end_hint.strip() or extract_fixed_end(raw_text)
+    objective = infer_objective(raw_text, objective_hint)
+    return_to_start = infer_return_to_start(raw_text, return_to_start_hint)
+    constraints = extract_constraints(raw_text)
+
+    return RouteTask(
+        raw_request=raw_text,
+        start_place=start,
+        destination_places=destinations,
+        objective=objective,
+        return_to_start=return_to_start,
+        fixed_end_place=fixed_end or None,
+        constraints=constraints,
+    )
+
+
+def extract_start(text: str) -> str:
+    patterns = [
+        r"从(?P<place>.+?)(?:出发|开始)",
+        r"起点(?:是|为|:|：)?(?P<place>[^，。,；;\n]+)",
+        r"我在(?P<place>[^，。,；;\n]+)",
+    ]
+    for pattern in patterns:
+        match = re.search(pattern, text)
+        if match:
+            return cleanup_place(match.group("place"))
+    return ""
+
+
+def extract_destinations(text: str) -> list[str]:
+    patterns = [
+        r"(?:去|逛完|访问|经过|打卡|游览)(?P<places>.+?)(?:，?最后|，?返回|，?回到|，?尽量|，?要求|。|$)",
+        r"目的地(?:是|为|:|：)?(?P<places>.+?)(?:。|$)",
+    ]
+    for pattern in patterns:
+        match = re.search(pattern, text)
+        if match:
+            return split_places(match.group("places"))
+    return []
+
+
+def extract_fixed_end(text: str) -> str:
+    if any(word in text for word in ["返回", "回到", "回起点", "最后回"]):
+        return ""
+    match = re.search(r"最后(?:到|去|抵达)(?P<place>[^，。,；;\n]+)", text)
+    if match:
+        return cleanup_place(match.group("place"))
+    return ""
+
+
+def split_places(text: str) -> list[str]:
+    text = re.sub(r"(?:并且|然后|再|以及)", "、", text)
+    text = re.sub(r"(?:和|与)", "、", text)
+    pieces = re.split(r"[、,，;；\n]+", text)
+    return dedupe_preserve_order([cleanup_place(piece) for piece in pieces if cleanup_place(piece)])
+
+
+def cleanup_place(place: str) -> str:
+    place = place.strip()
+    place = re.sub(r"^(?:先|再|然后|顺便|还要|想|希望|需要)", "", place)
+    place = re.sub(r"(?:这些地方|几个地方|等地|附近)$", "", place)
+    return place.strip(" \t\r\n。.!！?？")
+
+
+def extract_constraints(text: str) -> list[str]:
+    constraints: list[str] = []
+    if any(word in text for word in ["必须先", "先去", "优先去"]):
+        constraints.append("用户表达了先后顺序偏好，当前版本会在报告中提示但不强制锁定顺序。")
+    if any(word in text for word in ["一天内", "半天", "预算", "不超过"]):
+        constraints.append("用户表达了时间/预算类软约束，当前版本会用于解释，不改变 TSP 最优目标。")
+    return constraints

routeopt_agent/report.py

@@ -0,0 +1,216 @@
+from __future__ import annotations
+
+import html
+from datetime import datetime
+from pathlib import Path
+
+from reportlab.lib import colors
+from reportlab.lib.enums import TA_LEFT
+from reportlab.lib.pagesizes import A4
+from reportlab.lib.styles import ParagraphStyle, getSampleStyleSheet
+from reportlab.lib.units import cm
+from reportlab.pdfbase import pdfmetrics
+from reportlab.pdfbase.cidfonts import UnicodeCIDFont
+from reportlab.platypus import (
+    PageBreak,
+    Paragraph,
+    SimpleDocTemplate,
+    Spacer,
+    Table,
+    TableStyle,
+)
+
+from .models import AgentResult, GeoPoint, RouteSolution, RouteTask, ToolEvent
+from .solver import format_km, format_minutes
+
+
+REPORT_DIR = Path("outputs")
+
+
+def generate_pdf_report(
+    task: RouteTask,
+    points: list[GeoPoint],
+    solution: RouteSolution,
+    trace: list[ToolEvent],
+    summary_markdown: str,
+) -> str:
+    REPORT_DIR.mkdir(parents=True, exist_ok=True)
+    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+    path = REPORT_DIR / f"routeopt_report_{timestamp}.pdf"
+
+    pdfmetrics.registerFont(UnicodeCIDFont("STSong-Light"))
+    doc = SimpleDocTemplate(
+        str(path),
+        pagesize=A4,
+        rightMargin=1.8 * cm,
+        leftMargin=1.8 * cm,
+        topMargin=1.6 * cm,
+        bottomMargin=1.6 * cm,
+        title="RouteOpt Agent 求解报告",
+    )
+
+    styles = make_styles()
+    story: list = []
+
+    story.append(Paragraph("RouteOpt Agent 求解报告", styles["TitleCJK"]))
+    story.append(Paragraph(f"生成时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}", styles["BodyCJK"]))
+    story.append(Spacer(1, 0.3 * cm))
+
+    story.append(Paragraph("1. 问题定义", styles["HeadingCJK"]))
+    story.extend(task_paragraphs(task, styles))
+    story.append(Spacer(1, 0.2 * cm))
+
+    story.append(Paragraph("2. 地点解析结果", styles["HeadingCJK"]))
+    story.append(make_points_table(points))
+    story.append(Spacer(1, 0.2 * cm))
+
+    story.append(Paragraph("3. 优化结果", styles["HeadingCJK"]))
+    story.extend(solution_paragraphs(solution, styles))
+    story.append(make_leg_table(solution))
+    story.append(Spacer(1, 0.2 * cm))
+
+    story.append(Paragraph("4. Agent 工具调用轨迹", styles["HeadingCJK"]))
+    story.append(make_trace_table(trace))
+
+    story.append(PageBreak())
+    story.append(Paragraph("5. 结果解释", styles["HeadingCJK"]))
+    for paragraph in markdown_to_plain_paragraphs(summary_markdown):
+        story.append(Paragraph(escape_for_pdf(paragraph), styles["BodyCJK"]))
+        story.append(Spacer(1, 0.12 * cm))
+
+    story.append(Paragraph("6. 局限性说明", styles["HeadingCJK"]))
+    limitations = [
+        "公开 API 可能存在限速、网络延迟或临时不可用；系统内置了上海示例地点和直线距离兜底策略。",
+        "OSRM 公共服务适合课程演示和小规模样例，不建议用于商业高并发生产环境。",
+        "本系统把大模型用于自然语言理解、工具调用规划和解释生成，把路线最优化交给确定性算法完成。",
+    ]
+    for item in limitations:
+        story.append(Paragraph("• " + escape_for_pdf(item), styles["BodyCJK"]))
+
+    doc.build(story)
+    return str(path)
+
+
+def make_styles() -> dict[str, ParagraphStyle]:
+    base = getSampleStyleSheet()
+    return {
+        "TitleCJK": ParagraphStyle(
+            "TitleCJK",
+            parent=base["Title"],
+            fontName="STSong-Light",
+            fontSize=20,
+            leading=26,
+            alignment=TA_LEFT,
+        ),
+        "HeadingCJK": ParagraphStyle(
+            "HeadingCJK",
+            parent=base["Heading2"],
+            fontName="STSong-Light",
+            fontSize=14,
+            leading=20,
+            spaceBefore=8,
+            spaceAfter=6,
+        ),
+        "BodyCJK": ParagraphStyle(
+            "BodyCJK",
+            parent=base["BodyText"],
+            fontName="STSong-Light",
+            fontSize=10.5,
+            leading=16,
+            wordWrap="CJK",
+        ),
+    }
+
+
+def task_paragraphs(task: RouteTask, styles: dict[str, ParagraphStyle]) -> list[Paragraph]:
+    destination_text = "、".join(task.destination_places)
+    objective_text = "最短时间" if task.objective == "time" else "最短距离"
+    return_text = "需要回到起点" if task.return_to_start else "不需要回到起点"
+    fixed_end = task.fixed_end_place or "无"
+    raw_request = task.raw_request or "用户通过结构化表单输入任务。"
+    return [
+        Paragraph(f"原始需求：{escape_for_pdf(raw_request)}", styles["BodyCJK"]),
+        Paragraph(f"起点：{escape_for_pdf(task.start_place)}", styles["BodyCJK"]),
+        Paragraph(f"目的地：{escape_for_pdf(destination_text)}", styles["BodyCJK"]),
+        Paragraph(f"优化目标：{objective_text}；回程设置：{return_text}；固定终点：{escape_for_pdf(fixed_end)}", styles["BodyCJK"]),
+    ]
+
+
+def solution_paragraphs(solution: RouteSolution, styles: dict[str, ParagraphStyle]) -> list[Paragraph]:
+    return [
+        Paragraph(f"访问顺序：{escape_for_pdf(' → '.join(solution.route_names))}", styles["BodyCJK"]),
+        Paragraph(
+            f"总距离：{format_km(solution.total_distance_meters)}；预计驾驶时间：{format_minutes(solution.total_duration_seconds)}",
+            styles["BodyCJK"],
+        ),
+        Paragraph(f"求解算法：{solution.algorithm}", styles["BodyCJK"]),
+    ]
+
+
+def make_points_table(points: list[GeoPoint]) -> Table:
+    data = [["序号", "地点", "纬度", "经度", "数据源"]]
+    for idx, point in enumerate(points):
+        data.append([idx, point.name, f"{point.lat:.6f}", f"{point.lon:.6f}", point.source])
+    table = Table(data, colWidths=[1.0 * cm, 4.3 * cm, 3.0 * cm, 3.0 * cm, 4.1 * cm])
+    apply_table_style(table)
+    return table
+
+
+def make_leg_table(solution: RouteSolution) -> Table:
+    data = [["段", "从", "到", "距离", "时间"]] + solution.leg_rows
+    table = Table(data, colWidths=[1.0 * cm, 4.3 * cm, 4.3 * cm, 2.5 * cm, 2.5 * cm])
+    apply_table_style(table)
+    return table
+
+
+def make_trace_table(trace: list[ToolEvent]) -> Table:
+    data = [["步骤", "工具", "状态", "结果摘要"]]
+    for event in trace:
+        data.append([event.step, event.tool, event.status, event.result[:120]])
+    table = Table(data, colWidths=[1.0 * cm, 4.0 * cm, 2.0 * cm, 8.6 * cm], repeatRows=1)
+    apply_table_style(table)
+    return table
+
+
+def apply_table_style(table: Table) -> None:
+    table.setStyle(
+        TableStyle(
+            [
+                ("FONTNAME", (0, 0), (-1, -1), "STSong-Light"),
+                ("FONTSIZE", (0, 0), (-1, -1), 8.5),
+                ("BACKGROUND", (0, 0), (-1, 0), colors.HexColor("#eef2f7")),
+                ("TEXTCOLOR", (0, 0), (-1, 0), colors.HexColor("#111827")),
+                ("GRID", (0, 0), (-1, -1), 0.25, colors.HexColor("#d1d5db")),
+                ("VALIGN", (0, 0), (-1, -1), "TOP"),
+                ("ROWBACKGROUNDS", (0, 1), (-1, -1), [colors.white, colors.HexColor("#f8fafc")]),
+            ]
+        )
+    )
+
+
+def markdown_to_plain_paragraphs(markdown: str) -> list[str]:
+    lines = []
+    for line in markdown.splitlines():
+        clean = line.strip()
+        if not clean:
+            continue
+        clean = clean.replace("**", "")
+        clean = clean.lstrip("#").strip()
+        clean = clean.lstrip("-").strip()
+        lines.append(clean)
+    return lines or ["Agent 已完成路线优化并生成结构化结果。"]
+
+
+def escape_for_pdf(text: str) -> str:
+    return html.escape(str(text)).replace("\n", "<br/>")
+
+
+def result_to_markdown(result: AgentResult) -> str:
+    solution = result.solution
+    return (
+        f"**访问顺序**：{' → '.join(solution.route_names)}\n\n"
+        f"**总距离**：{format_km(solution.total_distance_meters)}  \n"
+        f"**预计时间**：{format_minutes(solution.total_duration_seconds)}\n\n"
+        f"**算法**：{solution.algorithm}\n\n"
+        f"{result.summary_markdown}"
+    )

routeopt_agent/solver.py

@@ -0,0 +1,189 @@
+from __future__ import annotations
+
+from typing import Any
+
+from .models import Objective, RouteMatrix, RouteSolution
+
+
+class SolverError(RuntimeError):
+    pass
+
+
+def solve_route(
+    matrix: RouteMatrix,
+    objective: Objective,
+    return_to_start: bool,
+    fixed_end_place: str | None = None,
+) -> RouteSolution:
+    if len(matrix.points) < 2:
+        raise SolverError("至少需要起点和 1 个目的地。")
+
+    cost_matrix = matrix.durations if objective == "time" else matrix.distances
+    fixed_end_idx = find_fixed_end_index(matrix, fixed_end_place)
+    route_indices, objective_cost = held_karp(
+        cost_matrix=cost_matrix,
+        return_to_start=return_to_start,
+        fixed_end_idx=fixed_end_idx,
+    )
+
+    total_duration = route_cost(matrix.durations, route_indices)
+    total_distance = route_cost(matrix.distances, route_indices)
+    route_names = [matrix.points[idx].name for idx in route_indices]
+    leg_rows = build_leg_rows(matrix, route_indices)
+
+    if len(matrix.points) <= 10:
+        algorithm = "Held-Karp dynamic programming exact TSP solver"
+    else:
+        algorithm = "Nearest-neighbor + 2-opt heuristic"
+
+    if objective_cost < 0:
+        raise SolverError("求解失败：目标函数成本异常。")
+
+    return RouteSolution(
+        route_indices=route_indices,
+        route_names=route_names,
+        total_duration_seconds=total_duration,
+        total_distance_meters=total_distance,
+        objective=objective,
+        algorithm=algorithm,
+        leg_rows=leg_rows,
+    )
+
+
+def held_karp(
+    cost_matrix: list[list[float]],
+    return_to_start: bool,
+    fixed_end_idx: int | None,
+) -> tuple[list[int], float]:
+    node_count = len(cost_matrix)
+    if node_count == 1:
+        return [0], 0.0
+
+    if return_to_start:
+        fixed_end_idx = None
+
+    all_destinations = list(range(1, node_count))
+    visit_nodes = [idx for idx in all_destinations if idx != fixed_end_idx]
+
+    if not visit_nodes:
+        if fixed_end_idx is None:
+            return [0, 0] if return_to_start else [0], 0.0
+        return [0, fixed_end_idx], safe_cost(cost_matrix, 0, fixed_end_idx)
+
+    bit_for_node = {node: 1 << pos for pos, node in enumerate(visit_nodes)}
+    dp: dict[tuple[int, int], tuple[float, int | None]] = {}
+
+    for node in visit_nodes:
+        mask = bit_for_node[node]
+        dp[(mask, node)] = (safe_cost(cost_matrix, 0, node), None)
+
+    full_mask = (1 << len(visit_nodes)) - 1
+    for mask in range(1, full_mask + 1):
+        for last in visit_nodes:
+            if not mask & bit_for_node[last]:
+                continue
+            current = dp.get((mask, last))
+            if current is None:
+                continue
+            current_cost, _ = current
+            for nxt in visit_nodes:
+                nxt_bit = bit_for_node[nxt]
+                if mask & nxt_bit:
+                    continue
+                next_mask = mask | nxt_bit
+                next_cost = current_cost + safe_cost(cost_matrix, last, nxt)
+                old = dp.get((next_mask, nxt))
+                if old is None or next_cost < old[0]:
+                    dp[(next_mask, nxt)] = (next_cost, last)
+
+    best_last: int | None = None
+    best_cost = float("inf")
+    for last in visit_nodes:
+        state = dp.get((full_mask, last))
+        if state is None:
+            continue
+        total = state[0]
+        if fixed_end_idx is not None:
+            total += safe_cost(cost_matrix, last, fixed_end_idx)
+        elif return_to_start:
+            total += safe_cost(cost_matrix, last, 0)
+        if total < best_cost:
+            best_cost = total
+            best_last = last
+
+    if best_last is None:
+        raise SolverError("Held-Karp 求解失败。")
+
+    path = reconstruct_path(dp, bit_for_node, full_mask, best_last)
+    route = [0] + path
+    if fixed_end_idx is not None:
+        route.append(fixed_end_idx)
+    elif return_to_start:
+        route.append(0)
+    return route, best_cost
+
+
+def reconstruct_path(
+    dp: dict[tuple[int, int], tuple[float, int | None]],
+    bit_for_node: dict[int, int],
+    mask: int,
+    last: int,
+) -> list[int]:
+    reversed_path: list[int] = []
+    current_last: int | None = last
+    current_mask = mask
+    while current_last is not None:
+        reversed_path.append(current_last)
+        _, prev = dp[(current_mask, current_last)]
+        current_mask ^= bit_for_node[current_last]
+        current_last = prev
+    return list(reversed(reversed_path))
+
+
+def find_fixed_end_index(matrix: RouteMatrix, fixed_end_place: str | None) -> int | None:
+    if not fixed_end_place:
+        return None
+    normalized = fixed_end_place.strip().lower()
+    for idx, point in enumerate(matrix.points):
+        if idx == 0:
+            continue
+        if normalized in point.name.lower() or point.name.lower() in normalized:
+            return idx
+    return None
+
+
+def route_cost(cost_matrix: list[list[float]], route_indices: list[int]) -> float:
+    total = 0.0
+    for origin, dest in zip(route_indices, route_indices[1:]):
+        total += safe_cost(cost_matrix, origin, dest)
+    return total
+
+
+def safe_cost(cost_matrix: list[list[float]], origin: int, dest: int) -> float:
+    value = cost_matrix[origin][dest]
+    if value is None:
+        return float("inf")
+    return float(value)
+
+
+def build_leg_rows(matrix: RouteMatrix, route_indices: list[int]) -> list[list[Any]]:
+    rows: list[list[Any]] = []
+    for step, (origin, dest) in enumerate(zip(route_indices, route_indices[1:]), start=1):
+        rows.append(
+            [
+                step,
+                matrix.points[origin].name,
+                matrix.points[dest].name,
+                format_km(matrix.distances[origin][dest]),
+                format_minutes(matrix.durations[origin][dest]),
+            ]
+        )
+    return rows
+
+
+def format_km(meters: float) -> str:
+    return f"{meters / 1000:.2f} km"
+
+
+def format_minutes(seconds: float) -> str:
+    return f"{seconds / 60:.1f} min"

routeopt_agent/viz.py

@@ -0,0 +1,64 @@
+from __future__ import annotations
+
+import html
+
+from .models import GeoPoint
+
+
+def build_route_svg(points: list[GeoPoint], route_indices: list[int]) -> str:
+    if not points or not route_indices:
+        return ""
+
+    width = 760
+    height = 420
+    padding = 46
+    lats = [point.lat for point in points]
+    lons = [point.lon for point in points]
+    min_lat, max_lat = min(lats), max(lats)
+    min_lon, max_lon = min(lons), max(lons)
+    lat_span = max(max_lat - min_lat, 0.001)
+    lon_span = max(max_lon - min_lon, 0.001)
+
+    def project(point: GeoPoint) -> tuple[float, float]:
+        x = padding + (point.lon - min_lon) / lon_span * (width - 2 * padding)
+        y = height - padding - (point.lat - min_lat) / lat_span * (height - 2 * padding)
+        return x, y
+
+    projected = [project(point) for point in points]
+    route_points = [projected[idx] for idx in route_indices]
+    polyline = " ".join(f"{x:.1f},{y:.1f}" for x, y in route_points)
+
+    marker_svg = []
+    first_visit_number: dict[int, int] = {}
+    for order, idx in enumerate(route_indices, start=1):
+        first_visit_number.setdefault(idx, order)
+
+    for idx, point in enumerate(points):
+        x, y = projected[idx]
+        order = first_visit_number.get(idx, idx + 1)
+        is_start = idx == 0
+        fill = "#0f766e" if is_start else "#2563eb"
+        marker_svg.append(
+            f"""
+            <g>
+              <circle cx="{x:.1f}" cy="{y:.1f}" r="15" fill="{fill}" stroke="#ffffff" stroke-width="3"/>
+              <text x="{x:.1f}" y="{y + 4:.1f}" text-anchor="middle" font-size="12" font-weight="700" fill="#ffffff">{order}</text>
+              <text x="{x + 19:.1f}" y="{y - 16:.1f}" font-size="12" font-weight="600" fill="#111827">{html.escape(point.name)}</text>
+            </g>
+            """
+        )
+
+    return f"""
+    <div style="width: 100%;">
+      <svg viewBox="0 0 {width} {height}" role="img" aria-label="Route diagram" style="width:100%;height:auto;border:1px solid #d9e2ec;border-radius:8px;">
+        <defs>
+          <marker id="arrow" markerWidth="10" markerHeight="10" refX="7" refY="3" orient="auto" markerUnits="strokeWidth">
+            <path d="M0,0 L0,6 L8,3 z" fill="#334155" />
+          </marker>
+        </defs>
+        <rect x="0" y="0" width="{width}" height="{height}" rx="8" fill="#f8fafc"/>
+        <path d="M {polyline.replace(' ', ' L ')}" fill="none" stroke="#334155" stroke-width="4" stroke-linecap="round" stroke-linejoin="round" marker-end="url(#arrow)"/>
+        {''.join(marker_svg)}
+      </svg>
+    </div>
+    """

run.sh

@@ -0,0 +1,44 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+ENV_NAME="${ROUTEOPT_ENV:-routeopt-agent}"
+
+ensure_deps() {
+  if conda run -n "$ENV_NAME" python -c "import gradio, requests, reportlab" >/dev/null 2>&1; then
+    echo "依赖已安装"
+  else
+    echo "安装 Python 依赖"
+    conda run -n "$ENV_NAME" python -m pip install -r requirements.txt
+  fi
+}
+
+if ! command -v conda >/dev/null 2>&1; then
+  echo "未找到 conda。请先安装 Miniconda 或 Anaconda，然后重新运行 ./run.sh"
+  exit 1
+fi
+
+if [ -f ".env" ]; then
+  echo "加载 .env 配置"
+  set -a
+  # shellcheck disable=SC1091
+  source ".env"
+  set +a
+fi
+
+CONDA_BASE="$(conda info --base)"
+source "$CONDA_BASE/etc/profile.d/conda.sh"
+
+if ! conda env list | awk '{print $1}' | grep -qx "$ENV_NAME"; then
+  echo "首次运行：创建 conda 环境 $ENV_NAME"
+  if ! conda env create -f environment.yml; then
+    echo "标准环境创建失败，尝试克隆 base 环境后用 pip 安装依赖"
+    conda create -n "$ENV_NAME" --clone base -y
+  fi
+else
+  echo "检查 Python 依赖"
+fi
+
+ensure_deps
+
+echo "启动 RouteOpt Agent"
+conda run -n "$ENV_NAME" python app.py