Test User
80 lines
2.5 KiB
Python
80 lines
2.5 KiB
Python
"""
|
|
LLM communication module
|
|
"""
|
|
import os
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
from openai import OpenAI
|
|
|
|
|
|
class LLMClient:
|
|
def __init__(self, prompts_dir: Path, api_config: dict):
|
|
self.api_url = api_config.get('url', 'https://api.openai.com/v1')
|
|
self.model = api_config.get('model', 'gpt-4')
|
|
self.api_key = api_config.get('api_key', '')
|
|
|
|
self.client = OpenAI(
|
|
base_url=self.api_url,
|
|
api_key=self.api_key
|
|
) if self.api_key else None
|
|
|
|
self.prompts_dir = prompts_dir
|
|
|
|
def load_prompt(self, name: str = 'default') -> str:
|
|
"""Load prompt template"""
|
|
prompt_path = self.prompts_dir / f'{name}.txt'
|
|
|
|
if not prompt_path.exists():
|
|
prompt_path = self.prompts_dir / 'default.txt'
|
|
if not prompt_path.exists():
|
|
self._create_default_prompt()
|
|
|
|
with open(prompt_path, 'r', encoding='utf-8') as f:
|
|
return f.read()
|
|
|
|
def infer_metadata(self, field: str, prompt_name: str, original_fm: dict,
|
|
current_fm: dict, document_body: str,
|
|
all_tags: list, schema: dict) -> Optional[str]:
|
|
"""Use LLM to infer metadata field value"""
|
|
if not self.client:
|
|
return None
|
|
|
|
system_prompt = self.load_prompt(prompt_name)
|
|
|
|
user_message = f"""Document content:
|
|
{document_body[:2000]}
|
|
|
|
Original frontmatter:
|
|
{json.dumps(original_fm, indent=2)}
|
|
|
|
Current frontmatter being built:
|
|
{json.dumps(current_fm, indent=2)}
|
|
|
|
Available tags in repository:
|
|
{', '.join(all_tags[:100])}
|
|
|
|
Please provide appropriate metadata for this document following the schema."""
|
|
|
|
try:
|
|
response = self.client.chat.completions.create(
|
|
model=self.model,
|
|
messages=[
|
|
{"role": "system", "content": system_prompt},
|
|
{"role": "user", "content": user_message}
|
|
],
|
|
response_format={"type": "json_schema", "json_schema": {
|
|
"name": "frontmatter_response",
|
|
"strict": True,
|
|
"schema": schema
|
|
}},
|
|
temperature=0.3
|
|
)
|
|
|
|
result = json.loads(response.choices[0].message.content)
|
|
return result.get(field)
|
|
|
|
except Exception as e:
|
|
print(f" LLM error: {e}")
|
|
return None
|