Spaces:
Runtime error
Runtime error
File size: 2,896 Bytes
1033417 8987ddb 1033417 6306cd3 1033417 6c496de 1033417 6c496de 1033417 5ffb4a3 1033417 5ffb4a3 1033417 c8740a0 1033417 5ffb4a3 |
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 |
import os
from dotenv import load_dotenv
from langchain.tools import Tool
from langchain.utilities import SerpAPIWrapper
from langgraph.graph.message import add_messages
from langgraph.graph import START, StateGraph
from langgraph.prebuilt import ToolNode, tools_condition
from langchain_core.messages import AnyMessage, HumanMessage
from langchain_groq import ChatGroq
from typing import TypedDict, Annotated
# Load API keys
load_dotenv()
groq_api_key = os.getenv("GROQ_API_KEY")
serpapi_api_key = os.getenv("SERPAPI_API_KEY")
# --- أدوات ---
def calculator_tool_func(query: str):
try:
result = str(eval(query, {"__builtins__": {}}))
return result
except Exception:
return "تعذر حساب التعبير المدخل."
calculator_tool = Tool(
name="Calculator",
func=calculator_tool_func,
description="Performs simple arithmetic calculations."
)
def weather_tool_func(location: str):
return f"حالة الطقس في {location}: مشمس، درجة الحرارة 25 درجة مئوية."
weather_tool = Tool(
name="Weather",
func=weather_tool_func,
description="Provides weather information for a given location."
)
serpapi = SerpAPIWrapper(serpapi_api_key=serpapi_api_key)
SerpAPI_tool = Tool(
name="WebSearch",
func=serpapi.run,
description="Searches the web for recent information."
)
# --- تعريف الموديل والوكلاء ---
tools = [SerpAPI_tool, calculator_tool, weather_tool]
llm = ChatGroq(model="deepseek-r1-distill-llama-70b", groq_api_key=groq_api_key)
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")
# 👇 compiled agent object (must be used with .invoke({...}))
ninu = builder.compile()
# --- دالة تشغيل مستقلة (اختياري) ---
def run_ninu(query: str):
conversation = []
intro_prompt = """
You are a helpful assistant. When you have the final answer, respond exactly with: FINAL ANSWER: [your answer here].
If you cannot answer, respond: FINAL ANSWER: I do not know.
"""
conversation.append(HumanMessage(content=intro_prompt))
conversation.append(HumanMessage(content=query))
response = ninu.invoke({"messages": conversation})
for message in reversed(response["messages"]):
if isinstance(message.content, str) and "FINAL ANSWER:" in message.content:
return message.content.split("FINAL ANSWER:")[-1].strip()
return "لم أتمكن من إيجاد إجابة نهائية."
|