forked from trpc-group/trpc-agent-python
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtools.py
More file actions
79 lines (65 loc) · 2.4 KB
/
Copy pathtools.py
File metadata and controls
79 lines (65 loc) · 2.4 KB
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
# Tencent is pleased to support the open source community by making tRPC-Agent-Python available.
#
# Copyright (C) 2026 Tencent. All rights reserved.
#
# tRPC-Agent-Python is licensed under Apache-2.0.
""" Tools for the LangGraph agent. """
from langchain_core.tools import tool
from trpc_agent_sdk.agents import langgraph_tool_node
@tool
@langgraph_tool_node
def calculate(operation: str, a: float, b: float) -> str:
"""Perform basic mathematical operations.
This function simulates a slow calculation (2 seconds) to demonstrate
cancellation during tool execution in LangGraph.
Args:
operation: Operation type ('add', 'subtract', 'multiply', 'divide')
a: First number
b: Second number
Returns:
Calculation result as a string
"""
# Simulate slow calculation - this gives us time to cancel
print(f"[Tool executing: calculating {a} {operation} {b}...]", flush=True)
try:
if operation == "add":
result = a + b
elif operation == "subtract":
result = a - b
elif operation == "multiply":
result = a * b
elif operation == "divide":
if b == 0:
return "Error: Cannot divide by zero"
result = a / b
else:
return f"Error: Unknown operation '{operation}'"
print(f"[Tool completed: result = {result}]", flush=True)
return f"Calculation result: {a} {operation} {b} = {result}"
except Exception as e: # pylint: disable=broad-except
return f"Calculation error: {str(e)}"
@tool
@langgraph_tool_node
def analyze_data(data_type: str, sample_size: int) -> str:
"""Analyze data and generate statistical report.
This function simulates a longer analysis (3 seconds) to demonstrate
cancellation during extended tool execution.
Args:
data_type: Type of data to analyze ('sales', 'user_behavior', 'performance')
sample_size: Number of data points to analyze
Returns:
Analysis report as a string
"""
print(f"[Tool executing: analyzing {sample_size} {data_type} data points...]", flush=True)
# Simulate data analysis result
report = f"""
Data Analysis Report:
- Data Type: {data_type}
- Sample Size: {sample_size}
- Mean: 42.5
- Median: 40.0
- Std Dev: 15.3
- Key Insight: Data shows positive trend
"""
print(f"[Tool completed: analysis done]", flush=True)
return report.strip()