Spaces:
Sleeping
Sleeping
File size: 3,414 Bytes
9e26969 c7e93c3 88b3058 836390e 9e26969 88b3058 9e26969 dea6adc 9e26969 dea6adc 9e26969 dea6adc 9e26969 dea6adc |
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 103 |
import os
from langchain.agents import AgentExecutor, create_react_agent
from langchain.tools import tool
from langgraph.prebuilt import tools_condition
from langgraph.graph import START, StateGraph, MessagesState
from langgraph.prebuilt import ToolNode
from langchain.prompts import PromptTemplate
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_huggingface import ChatHuggingFace, HuggingFaceEndpoint, HuggingFaceEmbeddings
from langchain_core.messages import SystemMessage, HumanMessage
#from langchain_openai import ChatOpenAI # OpenAI-compatible for Groq API
from ddgs import DDGS # Updated DuckDuckGo Search
from dotenv import load_dotenv
import re
import json
# Load environment variables
load_dotenv()
# --- Define Tools ---
@tool
def python_code_executor(code: str) -> str:
"""Execute Python code and return the result as a string. Use for calculations or data processing."""
try:
local_vars = {}
exec(code, {}, local_vars)
return str(local_vars.get("result", "No result defined. Set 'result' variable."))
except Exception as e:
return f"Error: {str(e)}"
@tool
def download_file(url: str) -> str:
"""Download a file from URL and return its content (text if possible)."""
try:
response = requests.get(url, timeout=10)
response.raise_for_status()
return response.text[:1000] # Truncate for brevity
except Exception as e:
return f"Error downloading: {str(e)}"
@tool
def duckduckgo_search(query: str) -> str:
"""Perform a DuckDuckGo search and return top results as a short summary."""
try:
with DDGS() as ddgs:
results = list(ddgs.text(query, max_results=3))
if not results:
return "No good results found."
return json.dumps([{"title": r["title"], "snippet": r["body"]} for r in results])
except Exception as e:
return f"Search error: {str(e)}"
# load the system prompt from the file
with open("system_prompt.txt", "r", encoding="utf-8") as f:
system_prompt = f.read()
# System message
sys_msg = SystemMessage(content=system_prompt)
tools = [
python_code_executor,
download_file,
duckduckgo_search,
]
# Build graph function
#def build_graph(provider: str = "huggingface"):
def build_graph(provider: str = "google"):
"""Build the graph"""
# Load environment variables from .env file
if provider == "google":
# Google Gemini
llm = ChatGoogleGenerativeAI(model="gemini-2.0-flash", temperature=0)
elif provider == "huggingface":
llm = ChatHuggingFace(
llm=HuggingFaceEndpoint(
repo_id = "Qwen/Qwen2.5-Coder-32B-Instruct"
),
)
else:
raise ValueError("Invalid provider. Choose 'google', 'groq' or 'huggingface'.")
# Bind tools to LLM
llm_with_tools = llm.bind_tools(tools)
# Node
def assistant(state: MessagesState):
"""Assistant node"""
return {"messages": [llm_with_tools.invoke([sys_msg] + state["messages"])]}
builder = StateGraph(MessagesState)
builder.add_node("assistant", assistant)
builder.add_node("tools", ToolNode(tools))
builder.add_edge(START, "assistant")
builder.add_conditional_edges(
"assistant",
tools_condition,
)
builder.add_edge("tools", "assistant")
# Compile graph
return builder.compile() |