app.py

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)