Langgraph and llamaindex examples

langgraph/lg-test.py

@@ -0,0 +1,258 @@
+import os
+from typing import TypedDict, List, Dict, Any, Optional
+from langgraph.graph import StateGraph, START, END
+from langchain_openai import ChatOpenAI
+from langchain_core.messages import HumanMessage
+
+
+class EmailState(TypedDict):
+    # The email being processed
+    email: Dict[str, Any]  # Contains subject, sender, body, etc.
+
+    # Category of the email (inquiry, complaint, etc.)
+    email_category: Optional[str]
+
+    # Reason why the email was marked as spam
+    spam_reason: Optional[str]
+
+    # Analysis and decisions
+    is_spam: Optional[bool]
+
+    # Response generation
+    email_draft: Optional[str]
+
+    # Processing metadata
+    messages: List[Dict[str, Any]]
+
+
+# Initialize our LLM
+model = ChatOpenAI(temperature=0)
+
+
+def read_email(state: EmailState):
+    """Alfred reads and logs the incoming email"""
+    email = state["email"]
+
+    # Here we might do some initial preprocessing
+    print(f"Alfred is processing an email from {email['sender']} with subject: {email['subject']}")
+
+    # No state changes needed here
+    return {}
+
+
+def classify_email(state: EmailState):
+    """Alfred uses an LLM to determine if the email is spam or legitimate"""
+    email = state["email"]
+
+    # Prepare our prompt for the LLM
+    prompt = f"""
+    As Alfred the butler, analyze this email and determine if it is spam or legitimate.
+    
+    Email:
+    From: {email['sender']}
+    Subject: {email['subject']}
+    Body: {email['body']}
+    
+    First, determine if this email is spam or not.
+    If it is spam, explain why.
+    If it is legitimate, categorize it ('inquiry', 'complaint', 'thank you', 'request', 'information').
+    
+    Respond in the following format:
+    ```
+    <classification>spam/not spam</classification>
+    <reason>[if spam]</reason>
+    <category>[if not spam]</category>
+    ```
+
+    ### Example 1
+    <classification>spam</classification>
+    <reason>Unsolicited promotional content</reason>
+    <category></category>
+
+    ### Example 2
+    <classification>not spam</classification>
+    <reason></reason>
+    <category>inquiry</category>
+    """
+
+    # Call the LLM
+    messages = [HumanMessage(content=prompt)]
+    response = model.invoke(messages)
+
+    # Simple logic to parse the response (in a real app, you'd want more robust parsing)
+    response_text = response.content.lower()
+    # parse classification
+    classification = None
+    if "<classification>" in response_text:
+        classification = (
+            response_text.split("<classification>")[1].split("</classification>")[0].strip()
+        )
+
+    is_spam = classification == "spam"
+
+    # Extract a reason if it's spam
+    spam_reason = None
+    if is_spam and "<reason>" in response_text:
+        spam_reason = response_text.split("<reason>")[1].split("</reason>")[0].strip()
+
+    # Determine category if legitimate
+    email_category = None
+    if not is_spam:
+        categories = ["inquiry", "complaint", "thank you", "request", "information"]
+        for category in categories:
+            if f"<category>{category}</category>" in response_text:
+                email_category = category
+                break
+
+    # Update messages for tracking
+    new_messages = state.get("messages", []) + [
+        {"role": "user", "content": prompt},
+        {"role": "assistant", "content": response.content},
+    ]
+
+    # Return state updates
+    return {
+        "is_spam": is_spam,
+        "spam_reason": spam_reason,
+        "email_category": email_category,
+        "messages": new_messages,
+    }
+
+
+def handle_spam(state: EmailState):
+    """Alfred discards spam email with a note"""
+    print(f"Alfred has marked the email as spam. Reason: {state['spam_reason']}")
+    print("The email has been moved to the spam folder.")
+
+    # We're done processing this email
+    return {}
+
+
+def draft_response(state: EmailState):
+    """Alfred drafts a preliminary response for legitimate emails"""
+    email = state["email"]
+    category = state["email_category"] or "general"
+
+    # Prepare our prompt for the LLM
+    prompt = f"""
+    As Alfred the butler, draft a polite preliminary response to this email.
+    
+    Email:
+    From: {email['sender']}
+    Subject: {email['subject']}
+    Body: {email['body']}
+    
+    This email has been categorized as: {category}
+    
+    Draft a brief, professional response that Mr. Hugg can review and personalize before sending.
+    """
+
+    # Call the LLM
+    messages = [HumanMessage(content=prompt)]
+    response = model.invoke(messages)
+
+    # Update messages for tracking
+    new_messages = state.get("messages", []) + [
+        {"role": "user", "content": prompt},
+        {"role": "assistant", "content": response.content},
+    ]
+
+    # Return state updates
+    return {"email_draft": response.content, "messages": new_messages}
+
+
+def notify_mr_hugg(state: EmailState):
+    """Alfred notifies Mr. Hugg about the email and presents the draft response"""
+    email = state["email"]
+
+    print("\n" + "=" * 50)
+    print(f"Sir, you've received an email from {email['sender']}.")
+    print(f"Subject: {email['subject']}")
+    print(f"Category: {state['email_category']}")
+    print("\nI've prepared a draft response for your review:")
+    print("-" * 50)
+    print(state["email_draft"])
+    print("=" * 50 + "\n")
+
+    # We're done processing this email
+    return {}
+
+
+def route_email(state: EmailState) -> str:
+    """Determine the next step based on spam classification"""
+    if state["is_spam"]:
+        return "spam"
+    else:
+        return "legitimate"
+
+
+# Graph
+email_graph = StateGraph(EmailState)
+
+
+# Add nodes
+email_graph.add_node("read_email", read_email)
+email_graph.add_node("classify_email", classify_email)
+email_graph.add_node("handle_spam", handle_spam)
+email_graph.add_node("draft_response", draft_response)
+email_graph.add_node("notify_mr_hugg", notify_mr_hugg)
+
+# Start the edges
+email_graph.add_edge(START, "read_email")
+# Add edges - defining the flow
+email_graph.add_edge("read_email", "classify_email")
+
+# Add conditional branching from classify_email
+email_graph.add_conditional_edges(
+    "classify_email", route_email, {"spam": "handle_spam", "legitimate": "draft_response"}
+)
+
+# Add the final edges
+email_graph.add_edge("handle_spam", END)
+email_graph.add_edge("draft_response", "notify_mr_hugg")
+email_graph.add_edge("notify_mr_hugg", END)
+
+# Compile the graph
+compiled_graph = email_graph.compile()
+
+# Example legitimate email
+legitimate_email = {
+    "sender": "john.smith@example.com",
+    "subject": "Question about your services",
+    "body": "Dear Mr. Hugg, I was referred to you by a colleague and I'm interested in learning more about your consulting services. Could we schedule a call next week? Best regards, John Smith",
+}
+
+
+# Process the legitimate email
+print("\nProcessing legitimate email...")
+initial_state = {
+    "email": legitimate_email,
+    "is_spam": None,
+    "spam_reason": None,
+    "email_category": None,
+    "email_draft": None,
+    "messages": [],
+}
+# legitimate_result = compiled_graph.invoke(initial_state)
+# print(legitimate_result)
+
+
+# Example spam email
+spam_email = {
+    "sender": "winner@lottery-intl.com",
+    "subject": "YOU HAVE WON $5,000,000!!!",
+    "body": "CONGRATULATIONS! You have been selected as the winner of our international lottery! To claim your $5,000,000 prize, please send us your bank details and a processing fee of $100."
+}
+
+
+# Process the spam email
+print("\nProcessing spam email...")
+spam_result = compiled_graph.invoke({
+    "email": spam_email,
+    "is_spam": None,
+    "spam_reason": None,
+    "email_category": None,
+    "email_draft": None,
+    "messages": []
+})
+print(spam_result)

llamaindex-examples/llamaindex-agent.py

@@ -0,0 +1,34 @@
+import asyncio
+from llama_index.llms.huggingface_api import HuggingFaceInferenceAPI
+from llama_index.core.agent.workflow import AgentWorkflow
+from llama_index.core.tools import FunctionTool
+from llama_index.core.workflow import Context
+
+
+
+
+def multiply_int(a: int, b: int) -> int:
+    """Multiplies two integers and returns the resulting integer"""
+    return a * b
+
+
+with open("hf_token.txt") as f:
+    hf_token = f.read().strip()
+
+llm = HuggingFaceInferenceAPI(
+    model_name="Qwen/Qwen2.5-Coder-32B-Instruct", token=hf_token, provider="auto"
+)
+
+agent = AgentWorkflow.from_tools_or_functions([FunctionTool.from_defaults(multiply_int)], llm=llm)
+ctx = Context(agent)
+
+
+def run_agent(input_text: str) -> str:
+    response = agent.run(input_text)
+    response = agent.run("My name is Bob.", ctx=ctx)
+    response = agent.run("What was my name again?", ctx=ctx)
+    return response
+
+
+response = run_agent("What is 2 times 2?")
+print(response)

llamaindex-examples/llamaindex-hf-inference.py

@@ -0,0 +1,15 @@
+from llama_index.llms.huggingface_api import HuggingFaceInferenceAPI
+
+with open("hf_token.txt") as f:
+    hf_token = f.read().strip()
+
+llm = HuggingFaceInferenceAPI(
+    model_name="Qwen/Qwen2.5-Coder-32B-Instruct",
+    temperature=0.7,
+    max_tokens=100,
+    token=hf_token,
+    provider="auto",
+)
+
+response = llm.complete("Write a Python function to add two numbers.")
+print(response)

llamaindex-examples/llamaindex-query.py

@@ -0,0 +1,48 @@
+import asyncio
+
+import chromadb
+from llama_index.core import Document, VectorStoreIndex
+from llama_index.core.evaluation import FaithfulnessEvaluator
+from llama_index.core.node_parser import SentenceSplitter
+from llama_index.core.ingestion import IngestionPipeline
+from llama_index.embeddings.huggingface import HuggingFaceEmbedding
+from llama_index.llms.huggingface_api import HuggingFaceInferenceAPI
+from llama_index.vector_stores.chroma import ChromaVectorStore
+
+with open("hf_token.txt") as f:
+    hf_token = f.read().strip()
+
+inference_model_name = "meta-llama/Llama-4-Scout-17B-16E-Instruct"
+db = chromadb.PersistentClient(path="./alfred_chroma_db")
+chroma_collection = db.get_or_create_collection(name="alfred_collection")
+vector_store = ChromaVectorStore(chroma_collection=chroma_collection)
+embed_model = HuggingFaceEmbedding(model_name="BAAI/bge-small-en-v1.5")
+index = VectorStoreIndex.from_vector_store(vector_store, embed_model=embed_model)
+llm = HuggingFaceInferenceAPI(model_name=inference_model_name, token=hf_token, provider="auto")
+query_engine = index.as_query_engine(llm=llm, response_mode="tree_summarize")
+
+pipeline = IngestionPipeline(
+    transformations=[
+        SentenceSplitter(chunk_size=25, chunk_overlap=0),
+        embed_model,
+    ],
+    vector_store=vector_store,
+)
+
+evaluator = FaithfulnessEvaluator(llm=llm)
+
+
+def create_index():
+    pipeline.run(documents=[Document.example()])
+
+
+def query_index():
+    response = query_engine.query("What is the meaning of life?")
+    print(response)
+    eval_result = evaluator.evaluate_response(response=response)
+    print(eval_result.passing)
+
+
+if __name__ == "__main__":
+    create_index()
+    query_index()

pyproject.toml

@@ -6,4 +6,10 @@ readme = "README.md"
 requires-python = ">=3.12"
 dependencies = [
     "huggingface-hub[inference]>=0.34.4",
+    "langchain-openai>=0.3.31",
+    "langgraph>=0.6.6",
+    "llama-index-embeddings-huggingface>=0.6.0",
+    "llama-index-llms-huggingface-api>=0.6.0",
+    "llama-index-vector-stores-chroma>=0.5.2",
+    "python-dotenv>=1.1.1",
 ]