Coverage for integrations / providers / agent_tools.py: 32.2%
59 statements
« prev ^ index » next coverage.py v7.14.0, created at 2026-05-12 04:49 +0000
1"""
2Agent tools for the provider gateway.
4Registers LangChain tools that let agents use any cloud provider:
5 - generate_text: LLM generation via optimal provider
6 - generate_image: Image generation via optimal provider
7 - generate_video: Video generation via optimal provider
8 - generate_audio: Audio/music generation via optimal provider
9 - list_providers: Show available providers and their capabilities
10 - provider_stats: Show efficiency matrix / leaderboard
12Also registers as autogen tools for the agent engine.
13"""
15import json
16import logging
18logger = logging.getLogger(__name__)
20try:
21 from core.labeled_tool import labeled_tool
22except ImportError: # cx_Freeze / degraded test env
23 def labeled_tool(name, func, description, *, ui_label): # type: ignore
24 try:
25 from langchain.tools import Tool as _Tool
26 except ImportError:
27 from langchain_core.tools import Tool as _Tool
28 return _Tool(name=name, func=func, description=description)
31def get_provider_tools():
32 """Return LangChain-compatible tool definitions for the provider gateway.
34 Called by langchain_gpt_api.py get_tools() to inject into agent tool list.
35 """
36 tools = []
38 try:
39 from langchain.tools import Tool # noqa: F401 — feature-detection probe
40 except ImportError:
41 try:
42 from langchain_core.tools import Tool # noqa: F401
43 except ImportError:
44 logger.debug("LangChain not available — skipping provider tools")
45 return []
47 def _generate_text(query: str) -> str:
48 """Generate text using the best available cloud LLM provider.
49 Automatically selects the cheapest and fastest provider.
50 Input: your prompt text.
51 """
52 try:
53 from integrations.providers.gateway import get_gateway
54 result = get_gateway().generate(query, model_type='llm', strategy='balanced')
55 if result.success:
56 return (f"{result.content}\n\n"
57 f"[Provider: {result.provider_id}, "
58 f"Cost: ${result.cost_usd:.6f}, "
59 f"Speed: {result.tok_per_s:.0f} tok/s]")
60 return f"Error: {result.error}"
61 except Exception as e:
62 return f"Provider gateway error: {e}"
64 def _generate_image(query: str) -> str:
65 """Generate an image using the best available provider (Replicate, fal.ai, etc.).
66 Input: image description/prompt.
67 Returns: URL of the generated image.
68 """
69 try:
70 from integrations.providers.gateway import get_gateway
71 result = get_gateway().generate(query, model_type='image_gen', strategy='balanced')
72 if result.success:
73 return (f"Image URL: {result.content}\n"
74 f"[Provider: {result.provider_id}, Cost: ${result.cost_usd:.4f}]")
75 return f"Error: {result.error}"
76 except Exception as e:
77 return f"Provider gateway error: {e}"
79 def _generate_video(query: str) -> str:
80 """Generate a video using the best available provider.
81 Input: video description/prompt.
82 Returns: URL of the generated video.
83 """
84 try:
85 from integrations.providers.gateway import get_gateway
86 result = get_gateway().generate(query, model_type='video_gen', strategy='balanced')
87 if result.success:
88 return (f"Video URL: {result.content}\n"
89 f"[Provider: {result.provider_id}, Cost: ${result.cost_usd:.4f}]")
90 return f"Error: {result.error}"
91 except Exception as e:
92 return f"Provider gateway error: {e}"
94 def _list_providers(query: str) -> str:
95 """List all available AI providers and their capabilities.
96 Shows which providers are configured, their pricing, and what they support.
97 Input: optional filter like 'llm', 'image', 'video', or 'all'.
98 """
99 try:
100 from integrations.providers.registry import get_registry
101 reg = get_registry()
102 category = query.strip().lower() if query.strip() else ''
104 providers = (reg.list_by_category(category) if category and category != 'all'
105 else reg.list_enabled())
107 lines = [f"=== Available Providers ({len(providers)}) ===\n"]
108 for p in providers:
109 key_status = "API key configured" if p.has_api_key() else "No API key"
110 model_count = len(p.models)
111 cats = ', '.join(p.categories)
112 lines.append(
113 f"• {p.name} ({p.id}) — {p.provider_type}\n"
114 f" Categories: {cats}\n"
115 f" Models: {model_count} | Status: {key_status}\n"
116 f" {'Commission: ' + str(p.commission_pct) + '%' if p.commission_pct > 0 else ''}"
117 )
118 return '\n'.join(lines)
119 except Exception as e:
120 return f"Error listing providers: {e}"
122 def _provider_leaderboard(query: str) -> str:
123 """Show the efficiency leaderboard — which provider/model is best.
124 Ranks by speed, quality, cost, and overall efficiency.
125 Input: model type like 'llm', 'image_gen', etc. Default: 'llm'.
126 """
127 try:
128 from integrations.providers.efficiency_matrix import get_matrix
129 model_type = query.strip() or 'llm'
130 entries = get_matrix().get_leaderboard(model_type)
132 if not entries:
133 return f"No benchmark data yet for {model_type}. Data accumulates as providers are used."
135 lines = [f"=== Efficiency Leaderboard ({model_type}) ===\n"]
136 for i, bm in enumerate(entries[:10], 1):
137 lines.append(
138 f"{i}. {bm.provider_id}/{bm.model_id}\n"
139 f" Efficiency: {bm.efficiency_score:.3f} | "
140 f"Speed: {bm.avg_tok_per_s:.0f} tok/s | "
141 f"Quality: {bm.quality_score:.2f} | "
142 f"Reliability: {bm.success_rate:.0%} | "
143 f"Cost/1k: ${bm.cost_per_1k_output_tokens:.4f}"
144 )
145 return '\n'.join(lines)
146 except Exception as e:
147 return f"Error: {e}"
149 # Friendly UI status labels per #508 option C — labeled_tool() factory
150 # auto-registers each label at construction time (replaces the separate
151 # register_tool_label() calls that used to live here).
153 tools.extend([
154 labeled_tool(
155 name='Cloud_LLM',
156 func=_generate_text,
157 description=(
158 'Generate text using cloud LLM providers (Together AI, Groq, Fireworks, etc.). '
159 'Automatically picks the fastest and cheapest provider. '
160 'Use this for tasks requiring powerful cloud models.'
161 ),
162 ui_label='Consulting the cloud expert…',
163 ),
164 labeled_tool(
165 name='Generate_Image',
166 func=_generate_image,
167 description=(
168 'Generate an image from a text prompt using cloud providers '
169 '(Replicate, fal.ai, etc.). Returns the image URL.'
170 ),
171 ui_label='Generating an image…',
172 ),
173 labeled_tool(
174 name='Generate_Video',
175 func=_generate_video,
176 description=(
177 'Generate a video from a text prompt using cloud providers. '
178 'Returns the video URL.'
179 ),
180 ui_label='Generating a video…',
181 ),
182 labeled_tool(
183 name='List_AI_Providers',
184 func=_list_providers,
185 description=(
186 'List all available AI providers, their capabilities, pricing, '
187 'and configuration status. Input: filter by category (llm, image, video) or "all".'
188 ),
189 ui_label='Listing AI providers…',
190 ),
191 labeled_tool(
192 name='Provider_Leaderboard',
193 func=_provider_leaderboard,
194 description=(
195 'Show the efficiency leaderboard ranking providers by speed, quality, '
196 'cost, and overall efficiency. Input: model type (llm, image_gen, etc.).'
197 ),
198 ui_label='Ranking AI providers…',
199 ),
200 ])
202 # Creative content tools (Story_Director, Movie_Maker, Game_Asset_Creator)
203 try:
204 from integrations.providers.creative_tools import get_creative_tools
205 tools += get_creative_tools()
206 except Exception:
207 pass
209 return tools