Repository navigation
Expand file tree
/
Copy pathagent.py
More file actions
150 lines (123 loc) · 5.26 KB
/
Copy pathagent.py
File metadata and controls
150 lines (123 loc) · 5.26 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
import logging
from collections.abc import AsyncIterable
from typing import Any, Literal
import httpx
from langchain_core.messages import AIMessage, ToolMessage
from langchain_core.runnables.config import (
RunnableConfig,
)
from langchain_core.tools import tool # type: ignore
from langchain_google_genai import ChatGoogleGenerativeAI
from langgraph.checkpoint.memory import MemorySaver
from langgraph.prebuilt import create_react_agent # type: ignore
from pydantic import BaseModel
logger = logging.getLogger(__name__)
memory = MemorySaver()
@tool
def get_exchange_rate(
currency_from: str = 'USD',
currency_to: str = 'EUR',
currency_date: str = 'latest',
):
"""Use this to get current exchange rate.
Args:
currency_from: The currency to convert from (e.g., "USD").
currency_to: The currency to convert to (e.g., "EUR").
currency_date: The date for the exchange rate or "latest". Defaults to "latest".
Returns:
A dictionary containing the exchange rate data, or an error message if the request fails.
"""
try:
response = httpx.get(
f'https://api.frankfurter.app/{currency_date}',
params={'from': currency_from, 'to': currency_to},
)
response.raise_for_status()
data = response.json()
if 'rates' not in data:
logger.error(f'rates not found in response: {data}')
return {'error': 'Invalid API response format.'}
logger.info(f'API response: {data}')
return data
except httpx.HTTPError as e:
logger.error(f'API request failed: {e}')
return {'error': f'API request failed: {e}'}
except ValueError:
logger.error('Invalid JSON response from API')
return {'error': 'Invalid JSON response from API.'}
class ResponseFormat(BaseModel):
"""Respond to the user in this format."""
status: Literal['input_required', 'completed', 'error'] = 'input_required'
message: str
class CurrencyAgent:
"""Currency Conversion Agent Example."""
SYSTEM_INSTRUCTION = (
'You are a specialized assistant for currency conversions. '
"Your sole purpose is to use the 'get_exchange_rate' tool to answer questions about currency exchange rates. "
'If the user asks about anything other than currency conversion or exchange rates, '
'politely state that you cannot help with that topic and can only assist with currency-related queries. '
'Do not attempt to answer unrelated questions or use tools for other purposes.'
)
RESPONSE_FORMAT_INSTRUCTION: str = (
'Select status as completed if the request is complete'
'Select status as input_required if the input is a question to the user'
'Set response status to error if the input indicates an error'
)
def __init__(self):
self.model = ChatGoogleGenerativeAI(model='gemini-2.0-flash')
self.tools = [get_exchange_rate]
self.graph = create_react_agent(
self.model,
tools=self.tools,
checkpointer=memory,
prompt=self.SYSTEM_INSTRUCTION,
response_format=(self.RESPONSE_FORMAT_INSTRUCTION, ResponseFormat),
)
async def stream(
self, query: str, sessionId: str
) -> AsyncIterable[dict[str, Any]]:
inputs: dict[str, Any] = {'messages': [('user', query)]}
config: RunnableConfig = {'configurable': {'thread_id': sessionId}}
for item in self.graph.stream(inputs, config, stream_mode='values'):
message = item['messages'][-1]
if (
isinstance(message, AIMessage)
and message.tool_calls
and len(message.tool_calls) > 0
):
yield {
'is_task_complete': False,
'require_user_input': False,
'content': 'Looking up the exchange rates...',
}
elif isinstance(message, ToolMessage):
yield {
'is_task_complete': False,
'require_user_input': False,
'content': 'Processing the exchange rates..',
}
yield self.get_agent_response(config)
def get_agent_response(self, config: RunnableConfig) -> dict[str, Any]:
current_state = self.graph.get_state(config)
structured_response = current_state.values.get('structured_response')
if structured_response and isinstance(
structured_response, ResponseFormat
):
if structured_response.status in {'input_required', 'error'}:
return {
'is_task_complete': False,
'require_user_input': True,
'content': structured_response.message,
}
if structured_response.status == 'completed':
return {
'is_task_complete': True,
'require_user_input': False,
'content': structured_response.message,
}
return {
'is_task_complete': False,
'require_user_input': True,
'content': 'We are unable to process your request at the moment. Please try again.',
}
SUPPORTED_CONTENT_TYPES = ['text', 'text/plain']