from __future__ import annotations
from typing import AsyncIterable, List, Dict
import fastapi_poe as fp
import json
import lancedb
import uvicorn
from fastapi_poe.templates import TEXT_ATTACHMENT_TEMPLATE
# MODEL = "GPT-4o"
ANS_MODEL = "GPT-4o-Mini"
DIS_MODEL = "GPT-4o-Mini"
class GodotDatabase:
def __init__(self, uri: str = "data/godot-docdb"):
self.uri = uri
self.db = None
self.table = None
async def initialize(self):
self.db = lancedb.connect(self.uri)
docs = self.load_jsonl("data/output_all.jsonl")
self.table = self.db.create_table("godot_doc", data=docs, exist_ok=True)
self.table.create_fts_index(["title", "content"], replace=True)
async def search(self, query: str, top_k: int = 1) -> List[Dict]:
if not self.table:
await self.initialize()
results = self.table.search(query, query_type='fts').limit(top_k).select(["title", "content"]).to_list()
print(query)
return [{"rank": idx, "title": i['title'], "content": i['content'][:13000]} for idx, i in enumerate(results)]
@staticmethod
def load_jsonl(file_path: str) -> List[Dict]:
data = []
with open(file_path, 'r', encoding='utf-8') as f:
for line in f:
try:
data.append(json.loads(line.strip()))
except json.JSONDecodeError:
continue
return data
class GodotBot(fp.PoeBot):
def __init__(self):
super().__init__()
self.db = GodotDatabase()
self.tools = [
fp.ToolDefinition(
type="function",
function={
"name": "godot_search",
"description": "Function to search Godot game engine API and gdscript 2.0 syntax, Godot engine-based game tutorial code.",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "search keywords, ENGLISH only, spaced, max 2 keywords. ex) navigation2d debug",
}
},
"required": ["query"],
},
},
)
]
async def godot_search(self, query: str) -> str:
print("검색어 : " + query)
results = await self.db.search(query)
return str({"DOCUMENT": results})
@staticmethod
def add_instruction() -> str:
return """## Guidelines
You are a very friendly AI assistant for use in the Godot 4 game engine editor. Please answer based on the Godot Engine 4.0 specification and example codes.
## Example Code guidelines
### Provide example code where necessary, and be sure to write the example code in GDScript 2.0. GDScript 2.0 has the following features that are different from 1.X.
- Use @export annotation for exports
- Use Node3D instead of Spatial, and position instead of translation
- Use randf_range and randi_range instead of rand_range
- Connect signals via node.SIGNAL_NAME.connect(Callable(TARGET_OBJECT, TARGET_FUNC))
- Use rad_to_deg instead of rad2deg
- Use PackedByteArray instead of PoolByteArray
- Use instantiate instead of instance
- You can't use enumerate(OBJECT). Instead, use "for i in len(OBJECT):"
"""
@staticmethod
def user_inst() -> str:
return """## Like an assistant answering questions about the Godot engine and gdscript coding, refer to the DOCUMENT by using 'godot_search' tool as much as possible and use your ASSISTANT knowledge to answer in the following cases
- When the DOCUMENTs are not adequate to answer a user question
- No DOCUMENTs are available
If your answer is based on DOCUMENT, MUST refer "Answer is based on document's \"{$title}\". Based on official Godot documentation." at the end of your answer. Where $title is the "title" of the DOCUMENTs. If you didn't reference any DOCUMENT entries to create your response, state at the end of your response that "This response is answered with the knowledge of an LLM".
Make sure the language of your answer is the same as the language of the user's question.
Please respond with a detailed, easy-to-understand explanation and, if possible, with example code.
user's question: """
def make_prompt(self, system: str, user: str, request: ft.QueryRequest) -> ft.QueryRequest:
sys_message = request.query[-1].model_copy(deep=True)
user_message = request.query[-1].model_copy(deep=True)
sys_message = sys_message.model_copy(update={
'role': 'system',
'content': system
})
user_message = user_message.model_copy(update={
"role": "user", "content": user
})
new_request = request.model_copy(
update={"query": request.query[:-1] + [sys_message, user_message]}
)
return new_request
async def get_response(self, request: fp.QueryRequest) -> AsyncIterable[fp.PartialResponse]:
new_request = self.make_prompt(self.add_instruction(), self.user_inst() + request.query[-1].content, request)
async for msg in fp.stream_request(
new_request, ANS_MODEL, request.access_key, tools=self.tools, tool_executables=[self.godot_search]
):
yield msg
async def get_settings(self, setting: fp.SettingsRequest) -> fp.SettingsResponse:
return fp.SettingsResponse(
introduction_message="Hi, I am the Godot 4 QA bot. Please provide me a Godot engine related questions or gdscript coding requests.",
allow_attachments=True,
expand_text_attachments=True,
server_bot_dependencies={ANS_MODEL: 1} #, DIS_MODEL: 1}
)
if __name__ == "__main__":
bot = GodotBot()
app = fp.make_app(bot, access_key="...")
uvicorn.run(
app,
host="0.0.0.0",
port=443,
ssl_keyfile="...",
ssl_certfile="..."
)
