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" "
技术调试信息\n\n" f"```text\n{traceback.format_exc()[-1800:]}\n```\n
" ) 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)