Spaces:
Running
Running
File size: 3,721 Bytes
2aa4095 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 |
from google.genai import types
from google.genai.types import Tool, GenerateContentConfig, GoogleSearch
from gemini_llm import GeminiLLM
from bs4 import BeautifulSoup
import requests
class GeminiThinking:
def __init__(self, geminillm: GeminiLLM, model_id="gemini-2.5-flash-preview-05-20"):
self.geminillm = geminillm
self.modelid = model_id
def activate_stream(self, prompt: str):
try:
stream = self.geminillm.client.models.generate_content_stream(
model=self.modelid,
contents=prompt,
config=types.GenerateContentConfig(
thinking_config=types.ThinkingConfig(include_thoughts=True)
)
)
for chunk in stream:
if not chunk.candidates:
continue
for part in chunk.candidates[0].content.parts:
if hasattr(part, "thought") and part.thought:
yield {"text": part.text, "thought": True}
elif hasattr(part, "text") and part.text:
yield {"text": part.text, "thought": False}
except Exception as e:
yield {"text": f"Error during GeminiThinking stream: {str(e)}", "thought": False}
class GoogleSearchTool:
def __init__(self, geminillm: GeminiLLM):
self.geminillm = geminillm
self.tool = Tool(google_search=GoogleSearch())
self.response = None
self.grounding = None
def activate_stream(self, prompt: str):
try:
stream = self.geminillm.client.models.generate_content_stream(
model=self.geminillm.model_id,
contents=prompt,
config=types.GenerateContentConfig(
tools=[self.tool],
response_modalities=["TEXT"]
)
)
for chunk in stream:
if not chunk.candidates:
continue
candidate = chunk.candidates[0]
self.grounding = candidate.grounding_metadata if hasattr(candidate, "grounding_metadata") else None
for part in candidate.content.parts:
if hasattr(part, "text") and part.text:
yield part.text
except Exception as e:
yield f"Error during GoogleSearchTool stream: {str(e)}"
class GeminiCodeExecutionTool:
def __init__(self, geminillm: GeminiLLM):
self.geminillm = geminillm
def activate_stream(self, prompt: str):
try:
stream = self.geminillm.client.models.generate_content_stream(
model=self.geminillm.model_id,
contents=prompt,
config=types.GenerateContentConfig(
tools=[types.Tool(code_execution=types.ToolCodeExecution())],
response_modalities=["TEXT"]
),
)
for chunk in stream:
if not chunk.candidates:
continue
candidate = chunk.candidates[0]
for part in candidate.content.parts:
if hasattr(part, "text") and part.text:
yield part.text
elif hasattr(part, "executable_code") and part.executable_code:
yield f"\n\n```python\n{part.executable_code.code}\n```\n"
elif hasattr(part, "code_execution_result") and part.code_execution_result:
yield f"\n\n**Output:**\n```\n{part.code_execution_result.output}\n```\n"
except Exception as e:
yield f"Error during code execution: {str(e)}"
|