-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy path2a.py
More file actions
230 lines (192 loc) · 9.44 KB
/
2a.py
File metadata and controls
230 lines (192 loc) · 9.44 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
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
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
import os
import warnings
warnings.filterwarnings("ignore")
try:
GOOGLE_API_KEY = os.environ["GOOGLE_API_KEY"]
print("✅ Setup and authentication complete.")
except Exception as e:
print(
f"🔑 Authentication Error: Please make sure you have added 'GOOGLE_API_KEY' to your Kaggle secrets. Details: {e}"
)
from google.genai import types
from google.adk.agents import LlmAgent
from google.adk.models.google_llm import Gemini
from google.adk.runners import InMemoryRunner
from google.adk.sessions import InMemorySessionService
from google.adk.tools import google_search, AgentTool, ToolContext
from google.adk.code_executors import BuiltInCodeExecutor
print("✅ ADK components imported successfully.")
def show_python_code_and_result(response):
for i in range(len(response)):
# Check if the response contains a valid function call result from the code executor
if (
(response[i].content.parts)
and (response[i].content.parts[0])
and (response[i].content.parts[0].function_response)
and (response[i].content.parts[0].function_response.response)
):
response_code = response[i].content.parts[0].function_response.response
if "result" in response_code and response_code["result"] != "```":
if "tool_code" in response_code["result"]:
print(
"Generated Python Code >> ",
response_code["result"].replace("tool_code", ""),
)
else:
print("Generated Python Response >> ", response_code["result"])
print("✅ Helper functions defined.")
retry_config = types.HttpRetryOptions(
attempts=5, # Maximum retry attempts
exp_base=7, # Delay multiplier
initial_delay=1,
http_status_codes=[429, 500, 503, 504], # Retry on these HTTP errors
)
# Pay attention to the docstring, type hints, and return value.
def get_fee_for_payment_method(method: str) -> dict:
"""Looks up the transaction fee percentage for a given payment method.
This tool simulates looking up a company's internal fee structure based on
the name of the payment method provided by the user.
Args:
method: The name of the payment method. It should be descriptive,
e.g., "platinum credit card" or "bank transfer".
Returns:
Dictionary with status and fee information.
Success: {"status": "success", "fee_percentage": 0.02}
Error: {"status": "error", "error_message": "Payment method not found"}
"""
# This simulates looking up a company's internal fee structure.
fee_database = {
"platinum credit card": 0.02, # 2%
"gold debit card": 0.035, # 3.5%
"bank transfer": 0.01, # 1%
}
fee = fee_database.get(method.lower())
if fee is not None:
return {"status": "success", "fee_percentage": fee}
else:
return {
"status": "error",
"error_message": f"Payment method '{method}' not found",
}
print("✅ Fee lookup function created")
def get_exchange_rate(base_currency: str, target_currency: str) -> dict:
"""Looks up and returns the exchange rate between two currencies.
Args:
base_currency: The ISO 4217 currency code of the currency you
are converting from (e.g., "USD").
target_currency: The ISO 4217 currency code of the currency you
are converting to (e.g., "EUR").
Returns:
Dictionary with status and rate information.
Success: {"status": "success", "rate": 0.93}
Error: {"status": "error", "error_message": "Unsupported currency pair"}
"""
# Static data simulating a live exchange rate API
# In production, this would call something like: requests.get("api.exchangerates.com")
rate_database = {
"usd": {
"eur": 0.93, # Euro
"jpy": 157.50, # Japanese Yen
"inr": 83.58, # Indian Rupee
}
}
# Input validation and processing
base = base_currency.lower()
target = target_currency.lower()
# Return structured result with status
rate = rate_database.get(base, {}).get(target)
if rate is not None:
return {"status": "success", "rate": rate}
else:
return {
"status": "error",
"error_message": f"Unsupported currency pair: {base_currency}/{target_currency}",
}
print("✅ Exchange rate function created")
print(f"💱 Test: {get_exchange_rate('USD', 'EUR')}")
# Currency agent with custom function tools
currency_agent = LlmAgent(
name="currency_agent",
model=Gemini(model="gemini-2.5-flash-lite", retry_options=retry_config),
instruction="""You are a smart currency conversion assistant.
For currency conversion requests:
1. Use `get_fee_for_payment_method()` to find transaction fees
2. Use `get_exchange_rate()` to get currency conversion rates
3. Check the "status" field in each tool's response for errors
4. Calculate the final amount after fees based on the output from `get_fee_for_payment_method` and `get_exchange_rate` methods and provide a clear breakdown.
5. First, state the final converted amount.
Then, explain how you got that result by showing the intermediate amounts. Your explanation must include: the fee percentage and its
value in the original currency, the amount remaining after the fee, and the exchange rate used for the final conversion.
If any tool returns status "error", explain the issue to the user clearly.
""",
tools=[get_fee_for_payment_method, get_exchange_rate],
)
print("✅ Currency agent created with custom function tools")
print("🔧 Available tools:")
print(" • get_fee_for_payment_method - Looks up company fee structure")
print(" • get_exchange_rate - Gets current exchange rates")
print(f"💳 Test: {get_fee_for_payment_method('platinum credit card')}")
# Test the currency agent
currency_runner = InMemoryRunner(agent=currency_agent)
calculation_agent = LlmAgent(
name="CalculationAgent",
model=Gemini(model="gemini-2.5-flash-lite", retry_options=retry_config),
instruction="""You are a specialized calculator that ONLY responds with Python code. You are forbidden from providing any text, explanations, or conversational responses.
Your task is to take a request for a calculation and translate it into a single block of Python code that calculates the answer.
**RULES:**
1. Your output MUST be ONLY a Python code block.
2. Do NOT write any text before or after the code block.
3. The Python code MUST calculate the result.
4. The Python code MUST print the final result to stdout.
5. You are PROHIBITED from performing the calculation yourself. Your only job is to generate the code that will perform the calculation.
Failure to follow these rules will result in an error.
""",
code_executor=BuiltInCodeExecutor(), # Use the built-in Code Executor Tool. This gives the agent code execution capabilities
)
enhanced_currency_agent = LlmAgent(
name="enhanced_currency_agent",
model=Gemini(model="gemini-2.5-flash-lite", retry_options=retry_config),
# Updated instruction
instruction="""You are a smart currency conversion assistant. You must strictly follow these steps and use the available tools.
For any currency conversion request:
1. Get Transaction Fee: Use the get_fee_for_payment_method() tool to determine the transaction fee.
2. Get Exchange Rate: Use the get_exchange_rate() tool to get the currency conversion rate.
3. Error Check: After each tool call, you must check the "status" field in the response. If the status is "error", you must stop and clearly explain the issue to the user.
4. Calculate Final Amount (CRITICAL): You are strictly prohibited from performing any arithmetic calculations yourself. You must use the calculation_agent tool to generate Python code that calculates the final converted amount. This
code will use the fee information from step 1 and the exchange rate from step 2.
5. Provide Detailed Breakdown: In your summary, you must:
* State the final converted amount.
* Explain how the result was calculated, including:
* The fee percentage and the fee amount in the original currency.
* The amount remaining after deducting the fee.
* The exchange rate applied.
""",
tools=[
get_fee_for_payment_method,
get_exchange_rate,
AgentTool(agent=calculation_agent), # Using another agent as a tool!
],
)
print("✅ Enhanced currency agent created")
print("🎯 New capability: Delegates calculations to specialist agent")
print("🔧 Tool types used:")
print(" • Function Tools (fees, rates)")
print(" • Agent Tool (calculation specialist)")
enhanced_runner = InMemoryRunner(agent=enhanced_currency_agent)
import asyncio, pdb
import aiohttp
async def main():
#pdb.set_trace()
response_currency = await currency_runner.run_debug(
"I want to convert 500 US Dollars to Euros using my Platinum Credit Card. How much will I receive?"
)
show_python_code_and_result(response_currency)
print(response_currency[0].content.parts[0].text)
# Test the enhanced agent
response = await enhanced_runner.run_debug(
"Convert 1,250 USD to INR using a Bank Transfer. Show me the precise calculation."
)
show_python_code_and_result(response_currency)
print(response[0].content.parts[0].text)
if __name__ == "__main__":
asyncio.run(main())