Update tools.py
Browse files
tools.py
CHANGED
@@ -3,6 +3,8 @@ from langgraph.graph import Graph, StateGraph
|
|
3 |
from langgraph.prebuilt import ToolNode
|
4 |
from duckduckgo_search import DDGS
|
5 |
|
|
|
|
|
6 |
class CalculatorInput(TypedDict):
|
7 |
operation: str
|
8 |
numbers: List[float]
|
@@ -15,6 +17,8 @@ class CalculatorState(TypedDict):
|
|
15 |
input: CalculatorInput
|
16 |
output: Optional[CalculatorOutput]
|
17 |
|
|
|
|
|
18 |
class SearchInput(TypedDict):
|
19 |
query: str
|
20 |
max_results: int
|
@@ -29,21 +33,23 @@ class SearchOutput(TypedDict):
|
|
29 |
query: str
|
30 |
|
31 |
class SearchState(TypedDict):
|
32 |
-
input: SearchInput
|
33 |
output: Optional[SearchOutput]
|
34 |
|
|
|
|
|
35 |
def create_calculator_tool() -> Graph:
|
36 |
"""Creates a calculator tool using LangGraph that can perform basic arithmetic operations."""
|
37 |
print("Creating calculator tool")
|
38 |
-
|
39 |
def calculator_function(state: CalculatorState) -> Dict[str, Any]:
|
40 |
print("Calculator function called")
|
41 |
input_data = state["input"]
|
|
|
42 |
if len(input_data["numbers"]) < 2:
|
43 |
raise ValueError("At least two numbers are required for calculation")
|
44 |
-
|
45 |
result = input_data["numbers"][0]
|
46 |
-
|
47 |
for num in input_data["numbers"][1:]:
|
48 |
if input_data["operation"] == "add":
|
49 |
result += num
|
@@ -57,7 +63,7 @@ def create_calculator_tool() -> Graph:
|
|
57 |
result /= num
|
58 |
else:
|
59 |
raise ValueError(f"Unsupported operation: {input_data['operation']}")
|
60 |
-
|
61 |
return {
|
62 |
"output": {
|
63 |
"result": result,
|
@@ -65,27 +71,26 @@ def create_calculator_tool() -> Graph:
|
|
65 |
}
|
66 |
}
|
67 |
|
68 |
-
# Create the graph with state schema
|
69 |
workflow = StateGraph(state_schema=CalculatorState)
|
70 |
-
print("Calculator graph for workflow created")
|
71 |
-
# Add the calculator tool node
|
72 |
workflow.add_node("calculator", ToolNode(calculator_function))
|
73 |
-
print("Calculator tool node added")
|
74 |
-
# Set the entry and exit points
|
75 |
workflow.set_entry_point("calculator")
|
76 |
workflow.set_finish_point("calculator")
|
77 |
-
print("Calculator workflow
|
78 |
return workflow.compile()
|
79 |
|
|
|
|
|
80 |
def create_search_tool() -> Graph:
|
81 |
"""Creates a search tool using DuckDuckGo that can search for information online."""
|
82 |
-
|
|
|
83 |
def search_function(state: SearchState) -> Dict[str, Any]:
|
|
|
|
|
|
|
|
|
84 |
with DDGS() as ddgs:
|
85 |
-
raw_results = list(ddgs.text(
|
86 |
-
state["input"]["query"],
|
87 |
-
max_results=state["input"]["max_results"]
|
88 |
-
))
|
89 |
|
90 |
results = []
|
91 |
for r in raw_results:
|
@@ -101,35 +106,13 @@ def create_search_tool() -> Graph:
|
|
101 |
return {
|
102 |
"output": {
|
103 |
"results": results,
|
104 |
-
"query":
|
105 |
}
|
106 |
}
|
107 |
|
108 |
-
# Create the graph with state schema
|
109 |
workflow = StateGraph(state_schema=SearchState)
|
110 |
-
|
111 |
-
# Add the search tool node
|
112 |
workflow.add_node("search", ToolNode(search_function))
|
113 |
-
|
114 |
-
# Set the entry and exit points
|
115 |
workflow.set_entry_point("search")
|
116 |
workflow.set_finish_point("search")
|
117 |
-
|
118 |
return workflow.compile()
|
119 |
-
|
120 |
-
# Example usage:
|
121 |
-
# if __name__ == "__main__":
|
122 |
-
# # Create the calculator tool
|
123 |
-
# calculator = create_calculator_tool()
|
124 |
-
|
125 |
-
# # Example calculation
|
126 |
-
# result = calculator.invoke({
|
127 |
-
# "input": {
|
128 |
-
# "operation": "add",
|
129 |
-
# "numbers": [1, 2, 3, 4]
|
130 |
-
# }
|
131 |
-
# })
|
132 |
-
|
133 |
-
# print(f"Result: {result['output'].result}")
|
134 |
-
|
135 |
-
|
|
|
3 |
from langgraph.prebuilt import ToolNode
|
4 |
from duckduckgo_search import DDGS
|
5 |
|
6 |
+
# --- Calculator Tool Types ---
|
7 |
+
|
8 |
class CalculatorInput(TypedDict):
|
9 |
operation: str
|
10 |
numbers: List[float]
|
|
|
17 |
input: CalculatorInput
|
18 |
output: Optional[CalculatorOutput]
|
19 |
|
20 |
+
# --- Search Tool Types ---
|
21 |
+
|
22 |
class SearchInput(TypedDict):
|
23 |
query: str
|
24 |
max_results: int
|
|
|
33 |
query: str
|
34 |
|
35 |
class SearchState(TypedDict):
|
36 |
+
input: SearchInput
|
37 |
output: Optional[SearchOutput]
|
38 |
|
39 |
+
# --- Calculator Tool Implementation ---
|
40 |
+
|
41 |
def create_calculator_tool() -> Graph:
|
42 |
"""Creates a calculator tool using LangGraph that can perform basic arithmetic operations."""
|
43 |
print("Creating calculator tool")
|
44 |
+
|
45 |
def calculator_function(state: CalculatorState) -> Dict[str, Any]:
|
46 |
print("Calculator function called")
|
47 |
input_data = state["input"]
|
48 |
+
|
49 |
if len(input_data["numbers"]) < 2:
|
50 |
raise ValueError("At least two numbers are required for calculation")
|
51 |
+
|
52 |
result = input_data["numbers"][0]
|
|
|
53 |
for num in input_data["numbers"][1:]:
|
54 |
if input_data["operation"] == "add":
|
55 |
result += num
|
|
|
63 |
result /= num
|
64 |
else:
|
65 |
raise ValueError(f"Unsupported operation: {input_data['operation']}")
|
66 |
+
|
67 |
return {
|
68 |
"output": {
|
69 |
"result": result,
|
|
|
71 |
}
|
72 |
}
|
73 |
|
|
|
74 |
workflow = StateGraph(state_schema=CalculatorState)
|
|
|
|
|
75 |
workflow.add_node("calculator", ToolNode(calculator_function))
|
|
|
|
|
76 |
workflow.set_entry_point("calculator")
|
77 |
workflow.set_finish_point("calculator")
|
78 |
+
print("Calculator workflow created and compiled")
|
79 |
return workflow.compile()
|
80 |
|
81 |
+
# --- Search Tool Implementation ---
|
82 |
+
|
83 |
def create_search_tool() -> Graph:
|
84 |
"""Creates a search tool using DuckDuckGo that can search for information online."""
|
85 |
+
print("Creating search tool")
|
86 |
+
|
87 |
def search_function(state: SearchState) -> Dict[str, Any]:
|
88 |
+
print("Search function called")
|
89 |
+
query = state["input"]["query"]
|
90 |
+
max_results = state["input"].get("max_results", 3)
|
91 |
+
|
92 |
with DDGS() as ddgs:
|
93 |
+
raw_results = list(ddgs.text(query, max_results=max_results))
|
|
|
|
|
|
|
94 |
|
95 |
results = []
|
96 |
for r in raw_results:
|
|
|
106 |
return {
|
107 |
"output": {
|
108 |
"results": results,
|
109 |
+
"query": query
|
110 |
}
|
111 |
}
|
112 |
|
|
|
113 |
workflow = StateGraph(state_schema=SearchState)
|
|
|
|
|
114 |
workflow.add_node("search", ToolNode(search_function))
|
|
|
|
|
115 |
workflow.set_entry_point("search")
|
116 |
workflow.set_finish_point("search")
|
117 |
+
print("Search workflow created and compiled")
|
118 |
return workflow.compile()
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|