File size: 3,913 Bytes
399fe38 ebec9e2 f5f2409 399fe38 ebec9e2 399fe38 ebec9e2 4beca24 f5f2409 a5f0e08 399fe38 b15267e 399fe38 b15267e 4beca24 f5f2409 4beca24 f5f2409 ebec9e2 e038027 f5f2409 4beca24 e038027 399fe38 f5f2409 399fe38 ebec9e2 f5f2409 399fe38 ebec9e2 399fe38 ebec9e2 399fe38 ebec9e2 399fe38 ebec9e2 399fe38 f5f2409 477c627 399fe38 477c627 2b5d5db 4beca24 2b5d5db ebec9e2 2b5d5db ebec9e2 2b5d5db ebec9e2 2b5d5db f5f2409 ebec9e2 f5f2409 ebec9e2 f5f2409 4beca24 f5f2409 ebec9e2 f5f2409 2683234 399fe38 a22806f 399fe38 2683234 399fe38 f5f2409 399fe38 2683234 ebec9e2 4beca24 ebec9e2 f5f2409 ebec9e2 |
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 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 |
from typing import Dict, Any, List, TypedDict, Optional
from langgraph.graph import Graph, StateGraph
from langgraph.prebuilt import ToolNode
from duckduckgo_search import DDGS
# --- Calculator Tool Types ---
class CalculatorInput(TypedDict):
operation: str
numbers: List[float]
class CalculatorOutput(TypedDict):
result: float
operation: str
class CalculatorState(TypedDict):
input: CalculatorInput
output: Optional[CalculatorOutput]
# --- Search Tool Types ---
class SearchInput(TypedDict):
query: str
max_results: int
class SearchResult(TypedDict):
title: str
link: str
snippet: str
class SearchOutput(TypedDict):
results: List[SearchResult]
query: str
class SearchState(TypedDict):
input: SearchInput
output: Optional[SearchOutput]
# --- Calculator Tool Implementation ---
def create_calculator_tool() -> Graph:
"""Creates a calculator tool using LangGraph that can perform basic arithmetic operations."""
print("Creating calculator tool")
def calculator_function(state: CalculatorState) -> Dict[str, Any]:
print("Calculator function called")
input_data = state["input"]
if len(input_data["numbers"]) < 2:
raise ValueError("At least two numbers are required for calculation")
result = input_data["numbers"][0]
for num in input_data["numbers"][1:]:
if input_data["operation"] == "add":
result += num
elif input_data["operation"] == "subtract":
result -= num
elif input_data["operation"] == "multiply":
result *= num
elif input_data["operation"] == "divide":
if num == 0:
raise ValueError("Cannot divide by zero")
result /= num
else:
raise ValueError(f"Unsupported operation: {input_data['operation']}")
return {
"output": {
"result": result,
"operation": input_data["operation"]
}
}
print("Calculator function defined")
workflow = StateGraph(state_schema=CalculatorState)
print("Calculator workflow created")
workflow.add_node("calculator", ToolNode(calculator_function))
print("Calculator tool node added")
workflow.set_entry_point("calculator")
print("Calculator workflow set entry point")
workflow.set_finish_point("calculator")
print("Calculator workflow set finish point")
print("Calculator workflow created and compiled")
return workflow.compile()
# --- Search Tool Implementation ---
def create_search_tool() -> Graph:
"""Creates a search tool using DuckDuckGo that can search for information online."""
print("Creating search tool")
def search_function(state: SearchState) -> Dict[str, Any]:
print("Search function called")
query = state["input"]["query"]
max_results = state["input"].get("max_results", 3)
with DDGS() as ddgs:
raw_results = list(ddgs.text(query, max_results=max_results))
results = []
for r in raw_results:
try:
results.append({
"title": r.get("title", ""),
"link": r.get("href", r.get("link", "")),
"snippet": r.get("body", r.get("snippet", ""))
})
except Exception as e:
print("Skipping malformed search result:", r, "Error:", e)
return {
"output": {
"results": results,
"query": query
}
}
workflow = StateGraph(state_schema=SearchState)
workflow.add_node("search", ToolNode(search_function))
workflow.set_entry_point("search")
workflow.set_finish_point("search")
print("Search workflow created and compiled")
return workflow.compile()
|