Spaces:
Configuration error
Configuration error
from langchain_community.tools import DuckDuckGoSearchRun | |
from typing import TypedDict,Annotated | |
from langgraph.graph.message import add_messages | |
from langchain_core.messages import AnyMessage ,HumanMessage,AIMessage | |
from langgraph.prebuilt import ToolNode | |
from langgraph.graph import START,StateGraph | |
from langgraph.prebuilt import tools_condition | |
from langchain_groq import ChatGroq | |
from langchain.tools import Tool | |
from huggingface_hub import list_models | |
import random | |
from dotenv import load_dotenv | |
import os | |
from langchain_community.utilities import SerpAPIWrapper | |
load_dotenv() | |
os.environ["GROQ_API_KEY"]=os.getenv("GROQ_API_KEY") | |
#os.environ["SERPAPI_API_KEY"]=os.getenv("SERPAPI_API_KEY") | |
groq_api_key=os.getenv("GROQ_API_KEY") | |
serp_api_key=os.getenv("SERPAPI_API_KEY") | |
from langchain_community.utilities import SerpAPIWrapper | |
search = SerpAPIWrapper(serpapi_api_key=serp_api_key) | |
search_tool = Tool( | |
name="SerpAPI Search", | |
func=search.run, | |
description="Search the web using SerpAPI" | |
) | |
### weather tool | |
def get_weather_info(location: str) -> str: | |
"""Fetches dummy weather information for a given location.""" | |
# Dummy weather data | |
weather_conditions = [ | |
{"condition": "Rainy", "temp_c": 15}, | |
{"condition": "Clear", "temp_c": 25}, | |
{"condition": "Windy", "temp_c": 20} | |
] | |
# Randomly select a weather condition | |
data = random.choice(weather_conditions) | |
return f"Weather in {location}: {data['condition']}, {data['temp_c']}°C" | |
# Initialize the tool | |
weather_info_tool = Tool( | |
name="get_weather_info", | |
func=get_weather_info, | |
description="Fetches dummy weather information for a given location." | |
) | |
##most downloaded | |
def get_hub_stats(author: str) -> str: | |
"""Fetches the most downloaded model from a specific author on the Hugging Face Hub.""" | |
try: | |
# List models from the specified author, sorted by downloads | |
models = list(list_models(author=author, sort="downloads", direction=-1, limit=1)) | |
if models: | |
model = models[0] | |
return f"The most downloaded model by {author} is {model.id} with {model.downloads:,} downloads." | |
else: | |
return f"No models found for author {author}." | |
except Exception as e: | |
return f"Error fetching models for {author}: {str(e)}" | |
# Initialize the tool | |
hub_stats_tool = Tool( | |
name="get_hub_stats", | |
func=get_hub_stats, | |
description="Fetches the most downloaded model from a specific author on the Hugging Face Hub." | |
) | |
### langchain | |
import datasets | |
from langchain.docstore.document import Document | |
from langchain_community.retrievers import BM25Retriever | |
from langchain.tools import Tool | |
from typing import TypedDict, Annotated | |
from langgraph.graph.message import add_messages | |
from langchain_core.messages import AnyMessage,HumanMessage,AIMessage | |
from langgraph.prebuilt import ToolNode | |
from langgraph.graph import START,StateGraph | |
from langgraph.prebuilt import tools_condition | |
from langchain_huggingface import HuggingFaceEndpoint ,ChatHuggingFace | |
from dotenv import load_dotenv | |
from langchain_groq import ChatGroq | |
import os | |
load_dotenv() | |
os.environ["GROQ_API_KEY"]=os.getenv("GROQ_API_KEY") | |
groq_api_key=os.getenv("GROQ_API_KEY") | |
# Load the dataset | |
guest_dataset = datasets.load_dataset("agents-course/unit3-invitees", split="train") | |
# Convert dataset entries into Document objects | |
docs = [ | |
Document( | |
page_content="\n".join([ | |
f"Name: {guest['name']}", | |
f"Relation: {guest['relation']}", | |
f"Description: {guest['description']}", | |
f"Email: {guest['email']}" | |
]), | |
metadata={"name": guest["name"]} | |
) | |
for guest in guest_dataset | |
] | |
bm25_retriever = BM25Retriever.from_documents(docs) | |
def extract_text(query: str) -> str: | |
"""Retrieves detailed information about gala guests based on their name or relation.""" | |
results = bm25_retriever.invoke(query) | |
if results: | |
return "\n\n".join([doc.page_content for doc in results[:3]]) | |
else: | |
return "No matching guest information found." | |
guest_info_tool = Tool( | |
name="guest_info_retriever", | |
func=extract_text, | |
description="Retrieves detailed information about gala guests based on their name or relation." | |
) | |
# Generate the chat interface , including the tools | |
llm = ChatGroq(model="Gemma2-9b-It",groq_api_key=groq_api_key) | |
tools = [guest_info_tool,search_tool,weather_info_tool,hub_stats_tool] | |
llm_with_tools = llm.bind_tools(tools) | |
# Generate the AgentState and Agent graph | |
class AgentState(TypedDict): | |
messages: Annotated[list[AnyMessage], add_messages] | |
def assistant(state: AgentState): | |
return { | |
"messages": [llm_with_tools.invoke(state["messages"])], | |
} | |
## The graph | |
builder = StateGraph(AgentState) | |
# Define nodes: these do the work | |
builder.add_node("assistant", assistant) | |
builder.add_node("tools", ToolNode(tools)) | |
# Define edges: these determine how the control flow moves | |
builder.add_edge(START, "assistant") | |
builder.add_conditional_edges( | |
"assistant", | |
# If the latest message requires a tool, route to tools | |
# Otherwise, provide a direct response | |
tools_condition, | |
) | |
builder.add_edge("tools", "assistant") | |
alfred = builder.compile() | |
messages = [HumanMessage(content="Tell me about our guest named 'Lady Ada Lovelace'.")] | |
response = alfred.invoke({"messages": messages}) | |