Add coordinate tool flow and tighten tool usage
Introduce ENU offset tool support with controlled tool-call handling, update prompts to gate coordinate tooling, and enable jinja tools in startup.
This commit is contained in:
@@ -233,5 +233,13 @@
|
||||
|
||||
**仅当指令涉及前往“具体地点”（如广场、大门）的偏移位置时，才需计算绝对坐标并使用`fly_to_waypoint`：**
|
||||
|
||||
当指令包含“具体地点 + 方向 + 距离”的偏移（如“广场西边200米”）时，**必须**先调用工具`calc_offset_enu`计算绝对坐标，再使用`fly_to_waypoint`。工具参数：
|
||||
- `base`: 参考地点的ENU坐标（含x/y/z）
|
||||
- `direction`: east/west/north/south/up/down
|
||||
- `distance`: 偏移距离（米）
|
||||
|
||||
当指令只有“方向 + 距离”且**没有具体地点名词**时，**禁止**调用`calc_offset_enu`，必须使用`move_direction`。
|
||||
当指令描述“附近/边上/区域内”等模糊位置且**无方向+距离**时，视为到该地点本身，不做偏移计算。
|
||||
|
||||
## 八、输出要求
|
||||
仅输出1个严格符合上述所有规则的JSON对象。
|
||||
|
||||
@@ -10,6 +10,7 @@ from openai import OpenAIError
|
||||
import jsonschema
|
||||
import requests
|
||||
import platform # 新增：用于选择合适的中文字体
|
||||
from .tools.coordinate_tools import calc_offset_enu
|
||||
|
||||
# --- 自定义远程嵌入函数 (与ingest.py中定义一致) ---
|
||||
from chromadb.api.types import Documents, EmbeddingFunction, Embeddings, Embeddable
|
||||
@@ -644,6 +645,99 @@ class PyTreeGenerator:
|
||||
logging.error(f"提示词文件未找到 -> {file_name}")
|
||||
return ""
|
||||
|
||||
def _get_tool_definitions(self) -> list[dict]:
|
||||
return [
|
||||
{
|
||||
"type": "function",
|
||||
"function": {
|
||||
"name": "calc_offset_enu",
|
||||
"description": "根据ENU坐标系基准点、方向和距离计算偏移后的坐标。",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"base": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"x": {"type": "number"},
|
||||
"y": {"type": "number"},
|
||||
"z": {"type": "number"}
|
||||
},
|
||||
"required": ["x", "y", "z"]
|
||||
},
|
||||
"direction": {
|
||||
"type": "string",
|
||||
"enum": [
|
||||
"east", "west", "north", "south",
|
||||
"up", "down"
|
||||
]
|
||||
},
|
||||
"distance": {"type": "number", "minimum": 0}
|
||||
},
|
||||
"required": ["base", "direction", "distance"]
|
||||
}
|
||||
}
|
||||
}
|
||||
]
|
||||
|
||||
def _normalize_tool_calls(self, tool_calls: list) -> list[dict]:
|
||||
normalized = []
|
||||
for call in tool_calls:
|
||||
if hasattr(call, "function"):
|
||||
normalized.append(
|
||||
{
|
||||
"id": getattr(call, "id", None),
|
||||
"name": call.function.name,
|
||||
"arguments": call.function.arguments,
|
||||
}
|
||||
)
|
||||
elif isinstance(call, dict):
|
||||
normalized.append(
|
||||
{
|
||||
"id": call.get("id"),
|
||||
"name": (call.get("function") or {}).get("name"),
|
||||
"arguments": (call.get("function") or {}).get("arguments"),
|
||||
}
|
||||
)
|
||||
return normalized
|
||||
|
||||
def _execute_tool_calls(self, tool_calls: list) -> list[dict]:
|
||||
tool_messages = []
|
||||
normalized = self._normalize_tool_calls(tool_calls)
|
||||
for call in normalized:
|
||||
tool_name = call.get("name")
|
||||
tool_args = call.get("arguments")
|
||||
tool_id = call.get("id")
|
||||
if not tool_name:
|
||||
logging.warning("工具调用缺少名称，已忽略。")
|
||||
continue
|
||||
try:
|
||||
args = json.loads(tool_args or "{}")
|
||||
except json.JSONDecodeError:
|
||||
logging.warning(f"工具调用参数解析失败: {tool_args}")
|
||||
continue
|
||||
try:
|
||||
if tool_name == "calc_offset_enu":
|
||||
result = calc_offset_enu(**args)
|
||||
else:
|
||||
result = {"error": f"unsupported tool: {tool_name}"}
|
||||
tool_messages.append(
|
||||
{
|
||||
"role": "tool",
|
||||
"tool_call_id": tool_id,
|
||||
"content": json.dumps(result, ensure_ascii=False)
|
||||
}
|
||||
)
|
||||
except Exception as exc:
|
||||
logging.warning(f"工具调用执行失败({tool_name}): {exc}")
|
||||
tool_messages.append(
|
||||
{
|
||||
"role": "tool",
|
||||
"tool_call_id": tool_id,
|
||||
"content": json.dumps({"error": str(exc)}, ensure_ascii=False)
|
||||
}
|
||||
)
|
||||
return tool_messages
|
||||
|
||||
def _retrieve_context(self, query: str) -> Optional[str]:
|
||||
logging.info("--- 开始从向量数据库检索上下文 ---")
|
||||
try:
|
||||
@@ -661,6 +755,33 @@ class PyTreeGenerator:
|
||||
logging.error(f"从向量数据库检索时发生错误: {e}")
|
||||
return None
|
||||
|
||||
def _should_enable_tools(self, user_prompt: str, retrieved_context: Optional[str]) -> bool:
|
||||
if not user_prompt:
|
||||
return False
|
||||
|
||||
direction_patterns = [
|
||||
"东边", "西边", "南边", "北边",
|
||||
"东侧", "西侧", "南侧", "北侧",
|
||||
"往东", "往西", "往南", "往北",
|
||||
"向东", "向西", "向南", "向北",
|
||||
]
|
||||
has_direction = any(pat in user_prompt for pat in direction_patterns)
|
||||
has_distance = re.search(r"\d+(\.\d+)?\s*(米|m)", user_prompt) is not None
|
||||
if not (has_direction and has_distance):
|
||||
return False
|
||||
|
||||
if not retrieved_context:
|
||||
return False
|
||||
|
||||
place_candidates: list[str] = []
|
||||
for line in retrieved_context.splitlines():
|
||||
if "地点：" in line or "别名：" in line:
|
||||
place_candidates.extend(re.findall(r"'([^']+)'", line))
|
||||
if not place_candidates:
|
||||
return False
|
||||
|
||||
return any(place in user_prompt for place in place_candidates)
|
||||
|
||||
def _save_history(self, prompt: str, result_dict: dict):
|
||||
"""保存请求和响应历史记录"""
|
||||
import datetime
|
||||
@@ -740,37 +861,91 @@ class PyTreeGenerator:
|
||||
# System Prompt: use_prompt（不包含RAG结果）
|
||||
# User Prompt: final_user_prompt（包含原始user_prompt + RAG检索结果）
|
||||
final_prompt = f"=== System Prompt ===\n{use_prompt}\n\n=== User Prompt ===\n{final_user_prompt}"
|
||||
tool_enabled = self._should_enable_tools(user_prompt, retrieved_context)
|
||||
for attempt in range(3):
|
||||
logging.info(f"--- 第 {attempt + 1}/3 次尝试生成Pytree ---")
|
||||
try:
|
||||
# 简单/复杂分流到不同模型与提示词
|
||||
client = self.simple_llm_client if mode == "simple" else self.complex_llm_client
|
||||
model_name = self.simple_model if mode == "simple" else self.complex_model
|
||||
messages = [
|
||||
{"role": "system", "content": use_prompt},
|
||||
{"role": "user", "content": final_user_prompt}
|
||||
]
|
||||
# 始终强制JSON响应并禁用思考功能
|
||||
response_kwargs = {
|
||||
"model": model_name,
|
||||
"messages": [
|
||||
{"role": "system", "content": use_prompt},
|
||||
{"role": "user", "content": final_user_prompt}
|
||||
],
|
||||
"messages": messages,
|
||||
"temperature": 0.1 if mode == "complex" else 0.0,
|
||||
"response_format": {"type": "json_object"}, # 始终强制JSON输出，禁用思考功能
|
||||
# 禁用 Qwen3 模型的思考功能（通过 extra_body 传递）
|
||||
# 注意：如果 API 服务器不支持此参数，会忽略
|
||||
"extra_body": {"chat_template_kwargs": {"enable_thinking": False}}
|
||||
}
|
||||
if tool_enabled:
|
||||
response_kwargs["tools"] = self._get_tool_definitions()
|
||||
response_kwargs["tool_choice"] = "auto"
|
||||
# 基于模式设定最大输出token数（直接在代码中配置）
|
||||
response_kwargs["max_tokens"] = self.simple_max_tokens if mode == "simple" else self.complex_max_tokens
|
||||
response = client.chat.completions.create(**response_kwargs)
|
||||
|
||||
# 工具调用处理：执行工具并回填后，强制模型输出JSON
|
||||
for tool_round in range(3):
|
||||
try:
|
||||
first_msg = response.choices[0].message
|
||||
tool_calls = getattr(first_msg, "tool_calls", None)
|
||||
except Exception:
|
||||
first_msg = response.choices[0].get("message") if isinstance(response.choices[0], dict) else None
|
||||
tool_calls = (first_msg or {}).get("tool_calls")
|
||||
|
||||
if not tool_calls:
|
||||
break
|
||||
|
||||
logging.info("检测到工具调用，执行第 %d 轮工具回填。", tool_round + 1)
|
||||
tool_messages = self._execute_tool_calls(tool_calls)
|
||||
if not tool_messages:
|
||||
break
|
||||
|
||||
tool_calls_for_messages = []
|
||||
for call in self._normalize_tool_calls(tool_calls):
|
||||
if not call.get("id") or not call.get("name"):
|
||||
continue
|
||||
tool_calls_for_messages.append(
|
||||
{
|
||||
"id": call["id"],
|
||||
"type": "function",
|
||||
"function": {
|
||||
"name": call["name"],
|
||||
"arguments": call.get("arguments", "")
|
||||
}
|
||||
}
|
||||
)
|
||||
messages = messages + [
|
||||
{
|
||||
"role": "assistant",
|
||||
"content": getattr(first_msg, "content", "") if first_msg else "",
|
||||
"tool_calls": tool_calls_for_messages
|
||||
}
|
||||
] + tool_messages
|
||||
followup_kwargs = dict(response_kwargs)
|
||||
followup_kwargs["messages"] = messages
|
||||
followup_kwargs.pop("tools", None)
|
||||
followup_kwargs.pop("tool_choice", None)
|
||||
response = client.chat.completions.create(**followup_kwargs)
|
||||
break
|
||||
# 兼容可能存在的 reasoning_content 字段
|
||||
try:
|
||||
msg = response.choices[0].message
|
||||
msg_content = getattr(msg, "content", None)
|
||||
msg_reasoning = getattr(msg, "reasoning_content", None)
|
||||
remaining_tool_calls = getattr(msg, "tool_calls", None)
|
||||
except Exception:
|
||||
msg = response.choices[0]["message"] if isinstance(response.choices[0], dict) else None
|
||||
msg_content = (msg or {}).get("content") if isinstance(msg, dict) else None
|
||||
msg_reasoning = (msg or {}).get("reasoning_content") if isinstance(msg, dict) else None
|
||||
remaining_tool_calls = (msg or {}).get("tool_calls") if isinstance(msg, dict) else None
|
||||
|
||||
if (msg_content is None or str(msg_content).strip() == "") and remaining_tool_calls:
|
||||
logging.warning("模型仍在请求工具调用，未返回JSON内容，重试下一次。")
|
||||
continue
|
||||
|
||||
combined_text = ""
|
||||
if isinstance(msg_reasoning, str) and msg_reasoning.strip():
|
||||
|
||||
1
backend_service/src/tools/__init__.py
Normal file
1
backend_service/src/tools/__init__.py
Normal file
@@ -0,0 +1 @@
|
||||
"""工具模块包。"""
|
||||
39
backend_service/src/tools/coordinate_tools.py
Normal file
39
backend_service/src/tools/coordinate_tools.py
Normal file
@@ -0,0 +1,39 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Dict, Tuple
|
||||
|
||||
|
||||
_DIRECTION_DELTAS: Dict[str, Tuple[int, int, int]] = {
|
||||
"east": (1, 0, 0),
|
||||
"west": (-1, 0, 0),
|
||||
"north": (0, 1, 0),
|
||||
"south": (0, -1, 0),
|
||||
"up": (0, 0, 1),
|
||||
"down": (0, 0, -1),
|
||||
}
|
||||
|
||||
|
||||
def calc_offset_enu(base: Dict[str, float], direction: str, distance: float) -> Dict[str, float]:
|
||||
"""在ENU坐标系下按方向偏移固定距离。"""
|
||||
if distance < 0:
|
||||
raise ValueError("distance must be non-negative")
|
||||
if not isinstance(base, dict):
|
||||
raise ValueError("base must be an object with x/y/z")
|
||||
|
||||
direction_key = (direction or "").strip().lower()
|
||||
if direction_key not in _DIRECTION_DELTAS:
|
||||
raise ValueError(f"unsupported direction: {direction}")
|
||||
|
||||
try:
|
||||
x = float(base["x"])
|
||||
y = float(base["y"])
|
||||
z = float(base["z"])
|
||||
except Exception as exc:
|
||||
raise ValueError("base must contain numeric x/y/z") from exc
|
||||
|
||||
dx, dy, dz = _DIRECTION_DELTAS[direction_key]
|
||||
offset_x = x + dx * distance
|
||||
offset_y = y + dy * distance
|
||||
offset_z = z + dz * distance
|
||||
|
||||
return {"x": offset_x, "y": offset_y, "z": offset_z}
|
||||
Reference in New Issue
Block a user