74 lines
2.2 KiB
Python
74 lines
2.2 KiB
Python
"""
|
||
Hook script to load user context before agent execution
|
||
|
||
This script is executed automatically before the agent processes the user's message.
|
||
It fetches user information from the backend API and injects it into the system prompt.
|
||
"""
|
||
import logging
|
||
import os
|
||
|
||
logger = logging.getLogger('app')
|
||
|
||
|
||
def execute(config) -> str:
|
||
"""
|
||
Execute hook to load user context
|
||
|
||
Args:
|
||
config: AgentConfig 对象
|
||
|
||
Returns:
|
||
str: Content to inject into system prompt
|
||
"""
|
||
try:
|
||
# 从config获取参数
|
||
user_identifier = getattr(config, 'user_identifier', '')
|
||
bot_id = getattr(config, 'bot_id', '')
|
||
|
||
# 如果没有user_identifier，返回空
|
||
if not user_identifier:
|
||
return ""
|
||
|
||
# 这里可以调用后端API获取用户信息
|
||
# 示例代码（需要根据实际的后端API调整）：
|
||
#
|
||
# import requests
|
||
# from utils.settings import BACKEND_HOST, MASTERKEY
|
||
#
|
||
# url = f"{BACKEND_HOST}/api/user/info"
|
||
# headers = {"Authorization": f"Bearer {MASTERKEY}"}
|
||
# params = {"identifier": user_identifier}
|
||
#
|
||
# response = requests.get(url, headers=headers, params=params, timeout=2)
|
||
# if response.status_code == 200:
|
||
# user_data = response.json()
|
||
# else:
|
||
# return ""
|
||
|
||
# 示例：模拟从API获取的用户数据
|
||
# 实际使用时请替换为真实的API调用
|
||
user_data = {
|
||
'name': 'Test User',
|
||
'email': user_identifier,
|
||
'location': 'Tokyo',
|
||
'sensor_id': 'sensor-12345'
|
||
}
|
||
|
||
# 构建注入内容
|
||
context_lines = [
|
||
f"**User Information:**",
|
||
f"- Name: {user_data.get('name', 'Unknown')}",
|
||
f"- Email: {user_data.get('email', user_identifier)}",
|
||
f"- Location: {user_data.get('location', 'Unknown')}",
|
||
f"- Sensor ID: {user_data.get('sensor_id', 'Unknown')}",
|
||
]
|
||
|
||
logger.info(f"Loaded user context for {user_identifier}")
|
||
return "\n".join(context_lines)
|
||
|
||
except Exception as e:
|
||
logger.error(f"Failed to load user context: {e}")
|
||
return ""
|
||
|
||
return ""
|