Spaces:
Sleeping
Sleeping
File size: 5,829 Bytes
cd63cf6 4b3bda3 cd63cf6 103de27 cd63cf6 103de27 cd63cf6 103de27 cd63cf6 4b3bda3 cd63cf6 103de27 cd63cf6 103de27 cd63cf6 103de27 cd63cf6 103de27 cd63cf6 |
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 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 |
import json
import os
from logging import getLogger
from openai import OpenAI
from openai import AzureOpenAI
from openai import OpenAIError
import fitz
import requests
import time
from langchain.document_loaders import PyPDFLoader
from langchain.prompts import ChatPromptTemplate
from langchain.text_splitter import ( # CharacterTextSplitter,
RecursiveCharacterTextSplitter,
)
from langchain.vectorstores import Chroma
from langchain_core.output_parsers import StrOutputParser
import json
# from langchain_openai import ChatOpenAI, OpenAIEmbeddings
# from langchain_openai import AzureChatOpenAI, AzureOpenAIEmbeddings
import csv
from bs4 import BeautifulSoup
from src.myLogger import set_logger
logger = set_logger("my_app", level="INFO")
system_prompt = """
あなたのタスクは、提供された資料が指定された会社のものであるかどうかを判断することです。以下の基準に基づいて判断を行ってください：
1. 資料に会社名が明記されているかどうかを確認してください。
2. 資料の内容の大部分が指定された会社に関する情報であるかどうかを確認してください。
これらの基準を満たしている場合、その資料は会社のものであると見なされます。出力フォーマットに従って結果を出力してください。
"""
json_schema = {
"type": "object",
"properties": {
"judge": {
"type": "integer",
"description": "{company_name}の資料であれば1,そうでなければ0"
},
"reason": {
"type": "string",
"description": "どうしてそう判断したのか"
}
},
"required": ["judge", "reason"]
}
def generate_check_(reference):
api_key = os.getenv("OPENAI_API_KEY")
client = OpenAI(
api_key=api_key,
)
retries = 0
max_retries = 100
delay = 5
while retries < max_retries:
try:
response = client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[
{
"role": "system",
"content": system_prompt,
},
{
"role": "user",
"content": reference,
},
],
functions=[{"name": "generate_queries", "parameters": json_schema}],
function_call={"name": "generate_queries"},
temperature=0.0,
top_p=0.0,
)
output = response.choices[0].message.function_call.arguments
time.sleep(1)
return output
except OpenAIError as e:
print(f"Error occurred: {e}. Retrying in {delay} seconds...")
retries += 1
time.sleep(delay)
except Exception as e:
print(f"Unexpected error: {e}. Retrying in {delay} seconds...")
retries += 1
time.sleep(delay)
raise RuntimeError("Maximum retries exceeded. Could not get a valid response.")
def ch(company_name, reference):
com = company_name.replace("株式会社", "")
user_prompt = f"""
以下の資料は，{company_name}のものであるか，理由をつけて答えてください。資料が{company_name}のものであるとは，
資料に{company_name}と明記されておりまた資料の内容の大部分が{company_name}に関する情報であるということです。
以下のフォーマットに従い出力してください。
[資料]
{reference}
[出力フォーマット]
{{
judge: 0 or 1({com}の資料であれば1,そうでなければ0)
reason: "どうしてそう判断したのか具体的に説明してください。"
}}
"""
ret = generate_check_(user_prompt)
js = json.loads(ret)
return js["judge"], js["reason"]
def pdf_to_text(url):
# リクエストをきちんと送るためのもの
headers = {
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_11_5) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/50.0.2661.102 Safari/537.36"
}
result = requests.get(url, headers=headers, timeout=5)
finename = (
# "./out/" + company_name + url.replace("/", "_").replace(":", "_") + "IR.pdf"
# url.replace("/", "_").replace(":", "+") + "IR.pdf"
"./out/tmp.pdf"
)
try:
with open(finename, "wb") as file:
for chunk in result.iter_content(1000000000):
file.write(chunk)
except Exception as e:
print(e)
document = fitz.open(finename)
full_text = ""
for page_num in range(len(document)):
page = document.load_page(page_num)
full_text += page.get_text()
document.close()
return full_text
def html_to_text(url):
# リクエストをきちんと送るためのもの
headers = {
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_11_5) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/50.0.2661.102 Safari/537.36"
}
logger.info(f"requesting {url}")
result = requests.get(url, headers=headers, timeout=5)
result.encoding = result.apparent_encoding
soup = BeautifulSoup(result.text, "html.parser")
paragraphs = soup.find_all("p")
return "\n".join([paragraph.get_text() for paragraph in paragraphs])
def co(company_name, url):
try:
if url.endswith(".pdf"):
full_text = pdf_to_text(url)
else:
full_text = html_to_text(url)
except Exception as e:
print(e)
return -1
reference = full_text[:10000]
judge, reason = ch(company_name, reference)
logger.info(f'Judge: {judge}, Reason: {reason}')
return judge, reason
|