import os, json, logging
from openai import OpenAI
from baidu_map_search_mcp import (
get_ip,
get_location_ip,
get_location_text,
get_places,
get_place_driving_distance,
get_place_riding_distance,
get_place_walking_distance,
get_place_transit_distance
)
logging.basicConfig(level=logging.WARNING)
logging.getLogger("openai").setLevel(logging.ERROR)
logging.getLogger("httpx").setLevel(logging.ERROR)
logging.getLogger("httpcore").setLevel(logging.ERROR)
api_key = os.getenv("DEEPSEEK_API_KEY")
client = OpenAI(
api_key=api_key,
base_url="https://api.deepseek.com/v1",
)
SYSTEM_PROMPT = """
你是一个专业的地理信息助手，专注于地点检索和路线规划服务。你可以通过调用以下工具获取所需信息，并基于此为用户提供详细的出行建议：
- get_ip: 获取用户公网IP地址，用于初步定位
- get_location_ip: 根据IP地址获取精确地理位置信息（经纬度、地区等）
- get_location_text: 将地址文本（如"北京市海淀区"）转换为经纬度坐标
- get_places: 搜索指定位置（经纬度）附近的地点，支持设置搜索半径
- get_place_driving_distance: 计算两点间驾驶距离和预计时间（含路线信息）
- get_place_riding_distance: 计算两点间骑行距离和预计时间
- get_place_walking_distance: 计算两点间步行距离和预计时间
- get_place_transit_distance: 计算两点间公共交通距离和预计时间（含换乘信息）
### 工作流程：
1. 当用户询问地点相关问题时，先明确位置信息（可通过IP自动获取或提示用户输入地址）
2. 如需路线规划，需确认起点和终点（支持地址文本或经纬度）
3. 根据距离、时间、用户可能的需求，自动调用多种交通方式工具进行对比
4. 优先选择最优方案，并提供多种交通方式的对比信息
### 输出要求：
必须以Markdown格式整理结果，包含：
- 清晰的标题（如"地点检索结果"、"路线规划建议"）
- 分点列出的关键信息（位置、距离、时间等）
- 交通方式对比表格（如有多种选项）
- 简要总结和推荐理由
### 工具调用规则：
- 严格按照指定格式调用工具，确保参数正确（尤其是经纬度格式为"纬度,经度"）
- 若工具返回结果不完整或错误，可重新调用或提示用户补充信息
- 无需调用工具时直接回答，需调用工具时先执行工具调用再整理结果
"""
with open("tools_definition.json", "r", encoding="utf-8") as f:
TOOLS_DEFINITION = json.load(f)
available_tools = {
"get_ip": get_ip,
"get_location_ip": get_location_ip,
"get_location_text": get_location_text,
"get_places": get_places,
"get_place_driving_distance": get_place_driving_distance,
"get_place_riding_distance": get_place_riding_distance,
"get_place_walking_distance": get_place_walking_distance,
"get_place_transit_distance": get_place_transit_distance
}
def start_client(user_input: str, history_messages: list = None):
print("agent开始运行...")
if history_messages is None:
messages = [{"role": "system", "content": SYSTEM_PROMPT}]
else:
messages = history_messages.copy()
messages.append({"role": "user", "content": user_input})
iteration_count = 0
final_response = None
while iteration_count < 100:
iteration_count += 1
print(f"-- Iteration {iteration_count} --")
try:
response = client.chat.completions.create(
model = "deepseek-chat",
messages = messages,
tools = TOOLS_DEFINITION,
tool_choice = "auto"
)
response_message = response.choices[0].message
if response_message.tool_calls:
print("Agent: 决定调用工具...")
messages.append(response_message)
tool_outputs = []
for tool_call in response_message.tool_calls:
function_name = tool_call.function.name
function_args = json.loads(tool_call.function.arguments) if tool_call.function.arguments else {}
print(f"Agent Action: 调用工具 '{function_name}'，参数：{function_args}")
if function_name in available_tools:
tool_function = available_tools[function_name]
import asyncio
tool_result = asyncio.run(tool_function(**function_args))
tool_outputs.append({
"tool_call_id": tool_call.id,
"role": "tool",
"content": str(tool_result)
})
else:
error_message = f"错误：未知的工具 '{function_name}'"
tool_outputs.append({
"tool_call_id": tool_call.id,
"role": "tool",
"content": error_message
})
messages.extend(tool_outputs)
elif response_message.content:
final_response = response_message.content
messages.append({"role": "assistant", "content": final_response})
return final_response, messages
else:
print("Agent: 未收到内容或工具调用，结束对话。")
break
except Exception as e:
print(f"调用API错误: {e}")
break
def chat_loop():
history = None
while True:
user_input = input("请输入您的请求内容(输入'退出'结束对话): ")
if user_input.lower() == '退出':
print("对话结束")
break
response, history = start_client(user_input, history)
print(response)
print("\n====================================================================\n")
if __name__ == "__main__":
chat_loop()
