今天我們來做更多應用,前面第一個實作以,文字生成應用為主,
今天我們來玩點稍微進階的主題,『對話式聊天機器人』,
一樣我們快速建立專案,可以參考 day3 怎麼建置專案的
BedrockChatBot
這裡我們要紀錄對話紀錄 conversation_history
class BedrockChatBot:
"""
一樣初始化 boto3
"""
def __init__(self, model_id="anthropic.claude-v2"):
self.bedrock_runtime = boto3.client(
service_name='bedrock-runtime',
region_name='ap-northeast-1'
)
self.model_id = model_id
self.conversation_history = []
我們這裡要記錄對話紀錄,所以當然要實作新增對話紀錄的 method add_to_history
def add_to_history(self, role, content):
"""添加對話到歷史記錄"""
self.conversation_history.append({
'role': role,
'content': content
})
def chat(self, user_input):
"""進行對話"""
try:
prompt = self.build_prompt(user_input)
body = {
"prompt": prompt,
"max_tokens_to_sample": 1000,
"temperature": 0.7,
"top_p": 0.9
}
response = self.bedrock_runtime.invoke_model(
body=json.dumps(body),
modelId=self.model_id,
accept='application/json',
contentType='application/json'
)
response_body = json.loads(response.get('body').read())
assistant_response = response_body.get('completion', '').strip()
# 更新對話歷史
self.add_to_history('user', user_input)
self.add_to_history('assistant', assistant_response)
return assistant_response
except Exception as e:
return f"抱歉,發生錯誤:{str(e)}"
def interactive_chat():
bot = BedrockChatBot()
print("開始對話囉!好誒~~~~。(輸入 quit 結束對話)\n")
while True:
user_input = input("您: ")
if user_input.lower() == 'quit':
print("再見!")
break
response = bot.chat(user_input)
print(f"你的未來老婆: {response}\n")
class BedrockChatBot:
def __init__(self, model_id='anthropic.claude-v2'):
self.bedrock_runtime = boto3.client(
service_name='bedrock-runtime',
region_name='us-east-1'
)
self.model_id = model_id
self.conversation_history = []
def add_to_history(self, role, content):
"""添加對話到歷史記錄"""
self.conversation_history.append({
'role': role,
'content': content
})
def build_prompt(self, user_input):
"""構建包含歷史對話的提示"""
prompt = ""
for msg in self.conversation_history[-5:]: # 保留最近5輪對話
if msg['role'] == 'user':
prompt += f"\n\nHuman: {msg['content']}"
else:
prompt += f"\n\nAssistant: {msg['content']}"
prompt += f"\n\nHuman: {user_input}\n\nAssistant:"
return prompt
def chat(self, user_input):
"""進行對話"""
try:
prompt = self.build_prompt(user_input)
body = {
"prompt": prompt,
"max_tokens_to_sample": 1000,
"temperature": 0.7,
"top_p": 0.9
}
response = self.bedrock_runtime.invoke_model(
body=json.dumps(body),
modelId=self.model_id,
accept='application/json',
contentType='application/json'
)
response_body = json.loads(response.get('body').read())
assistant_response = response_body.get('completion', '').strip()
# 更新對話歷史
self.add_to_history('user', user_input)
self.add_to_history('assistant', assistant_response)
return assistant_response
except Exception as e:
return f"抱歉,發生錯誤:{str(e)}"
def interactive_chat():
bot = BedrockChatBot()
print("開始對話囉!好誒~~~~。(輸入 quit 結束對話)\n")
while True:
user_input = input("您: ")
if user_input.lower() == 'quit':
print("再見!")
break
response = bot.chat(user_input)
print(f"你的未來老婆: {response}\n")
if __name__ == "__main__":
interactive_chat()
這樣我們可以跟未來的老婆對話了!好誒~~~