Update app.py
Browse files
app.py
CHANGED
@@ -29,12 +29,6 @@ class BasicAgent:
|
|
29 |
if not HF_TOKEN:
|
30 |
raise ValueError("HF_TOKEN environment variable not set. Please set your Hugging Face API token.")
|
31 |
|
32 |
-
# Initialize tools
|
33 |
-
print("In constructor of basic agent")
|
34 |
-
self.calculator = create_calculator_tool()
|
35 |
-
print("Calculator tool created")
|
36 |
-
self.search_tool = create_search_tool()
|
37 |
-
print("Search tool created")
|
38 |
# Set up headers for API calls
|
39 |
self.headers = {
|
40 |
"Authorization": f"Bearer {HF_TOKEN}",
|
@@ -54,22 +48,17 @@ class BasicAgent:
|
|
54 |
print("Stategraph created")
|
55 |
# Add nodes
|
56 |
workflow.add_node("analyze", self._analyze_question)
|
57 |
-
workflow.add_node("calculator", self._use_calculator)
|
58 |
workflow.add_node("search", self._use_search)
|
59 |
workflow.add_node("final_answer", self._generate_final_answer)
|
60 |
|
61 |
# Define edges
|
62 |
-
workflow.add_edge("analyze", "calculator")
|
63 |
workflow.add_edge("analyze", "search")
|
64 |
workflow.add_edge("analyze", "final_answer")
|
65 |
-
workflow.add_edge("calculator", "final_answer")
|
66 |
workflow.add_edge("search", "final_answer")
|
67 |
|
68 |
# Define conditional edges
|
69 |
def router(state: AgentState) -> str:
|
70 |
-
if state["current_step"] == '
|
71 |
-
return 'calculator'
|
72 |
-
elif state["current_step"] == 'search':
|
73 |
return 'search'
|
74 |
elif state["current_step"] == 'final_answer':
|
75 |
return 'final_answer'
|
@@ -79,7 +68,6 @@ class BasicAgent:
|
|
79 |
"analyze",
|
80 |
router,
|
81 |
{
|
82 |
-
"calculator": "calculator",
|
83 |
"search": "search",
|
84 |
"final_answer": "final_answer"
|
85 |
}
|
@@ -110,13 +98,8 @@ class BasicAgent:
|
|
110 |
prompt = f"""Analyze this question and determine what needs to be done: {state['question']}
|
111 |
Return your analysis in this format:
|
112 |
{{
|
113 |
-
"needs_calculation": true/false,
|
114 |
"needs_search": true/false,
|
115 |
-
"search_query": "query if needed"
|
116 |
-
"calculation": {{
|
117 |
-
"operation": "add/subtract/multiply/divide",
|
118 |
-
"numbers": [numbers if needed]
|
119 |
-
}}
|
120 |
}}
|
121 |
"""
|
122 |
|
@@ -124,37 +107,13 @@ class BasicAgent:
|
|
124 |
state["needs_more_info"] = analysis.get('needs_search', False)
|
125 |
state["search_query"] = analysis.get('search_query', '')
|
126 |
|
127 |
-
if analysis.get('
|
128 |
-
state["current_step"] = 'calculator'
|
129 |
-
state["tool_output"] = str(analysis['calculation'])
|
130 |
-
elif analysis.get('needs_search', False):
|
131 |
state["current_step"] = 'search'
|
132 |
else:
|
133 |
state["current_step"] = 'final_answer'
|
134 |
|
135 |
return state
|
136 |
|
137 |
-
def _use_calculator(self, state: AgentState) -> AgentState:
|
138 |
-
"""Use the calculator tool."""
|
139 |
-
try:
|
140 |
-
# Create calculator state with input from tool_output
|
141 |
-
calc_input = eval(state["tool_output"])
|
142 |
-
result = self.calculator.invoke({"input": calc_input})
|
143 |
-
|
144 |
-
state["history"].append({
|
145 |
-
'step': 'calculator',
|
146 |
-
'input': state["tool_output"],
|
147 |
-
'output': str(result['output']['result'])
|
148 |
-
})
|
149 |
-
state["current_step"] = 'final_answer'
|
150 |
-
except Exception as e:
|
151 |
-
state["history"].append({
|
152 |
-
'step': 'calculator_error',
|
153 |
-
'error': str(e)
|
154 |
-
})
|
155 |
-
state["current_step"] = 'final_answer'
|
156 |
-
return state
|
157 |
-
|
158 |
def _use_search(self, state: AgentState) -> AgentState:
|
159 |
"""Use the search tool."""
|
160 |
try:
|
|
|
29 |
if not HF_TOKEN:
|
30 |
raise ValueError("HF_TOKEN environment variable not set. Please set your Hugging Face API token.")
|
31 |
|
|
|
|
|
|
|
|
|
|
|
|
|
32 |
# Set up headers for API calls
|
33 |
self.headers = {
|
34 |
"Authorization": f"Bearer {HF_TOKEN}",
|
|
|
48 |
print("Stategraph created")
|
49 |
# Add nodes
|
50 |
workflow.add_node("analyze", self._analyze_question)
|
|
|
51 |
workflow.add_node("search", self._use_search)
|
52 |
workflow.add_node("final_answer", self._generate_final_answer)
|
53 |
|
54 |
# Define edges
|
|
|
55 |
workflow.add_edge("analyze", "search")
|
56 |
workflow.add_edge("analyze", "final_answer")
|
|
|
57 |
workflow.add_edge("search", "final_answer")
|
58 |
|
59 |
# Define conditional edges
|
60 |
def router(state: AgentState) -> str:
|
61 |
+
if state["current_step"] == 'search':
|
|
|
|
|
62 |
return 'search'
|
63 |
elif state["current_step"] == 'final_answer':
|
64 |
return 'final_answer'
|
|
|
68 |
"analyze",
|
69 |
router,
|
70 |
{
|
|
|
71 |
"search": "search",
|
72 |
"final_answer": "final_answer"
|
73 |
}
|
|
|
98 |
prompt = f"""Analyze this question and determine what needs to be done: {state['question']}
|
99 |
Return your analysis in this format:
|
100 |
{{
|
|
|
101 |
"needs_search": true/false,
|
102 |
+
"search_query": "query if needed"
|
|
|
|
|
|
|
|
|
103 |
}}
|
104 |
"""
|
105 |
|
|
|
107 |
state["needs_more_info"] = analysis.get('needs_search', False)
|
108 |
state["search_query"] = analysis.get('search_query', '')
|
109 |
|
110 |
+
if analysis.get('needs_search', False):
|
|
|
|
|
|
|
111 |
state["current_step"] = 'search'
|
112 |
else:
|
113 |
state["current_step"] = 'final_answer'
|
114 |
|
115 |
return state
|
116 |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
117 |
def _use_search(self, state: AgentState) -> AgentState:
|
118 |
"""Use the search tool."""
|
119 |
try:
|