diff --git a/pipeline_service/agent_loop_v2.py b/pipeline_service/agent_loop_v2.py index 02c9db3..552e13c 100644 --- a/pipeline_service/agent_loop_v2.py +++ b/pipeline_service/agent_loop_v2.py @@ -126,13 +126,59 @@ class AgentExecutor: if self.config.compression.enabled: await self._maybe_compress() - # 5b. LLM 调用 - raw = await self._call_llm() + # 5b. LLM 调用（native function calling，返回 dict） + resp = await self._call_llm() - # 5c. 解析 LLM 输出 + # 5c. 原生 function calling：优先处理 tool_calls + native_calls = (resp.get("tool_calls") or []) if isinstance(resp, dict) else [] + if native_calls: + # 回填 assistant 消息（含 tool_calls，OpenAI 原生格式要求） + self._msgs.append({ + "role": "assistant", + "content": resp.get("content") or None, + "tool_calls": native_calls, + }) + for tc in native_calls: + fn = tc.get("function", {}) if isinstance(tc, dict) else {} + tool_name = fn.get("name", "") + try: + tool_params = json.loads(fn.get("arguments") or "{}") + except Exception: + tool_params = {} + + # 需要确认的工具 + if self._needs_confirmation(tool_name): + yield json.dumps({ + "type": "confirm", "tool": tool_name, "params": tool_params, + }, ensure_ascii=False) + "\n" + await self._save_turn(user_input, f"[需确认] {tool_name}") + return + + yield json.dumps({ + "type": "tool_call", "tool": tool_name, "params": tool_params, + }, ensure_ascii=False) + "\n" + + result = await self._execute_tool(tool_name, tool_params) + if not result.startswith("未知工具") and not result.startswith("ERROR"): + self._tool_call_count += 1 + + yield json.dumps({ + "type": "tool_result", "tool": tool_name, "result": result[:500], + }, ensure_ascii=False) + "\n" + + # 原生回填：role=tool + tool_call_id + self._msgs.append({ + "role": "tool", + "tool_call_id": tc.get("id", "") if isinstance(tc, dict) else "", + "content": str(result), + }) + continue + + # 5d. 文本解析（回退路径，纯文本模型） + raw = resp.get("content", "") if isinstance(resp, dict) else str(resp) actions = self._parse_actions(raw) - # 5d. 分发处理 + # 5e. 分发处理 for act in actions: action_type = act.get("action", "") @@ -232,6 +278,12 @@ class AgentExecutor: try: from pipeline_core.tool_registry import get_tool_registry self._tool_registry = get_tool_registry() + # 把 config.tools 注册进 registry（registry 是全局单例，可能为空） + if self._tool_registry and self.config.tools: + existing = set(self._tool_registry.get_tool_names()) + for t in self.config.tools: + if t.name not in existing: + self._tool_registry.register(t) except ImportError: self._tool_registry = None @@ -315,22 +367,40 @@ class AgentExecutor: # LLM 调用 # ═══════════════════════════════════════════════════════ - async def _call_llm(self) -> str: - """调用 LLM，返回原始文本""" - from pipeline_service.llm_bridge import llm_call_msgs + async def _call_llm(self) -> dict: + """调用 LLM，返回 {"content": str, "tool_calls": [...]}。 + + 优先使用 OpenAI native function calling。若 registry 不可用或调用失败， + 回退到纯文本调用（content 为原始文本，tool_calls 为空）。 + """ + from pipeline_service.llm_bridge import llm_call_msgs_native - # 使用 OpenAI native function calling（如果模型支持） - use_native = False tools_schema = None - if self._tool_registry and use_native: - tools_schema = self._tool_registry.to_openai_schema() + if self._tool_registry: + try: + tools_schema = self._tool_registry.to_openai_schema() + except Exception as e: + logger.error(f"to_openai_schema failed: {e}") - # 简单消息调用（兼容所有模型） - return await llm_call_msgs( + if tools_schema: + try: + return await llm_call_msgs_native( + self._msgs, + tools=tools_schema, + model=self.model_name, + temperature=self.config.temperature, + ) + except Exception as e: + logger.error(f"native function calling failed, fallback to text: {e}") + + # 回退：纯文本调用（content 为原始文本） + from pipeline_service.llm_bridge import llm_call_msgs + content = await llm_call_msgs( self._msgs, model=self.model_name, temperature=self.config.temperature, ) + return {"content": content or "", "tool_calls": []} # ═══════════════════════════════════════════════════════ # 解析 LLM 输出 diff --git a/pipeline_service/llm_bridge.py b/pipeline_service/llm_bridge.py index ba90134..46c80ce 100644 --- a/pipeline_service/llm_bridge.py +++ b/pipeline_service/llm_bridge.py @@ -160,3 +160,47 @@ async def llm_call_msgs(messages: list, model: str = None, temperature: float = raise ValueError(f"LLM API error {resp.status}: {text[:300]}") data = await resp.json() return data["choices"][0]["message"]["content"] + + +async def llm_call_msgs_native(messages: list, tools: list = None, model: str = None, temperature: float = 0.7) -> dict: + """Native function calling. 传入 tools JSON schema，返回 message dict。 + + Returns: + {"content": str, "tool_calls": [{"id","type","function":{"name","arguments"}}]} + 当模型返回 tool_calls 时，content 通常为空字符串。 + """ + import aiohttp + + cfg = await _get_model_config(model) + if cfg.get("api_key") and cfg.get("api_base"): + api_base = cfg["api_base"] + api_key = cfg["api_key"] + model_id = cfg.get("model_id") or model or "default" + else: + api_base = os.environ.get("LLM_API_BASE", "https://api.openai.com/v1") + api_key = os.environ.get("LLM_API_KEY", "") + model_id = model or os.environ.get("LLM_MODEL", "gpt-4o-mini") + + if not api_key: + raise ValueError("No LLM API configured") + + headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"} + payload = {"model": model_id, "messages": messages, "temperature": temperature} + if tools: + payload["tools"] = tools + payload["tool_choice"] = "auto" + + url = api_base.rstrip("/") + "/chat/completions" + async with aiohttp.ClientSession() as session: + async with session.post(url, headers=headers, json=payload, + timeout=aiohttp.ClientTimeout(total=180)) as resp: + if resp.status != 200: + text = await resp.text() + raise ValueError(f"LLM API error {resp.status}: {text[:300]}") + data = await resp.json() + msg = data["choices"][0]["message"] + return { + "content": msg.get("content") or "", + "tool_calls": msg.get("tool_calls") or [], + } +