Spaces:
Sleeping
Sleeping
| """ | |
| LLM provider abstraction with tool-calling support (live web search). | |
| Launch on the free provider (Groq). When you're ready to upgrade, set | |
| LLM_PROVIDER=anthropic and ANTHROPIC_API_KEY - nothing else needs to change. | |
| Note: web-search tool-calling is currently wired up for the Groq path only; | |
| the Anthropic path still works, just without live search for now. | |
| """ | |
| import json | |
| from config import settings | |
| from search import WEB_SEARCH_TOOL_SCHEMA, search_web | |
| _groq_client = None | |
| _anthropic_client = None | |
| MAX_TOOL_ITERATIONS = 3 | |
| def _get_groq(): | |
| global _groq_client | |
| if _groq_client is None: | |
| from groq import Groq | |
| if not settings.GROQ_API_KEY: | |
| raise RuntimeError( | |
| "GROQ_API_KEY is not set. Get a free key at https://console.groq.com/keys" | |
| ) | |
| _groq_client = Groq(api_key=settings.GROQ_API_KEY) | |
| return _groq_client | |
| def _get_anthropic(): | |
| global _anthropic_client | |
| if _anthropic_client is None: | |
| import anthropic | |
| if not settings.ANTHROPIC_API_KEY: | |
| raise RuntimeError("ANTHROPIC_API_KEY is not set.") | |
| _anthropic_client = anthropic.Anthropic(api_key=settings.ANTHROPIC_API_KEY) | |
| return _anthropic_client | |
| async def generate_reply(system_prompt: str, messages: list[dict]) -> tuple[str, bool]: | |
| """ | |
| messages: list of {"role": "user"|"assistant", "content": str}, oldest first. | |
| Returns (reply_text, used_web_search). | |
| """ | |
| provider = settings.LLM_PROVIDER.lower() | |
| if provider == "groq": | |
| return await _generate_groq(system_prompt, messages) | |
| if provider == "anthropic": | |
| client = _get_anthropic() | |
| resp = client.messages.create( | |
| model=settings.ANTHROPIC_MODEL, | |
| max_tokens=500, | |
| system=system_prompt, | |
| messages=messages, | |
| ) | |
| text = "".join(b.text for b in resp.content if b.type == "text").strip() | |
| return text, False | |
| raise RuntimeError(f"Unknown LLM_PROVIDER: '{provider}'. Use 'groq' or 'anthropic'.") | |
| async def _generate_groq(system_prompt: str, messages: list[dict]) -> tuple[str, bool]: | |
| client = _get_groq() | |
| api_messages = [{"role": "system", "content": system_prompt}] + list(messages) | |
| tools_enabled = bool(settings.TAVILY_API_KEY) | |
| used_web_search = False | |
| for _ in range(MAX_TOOL_ITERATIONS): | |
| kwargs = dict( | |
| model=settings.GROQ_MODEL, | |
| messages=api_messages, | |
| max_tokens=500, | |
| temperature=0.7, | |
| ) | |
| if tools_enabled: | |
| kwargs["tools"] = [WEB_SEARCH_TOOL_SCHEMA] | |
| kwargs["tool_choice"] = "auto" | |
| resp = client.chat.completions.create(**kwargs) | |
| message = resp.choices[0].message | |
| if not message.tool_calls: | |
| return (message.content or "").strip(), used_web_search | |
| # The model wants to call a tool - append its request, then the result(s). | |
| api_messages.append( | |
| { | |
| "role": "assistant", | |
| "content": message.content or "", | |
| "tool_calls": [ | |
| { | |
| "id": tc.id, | |
| "type": "function", | |
| "function": {"name": tc.function.name, "arguments": tc.function.arguments}, | |
| } | |
| for tc in message.tool_calls | |
| ], | |
| } | |
| ) | |
| for tc in message.tool_calls: | |
| if tc.function.name == "search_web": | |
| args = json.loads(tc.function.arguments or "{}") | |
| result_text = await search_web(args.get("query", "")) | |
| used_web_search = True | |
| else: | |
| result_text = f"Unknown tool: {tc.function.name}" | |
| api_messages.append( | |
| { | |
| "role": "tool", | |
| "tool_call_id": tc.id, | |
| "content": result_text, | |
| } | |
| ) | |
| # Ran out of tool-calling rounds - ask once more without tools to force | |
| # a final answer rather than looping forever. | |
| resp = client.chat.completions.create( | |
| model=settings.GROQ_MODEL, | |
| messages=api_messages, | |
| max_tokens=500, | |
| temperature=0.7, | |
| ) | |
| return (resp.choices[0].message.content or "").strip(), used_web_search | |