File size: 3,920 Bytes
399fe38 ebec9e2 399fe38 ebec9e2 399fe38 ebec9e2 399fe38 b15267e 399fe38 b15267e ebec9e2 e038027 399fe38 e038027 399fe38 ebec9e2 399fe38 ebec9e2 399fe38 ebec9e2 399fe38 ebec9e2 399fe38 ebec9e2 399fe38 ebec9e2 399fe38 ebec9e2 477c627 399fe38 477c627 ebec9e2 73caeb8 399fe38 e038027 ebec9e2 e038027 ebec9e2 e038027 ebec9e2 399fe38 ebec9e2 2683234 399fe38 ebec9e2 2683234 399fe38 2683234 399fe38 2683234 ebec9e2 73caeb8 399fe38 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 124 125 |
from typing import Dict, Any, List, TypedDict, Optional
from langgraph.graph import Graph, StateGraph
from langgraph.prebuilt import ToolNode
from duckduckgo_search import DDGS
class CalculatorInput(TypedDict):
operation: str
numbers: List[float]
class CalculatorOutput(TypedDict):
result: float
operation: str
class SearchResult(TypedDict):
title: str
link: str
snippet: str
class SearchOutput(TypedDict):
results: List[SearchResult]
query: str
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: Dict[str, Any]) -> dict:
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"]
}
}
# Create the graph with state schema
workflow = StateGraph(state_schema=Dict[str, Any])
print("Calculator graph for workflow created")
# Add the calculator tool node
workflow.add_node("calculator", ToolNode(calculator_function))
print("Calculator tool node added")
# Set the entry and exit points
workflow.set_entry_point("calculator")
workflow.set_finish_point("calculator")
print("Calculator workflow set")
return workflow.compile()
def create_search_tool() -> Graph:
"""Creates a search tool using DuckDuckGo that can search for information online."""
def search_function(state: Dict[str, Any]) -> dict:
with DDGS() as ddgs:
# Run search
raw_results = list(ddgs.text(
state["input"]["query"],
max_results=state["input"].get("max_results", 3)
))
results = []
for r in raw_results:
try:
results.append({
"title": r.get("title", ""),
"link": r.get("href", r.get("link", "")), # DuckDuckGo sometimes uses "href"
"snippet": r.get("body", r.get("snippet", ""))
})
except Exception as e:
print("Skipping malformed search result:", r, "Error:", e)
return {
"output": {
"results": results,
"query": state["input"]["query"]
}
}
# Create the graph with state schema
workflow = StateGraph(state_schema=Dict[str, Any])
# Add the search tool node
workflow.add_node("search", ToolNode(search_function))
# Set the entry and exit points
workflow.set_entry_point("search")
workflow.set_finish_point("search")
return workflow.compile()
# Example usage:
# if __name__ == "__main__":
# # Create the calculator tool
# calculator = create_calculator_tool()
# # Example calculation
# result = calculator.invoke({
# "input": {
# "operation": "add",
# "numbers": [1, 2, 3, 4]
# }
# })
# print(f"Result: {result['output'].result}")
|