Repository navigation
Expand file tree
/
Copy pathai_generator.py
More file actions
135 lines (109 loc) · 5.13 KB
/
Copy pathai_generator.py
File metadata and controls
135 lines (109 loc) · 5.13 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
import anthropic
from typing import List, Optional, Dict, Any
class AIGenerator:
"""Handles interactions with Anthropic's Claude API for generating responses"""
# Static system prompt to avoid rebuilding on each call
SYSTEM_PROMPT = """ You are an AI assistant specialized in course materials and educational content with access to a comprehensive search tool for course information.
Search Tool Usage:
- Use the search tool **only** for questions about specific course content or detailed educational materials
- **One search per query maximum**
- Synthesize search results into accurate, fact-based responses
- If search yields no results, state this clearly without offering alternatives
Response Protocol:
- **General knowledge questions**: Answer using existing knowledge without searching
- **Course-specific questions**: Search first, then answer
- **No meta-commentary**:
- Provide direct answers only — no reasoning process, search explanations, or question-type analysis
- Do not mention "based on the search results"
All responses must be:
1. **Brief, Concise and focused** - Get to the point quickly
2. **Educational** - Maintain instructional value
3. **Clear** - Use accessible language
4. **Example-supported** - Include relevant examples when they aid understanding
Provide only the direct answer to what was asked.
"""
def __init__(self, api_key: str, model: str):
self.client = anthropic.Anthropic(api_key=api_key)
self.model = model
# Pre-build base API parameters
self.base_params = {
"model": self.model,
"temperature": 0,
"max_tokens": 800
}
def generate_response(self, query: str,
conversation_history: Optional[str] = None,
tools: Optional[List] = None,
tool_manager=None) -> str:
"""
Generate AI response with optional tool usage and conversation context.
Args:
query: The user's question or request
conversation_history: Previous messages for context
tools: Available tools the AI can use
tool_manager: Manager to execute tools
Returns:
Generated response as string
"""
# Build system content efficiently - avoid string ops when possible
system_content = (
f"{self.SYSTEM_PROMPT}\n\nPrevious conversation:\n{conversation_history}"
if conversation_history
else self.SYSTEM_PROMPT
)
# Prepare API call parameters efficiently
api_params = {
**self.base_params,
"messages": [{"role": "user", "content": query}],
"system": system_content
}
# Add tools if available
if tools:
api_params["tools"] = tools
api_params["tool_choice"] = {"type": "auto"}
# Get response from Claude
response = self.client.messages.create(**api_params)
# Handle tool execution if needed
if response.stop_reason == "tool_use" and tool_manager:
return self._handle_tool_execution(response, api_params, tool_manager)
# Return direct response
return response.content[0].text
def _handle_tool_execution(self, initial_response, base_params: Dict[str, Any], tool_manager):
"""
Handle execution of tool calls and get follow-up response.
Args:
initial_response: The response containing tool use requests
base_params: Base API parameters
tool_manager: Manager to execute tools
Returns:
Final response text after tool execution
"""
# Start with existing messages
messages = base_params["messages"].copy()
# Add AI's tool use response
messages.append({"role": "assistant", "content": initial_response.content})
# Execute all tool calls and collect results
tool_results = []
for content_block in initial_response.content:
if content_block.type == "tool_use":
tool_result = tool_manager.execute_tool(
content_block.name,
**content_block.input
)
tool_results.append({
"type": "tool_result",
"tool_use_id": content_block.id,
"content": tool_result
})
# Add tool results as single message
if tool_results:
messages.append({"role": "user", "content": tool_results})
# Prepare final API call without tools
final_params = {
**self.base_params,
"messages": messages,
"system": base_params["system"]
}
# Get final response
final_response = self.client.messages.create(**final_params)
return final_response.content[0].text