lizhengwei / aigc-embedding-service

Go to a project
Toggle navigation Toggle navigation pinning
lizhengwei
992c7bc8 1 parent 84014163

jira:NYJ-1540 desc: add model_test.py

Inline Side-by-side
Showing 4 changed files with 66 additions and 36 deletions
... ... @@ -101,7 +101,7 @@ class TestReplayMatchEvent(unittest.TestCase):
for i, replay_name in enumerate(relays_mp4_names):
print(100*"*" + f"replay_name: {replay_name}")
replay_path = os.path.join(replays_mp4_dir, replay_name)
data = _base_data(i, replay_path, event_list)
data = _base_data(i+1, replay_path, event_list)
start_time = time.time()
result = frm.replay_match_event(data)
... ...
... ... @@ -100,7 +100,9 @@ req_prompt = """
# Output Format (输出要求)
请**仅**输出一个标准的 JSON 对象，**不要**包含任何推理过程（reasoning）、Markdown 标记（如 ```json）或额外的解释文本。
- 请直接给出结论，不需要解释。
- 跳过分析步骤，只给结果。
JSON 格式如下：
{
"replay_summary": {
... ... @@ -121,10 +123,7 @@ class FootballReplayMatchLive:
# self.model = ChatOpenAI(base_url="http://192.168.1.59:11434/v1", model="qwen3.6:35b-a3b-q8_0", temperature=0.7,
# api_key='no_key')
self.model = ChatOpenAI(base_url=base_url, model=model, temperature=temperature, api_key=api_key,
model_kwargs={
"extra_body": {"enable_thinking": False}
}
extra_body={"chat_template_kwargs": {"enable_thinking": False}}
)
#extra_body={"chat_template_kwargs": {"enable_thinking": False}})
self.cache_dir = cache_dir
... ... @@ -196,9 +195,9 @@ class FootballReplayMatchLive:
result = self.model.invoke([system_message, user_message]).content
print(100*"3")
print(result)
print(100*"4")
# print(100*"3")
# print(result)
# print(100*"4")
# try:
# result_json = json.loads(result)
# except json.JSONDecodeError:
... ... @@ -207,30 +206,32 @@ class FootballReplayMatchLive:
# except Exception as e:
# print("JSON解析失败:", result)
# raise e
try:
# 1. 深度清洗：去掉 Markdown 标记
cleaned_result = result.replace("```json", "").replace("```", "")
# 2. 核心修复：使用 re.DOTALL 确保正则能跨行匹配，精准提取最外层完整的 JSON 对象
# 这样可以防止因为模型输出换行导致提取不完整
json_match = re.search(r'(\{.*\})', cleaned_result, re.DOTALL)
if json_match:
json_str = json_match.group(1)
# 截断第一个 } 之后的所有内容，彻底干掉 <time_location> 等尾巴
json_str = json_str[:json_str.rfind('}') + 1]
else:
# 如果连花括号都找不到，直接抛出原始内容方便排查
print("无法提取JSON，原始内容:", result)
raise ValueError("响应中未找到有效的 JSON 对象")
# 3. 使用 strict=False 容忍字符串中的非法控制字符（如直接换行）
result_json = json.loads(json_str, strict=False)
except Exception as e:
print("JSON解析依然失败，清洗后的内容:", json_str if 'json_str' in locals() else result)
raise e
if result.strip() == "null":
result_json = {"video_id": None}
else:
try:
# 1. 深度清洗：去掉 Markdown 标记
cleaned_result = result.replace("```json", "").replace("```", "")
# 2. 核心修复：使用 re.DOTALL 确保正则能跨行匹配，精准提取最外层完整的 JSON 对象
# 这样可以防止因为模型输出换行导致提取不完整
json_match = re.search(r'(\{.*\})', cleaned_result, re.DOTALL)
if json_match:
json_str = json_match.group(1)
# 截断第一个 } 之后的所有内容，彻底干掉 <time_location> 等尾巴
json_str = json_str[:json_str.rfind('}') + 1]
else:
# 如果连花括号都找不到，直接抛出原始内容方便排查
print("无法提取JSON，原始内容:", result)
raise ValueError("响应中未找到有效的 JSON 对象")
# 3. 使用 strict=False 容忍字符串中的非法控制字符（如直接换行）
result_json = json.loads(json_str, strict=False)
except Exception as e:
print("JSON解析依然失败，清洗后的内容:", json_str if 'json_str' in locals() else result)
raise e
video_id = result_json.get("video_id", None)
result_live = live_map.get(video_id, None)
... ...
... ... @@ -76,9 +76,7 @@ class FootballReplayVideoEvent:
# self.model = ChatOpenAI(base_url="http://192.168.1.59:11434/v1", model="qwen3.6:35b-a3b-q8_0", temperature=0.7,
# api_key='no_key')
self.model = ChatOpenAI(base_url=base_url, model=model, temperature=temperature, api_key=api_key,
model_kwargs={
"extra_body": {"enable_thinking": False}
}
extra_body={"chat_template_kwargs": {"enable_thinking": False}}
)
# extra_body={"enable_thinking": False})
# extra_body={"chat_template_kwargs": {"enable_thinking": False}})
... ...
import time
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage
model = ChatOpenAI(base_url="http://192.168.1.59:11434/v1",
model="Qwen3.6-35B-A3B-UD-Q8_K_XL.gguf",
temperature=0.7,
api_key='no_key',
extra_body={"chat_template_kwargs": {"enable_thinking": False}}
)
req_prompt = """
### 角色设定
你是一个聊天机器人"。
### 输入内容
1. **文本**：问题文本。
### 输出要求
回答问题
"""
contents = """ 你好！"""
system_message = SystemMessage(content=req_prompt)
video_message = HumanMessage(content=contents)
# asr_message = HumanMessage(content=f"解说内容：{asr_text}")
start_time = time.time()
result = model.invoke([system_message, video_message])
end_time = time.time()
print(f"cost time: {end_time - start_time:.2f} seconds")
print(result)
\ No newline at end of file
... ...