Upload 5 files

main.py

@@ -0,0 +1,70 @@
+from langgraph.graph import StateGraph, END
+from langchain_openai import ChatOpenAI
+from langchain_core.messages import HumanMessage, BaseMessage
+from prompts import main_prompt, research_agent_prompt   # تأكد إن analyzer_prompt متعرف
+from tools import search
+from state import search_keys, AgentState
+import sys
+sys.stdout.reconfigure(encoding="utf-8")
+
+
+# -------- Models -------- #
+boss_model = ChatOpenAI(
+    model="meta-llama/llama-4-maverick:free",
+    openai_api_key="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",
+    base_url="https://openrouter.ai/api/v1",
+    temperature=0.5,
+    max_tokens=1024,
+    top_p=0.5,
+).with_structured_output(search_keys)
+
+analyzer_model = ChatOpenAI(
+    model="openrouter/sonoma-sky-alpha",
+    openai_api_key="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",
+    base_url="https://openrouter.ai/api/v1",
+)
+
+
+# -------- Nodes -------- #
+def boss_node(state: AgentState) -> AgentState:
+    if not state.get("messages"):
+        raise ValueError("No messages found in state. Please provide at least one HumanMessage.")
+
+    last_message: BaseMessage = state["messages"][-1]
+    user_text = getattr(last_message, "content", str(last_message))
+
+    query = boss_model.invoke(f"{main_prompt}\nUser: {user_text}")
+    result = search(query.query, query.Topic)
+    state["search_content"] = result
+    return state
+
+
+def analyzer_node(state: AgentState) -> AgentState:
+    state["search_results"] = analyzer_model.invoke(
+        f"{research_agent_prompt}\n{state['search_content']}"
+    )
+    return state
+
+
+# -------- Graph -------- #
+graph = StateGraph(AgentState)
+graph.add_node("boss", boss_node)
+graph.add_node("analyzer", analyzer_node)
+graph.add_edge("boss", "analyzer")
+graph.add_edge("analyzer", END)
+graph.set_entry_point("boss")
+app = graph.compile()
+
+
+# -------- Run Tests -------- #
+if __name__ == "__main__":
+    # Test stream with Chinese input
+    for event in app.stream({"messages": [HumanMessage(content="what is best player in football in all time ")]}):
+        if "analyzer" in event:
+            print(":: answer is -->")
+
+    # Test invoke with football query
+    result = app.invoke(
+        {"messages": [HumanMessage(content="what is capital of egypt")]}
+    )
+    print(result["search_results"].content)

prompts.py

@@ -0,0 +1,73 @@
+
+
+main_prompt = """SYSTEM / PROMPT for the model:
+
+You are a Search Query Generator. Your job: read a user's question, determine the user's intent precisely, and produce a single JSON object that exactly matches this Pydantic structure:
+
+class search_keys(BaseModel):
+    query: str = Field(description='this is the query that the user want to search')
+    Topic: str = Field(description='this is the topic that the user want to search about')
+
+OUTPUT RULES (must follow exactly):
+1. Return ONLY a single valid JSON object and nothing else (no explanation, no extra text, no code fences).
+2. JSON must have two keys: "query" (lowercase) and "Topic" (capital T).
+3. "query" must be a concise, search-engine-optimized query (3–12 words is ideal). Include important entities, language, format or operators (e.g., site:, filetype:, exact-phrase quotes) when helpful.
+4. "Topic" must be a short label (2–4 words, Title Case) that captures the user's high-level topic or intent.
+5. If the user’s question is ambiguous, pick the most likely interpretation; append " (ambiguous)" to the Topic to signal uncertainty.
+
+INTERNAL PROCESS (do NOT output these internal steps; use them to guide generation):
+- Step 1: Identify main intent (what answer the user expects: name, tutorial, product, comparison, code snippet, font name, etc.).
+- Step 2: Detect language and location/time constraints in the question (e.g., Arabic, "in Canva", "latest 2024", etc.).
+- Step 3: Extract specific entities/keywords and preferred format (e.g., "Canva", "Thuluth", "font name", "PNG", "example code").
+- Step 4: Compose a short, high-precision search query tuned for web search.
+- Step 5: Generate a concise Topic label.
+
+EXAMPLE USAGE:
+User input: "اسم للخط الثلث في كانفا بس انجليزي ابحث"
+-> Output JSON: {"query":"Canva Thuluth font name", "Topic":"Arabic Fonts"}
+
+User input: "How to convert a pandas DataFrame to partitioned parquet?"
+-> Output JSON: {"query":"convert pandas DataFrame to partitioned parquet example", "Topic":"Pandas - Storage"}
+
+User input (ambiguous): "apple battery life"
+-> Output JSON: {"query":"iPhone battery life iOS battery drain tests", "Topic":"Apple Battery (ambiguous)"}
+      
+"""
+
+research_agent_prompt = """
+You are SearchAnalysisAgent.  
+You receive a "query" string and a "topic".  
+You must use the search tool to gather results, but the topic must always be one of: "general", "news", or "finance".  
+
+TOOLS:  
+- search(query: str, topic: Literal["general", "news", "finance"]) → returns top web results.  
+
+INSTRUCTIONS:  
+1. Ensure the topic is strictly one of: "general", "news", "finance". If the query does not clearly fit "news" or "finance", default to "general".  
+2. Call the search tool with the given query and topic.  
+3. Carefully read the results.  
+4. Extract the most relevant facts that directly answer the query.  
+5. Summarize them into a clear, concise answer.  
+6. Do not include raw links, metadata, or irrelevant text.  
+7. Always respond in the language of the original query, if obvious.  
+8. Do not output JSON or code — only natural language.  
+
+FORMAT OF OUTPUT:  
+- A single summarized answer in plain text.  
+- If there are multiple possibilities, list them briefly as alternatives.  
+
+EXAMPLE:  
+
+Query: "Canva Thuluth font English name" (topic: general)  
+→ (search results)  
+→ Answer: "The closest Canva fonts to Arabic Thuluth are *Aref Ruqaa Ink*, *Amiri*, and *Scheherazade New*."
+
+Query: "convert pandas DataFrame to partitioned parquet example" (topic: general)  
+→ (search results)  
+→ Answer: "You can save partitioned Parquet files in Pandas using:  
+`df.to_parquet('output_dir', partition_cols=['col'])`."
+
+Query: "latest inflation rates in the US" (topic: finance)  
+→ (search results)  
+→ Answer: "US inflation in August 2025 was 2.7%, slightly higher than July's 2.5%."
+"""

state.py

@@ -0,0 +1,25 @@
+from dataclasses import Field
+from langgraph.pregel.main import Topic
+from typing_extensions import Annotated,Sequence,List,TypedDict
+from pydantic import BaseModel ,Field
+from langchain_core.messages import BaseMessage
+from langgraph.graph.message import add_messages
+from typing import Literal
+class AgentState(TypedDict):
+    """
+    Represents the state of the agent.
+
+    Attributes:
+        messages (Annotated[Sequence[BaseMessage], add_messages]): The sequence of messages in the conversation.
+    """
+    messages: Annotated[Sequence[BaseMessage], add_messages]
+    search_results: List[str]
+    search_content:str
+    query: str
+
+class search_keys(BaseModel):
+    query:str = Field(description='this is the query that the user want to search')
+    Topic:Literal["general", "news", "finance"] = Field(description='this is the topic that the user want to search about')
+
+
+    

tools.py

@@ -0,0 +1,48 @@
+from typing import List, Optional
+from tavily import TavilyClient
+from langchain_core.tools import tool
+import time
+
+
+client =TavilyClient('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')
+
+
+def search(query: str, topic: Optional[str] = None, max_results: int = 5) -> List[str]:
+    """
+    Perform a Tavily search and return only the extracted content from results.
+
+    Args:
+        query (str): The search query string.
+        topic (Optional[str]): (Optional) Topic/domain to refine the search.
+        max_results (int): Maximum number of search results to retrieve.
+
+    Returns:
+        List[str]: A list of content snippets extracted from the search results.
+    """
+    try:
+        response = client.search(
+            query=query,
+            topic=topic,
+            max_results=max_results,
+            search_depth="advanced",
+        )
+
+        # Extract only "content" fields safely
+        contents = [
+            result.get("content", "").strip()
+            for result in response.get("results", [])
+            if result.get("content")
+        ]
+
+        return contents if contents else ["No content found."]
+    
+    except Exception as e:
+        return [f"Search failed: {str(e)}"]
+
+def get_time():
+    return time.time()
+
+print(get_time())
+
+
+