more

globals.py

@@ -0,0 +1,30 @@
+from typing import TypedDict, Annotated
+import os
+import random
+
+from lmnr import Laminar
+from dotenv import load_dotenv
+load_dotenv()
+
+import datasets
+from huggingface_hub import list_models
+from langchain.docstore.document import Document
+from langchain_community.retrievers import BM25Retriever
+from langchain_community.tools import DuckDuckGoSearchRun
+from langchain.tools import Tool
+from langchain_ollama import ChatOllama
+from langchain_huggingface import HuggingFaceEndpoint,ChatHuggingFace
+from langchain_core.messages import AnyMessage, HumanMessage, AIMessage
+from langgraph.graph import StateGraph, START, END
+from langgraph.graph.message import add_messages
+from langgraph.prebuilt import tools_condition, ToolNode
+
+
+
+# GLOBALS
+HF_TOKEN = os.getenv('HF_TOKEN')
+PHOENIX_API_KEY = os.getenv('PHOENIX_API_KEY')
+LANGFUSE_PUBLIC_KEY = os.getenv('LANGFUSE_PUBLIC_KEY')
+LANGFUSE_SECRET_KEY= os.getenv('LANGFUSE_SECRET_KEY')
+LANGFUSE_HOST= os.getenv('LANGFUSE_HOST')
+LAMINAR_API_KEY= os.getenv('LAMINAR_API_KEY')

part_1.py

@@ -0,0 +1,73 @@
+from globals import *
+
+# model_name = 'qwen3:8b'
+model_name = 'llama3.2:latest'
+
+
+# Initialize Laminar - this single step enables automatic tracing
+Laminar.initialize(project_api_key=LAMINAR_API_KEY)
+
+llm = ChatOllama(model=model_name)
+
+
+# def load_guest_dataset():
+guest_dataset = datasets.load_dataset("agents-course/unit3-invitees", split="train")
+
+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 match!'
+
+guest_info_tool = Tool(
+    name='guest_info_retriever',
+    func=extract_text,
+    description='Retrieves detailed information about gala guests based on their name or relation.'
+)
+
+tools = [guest_info_tool]
+llm_with_tools = llm.bind_tools(tools)
+
+
+class AgentState(TypedDict):
+    messages: Annotated[list[AnyMessage], add_messages]
+
+def assistant(state: AgentState):
+    return {
+        'messages': [llm_with_tools.invoke(state['messages'])]
+    }
+
+builder = StateGraph(AgentState)
+
+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')
+alfred = builder.compile()
+with open("langgraph.png", "wb") as f:
+    f.write(alfred.get_graph().draw_mermaid_png())
+
+messages = [HumanMessage(content="Tell me about our guest named 'Lady Ada Lovelace'.")]
+response = alfred.invoke({'messages': messages})
+
+print("🎩 Alfred's Response:")
+print(response['messages'][-1].content)

part_2.py

@@ -0,0 +1,88 @@
+from globals import *
+# Initialize Laminar - this single step enables automatic tracing
+Laminar.initialize(project_api_key=LAMINAR_API_KEY)
+
+# model_name = 'qwen3:8b'
+model_name = 'llama3.2:latest'
+
+llm = ChatOllama(model=model_name)
+
+search_tool = DuckDuckGoSearchRun()
+# results: str = search_tool.invoke("Who's the current President of France?")
+# res_list = results.split('...')
+# for r in res_list:
+#     print(r)
+# print(results)
+
+def get_weather_info(location: str) -> str:
+    """Fetches weather info."""
+    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"
+
+weather_info_tool = Tool(
+    name='get_weather_info',
+    func=get_weather_info,
+    description='Fetches weather info for a given location.'
+)
+
+
+def get_hub_stats(author: str) -> str:
+    """Fetches the most downloaded model from the author."""
+    try:
+        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)}"
+
+hub_stats_tool = Tool(
+    name='get_hub_stats',
+    func=get_hub_stats,
+    description='Fetches the most downloaded model from the author.'
+)
+
+# print(hub_stats_tool.invoke('facebook'))
+
+tools = [search_tool, weather_info_tool, hub_stats_tool]
+chat_with_tools = llm.bind_tools(tools)
+
+
+class AgentState(TypedDict):
+    messages: Annotated[list[AnyMessage], add_messages]
+
+
+def assistant(state: AgentState):
+    return {
+        'messages': [chat_with_tools.invoke(state["messages"])]
+    }
+
+
+builder = StateGraph(AgentState)
+
+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')
+
+alfred = builder.compile()
+
+messages = [HumanMessage(content="Who is Facebook and what's their most downloaded model?")]
+response = alfred.invoke({'messages': messages})
+
+print("🎩 Alfred's response:")
+print(response['messages'][-1].content)
+
+

part_3.py

@@ -0,0 +1,74 @@
+from globals import *
+from tools import  search_tool, weather_info_tool, hub_stats_tool
+from retriever import guest_info_tool
+
+# Initialize Laminar - this single step enables automatic tracing
+Laminar.initialize(project_api_key=LAMINAR_API_KEY)
+
+# model_name = 'qwen3:8b'
+model_name = 'llama3.2:latest'
+
+llm = ChatOllama(model=model_name)
+
+tools = [guest_info_tool, search_tool, weather_info_tool, hub_stats_tool]
+chat_with_tools = llm.bind_tools(tools)
+
+
+class AgentState(TypedDict):
+    messages: Annotated[list[AnyMessage], add_messages]
+
+
+def assistant(state: AgentState):
+    return {
+        'messages': [chat_with_tools.invoke(state["messages"])]
+    }
+
+
+builder = StateGraph(AgentState)
+
+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')
+
+alfred = builder.compile()
+with open("langgraph.png", "wb") as f:
+    f.write(alfred.get_graph().draw_mermaid_png())
+
+response = alfred.invoke({'messages': "Tell me more about 'Lady Ada Lovelace'"})
+
+print("🎩 Alfred's response:")
+print(response['messages'][-1].content)
+
+response = alfred.invoke({"messages": "What's the weather like in Paris tonight? Will it be suitable for our fireworks display?"})
+
+print("🎩 Alfred's Response:")
+print(response['messages'][-1].content)
+
+response = alfred.invoke({"messages": "One of our guests is from Qwen. What can you tell me about their most popular model?"})
+
+print("🎩 Alfred's Response:")
+print(response['messages'][-1].content)
+
+
+response = alfred.invoke({"messages":"I need to speak with 'Dr. Nikola Tesla' about recent advancements in wireless energy. Can you help me prepare for this conversation?"})
+
+print("🎩 Alfred's Response:")
+print(response['messages'][-1].content)
+
+# First interaction
+response = alfred.invoke({"messages": [HumanMessage(content="Tell me about 'Lady Ada Lovelace'. What's her background and how is she related to me?")]})
+
+
+print("🎩 Alfred's Response:")
+print(response['messages'][-1].content)
+print()
+
+# Second interaction (referencing the first)
+response = alfred.invoke({"messages": response["messages"] + [HumanMessage(content="What projects is she currently working on?")]})
+
+print("🎩 Alfred's Response:")
+print(response['messages'][-1].content)
+

requirements.txt

@@ -1 +1 @@
-huggingface_hub==0.25.2
+huggingface_hub

retriever.py

@@ -0,0 +1,37 @@
+from globals import *
+
+# model_name = 'qwen3:8b'
+model_name = 'llama3.2:latest'
+
+
+# def load_guest_dataset():
+guest_dataset = datasets.load_dataset("agents-course/unit3-invitees", split="train")
+
+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 match!'
+
+guest_info_tool = Tool(
+    name='guest_info_retriever',
+    func=extract_text,
+    description='Retrieves detailed information about gala guests based on their name or relation.'
+)

tools.py

@@ -0,0 +1,45 @@
+from globals import *
+
+
+search_tool = DuckDuckGoSearchRun()
+
+
+def get_weather_info(location: str) -> str:
+    """Fetches weather info."""
+    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"
+
+weather_info_tool = Tool(
+    name='get_weather_info',
+    func=get_weather_info,
+    description='Fetches weather info for a given location.'
+)
+
+
+def get_hub_stats(author: str) -> str:
+    """Fetches the most downloaded model from the author."""
+    try:
+        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)}"
+
+hub_stats_tool = Tool(
+    name='get_hub_stats',
+    func=get_hub_stats,
+    description='Fetches the most downloaded model from the author.'
+)
+
+