File size: 11,807 Bytes
3709c1b abf5292 3709c1b |
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 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 |
import json
import requests
BASE_URL = "https://dashscope.aliyuncs.com/compatible-mode/v1"
MODEL = "qwen3-max"
TOKEN = "sk-ef26097310ec45c184e8d84b31ea9356"
def call_llm(prompt: str):
"""模拟调用大模型的函数（你可以替换为真实请求）"""
import requests
headers = {
"Authorization": f"Bearer {TOKEN}",
"Content-Type": "application/json"
}
data = {
"model": MODEL,
"messages": [{"role": "user", "content": prompt}],
"stream": True # 启用流式响应
}
response = requests.post(f"{BASE_URL}/chat/completions", headers=headers, json=data, stream=True)
for line in response.iter_lines():
if line:
yield line.decode('utf-8')
def call_llm_non_streaming(prompt: str):
"""调用大模型的函数（非流式）"""
import requests
headers = {
"Authorization": f"Bearer {TOKEN}",
"Content-Type": "application/json"
}
data = {
"model": MODEL,
"messages": [{"role": "user", "content": prompt}],
"stream": False # 禁用流式响应
}
response = requests.post(f"{BASE_URL}/chat/completions", headers=headers, json=data)
if response.status_code == 200:
result = response.json()
return result['choices'][0]['message']['content']
else:
return "Error: Failed to get response from LLM"
def get_invest_suggest(
user_input: str
# code: str,
# fm: str,
# news: str,
# stock: str
):
if not user_input.strip():
return ['你没有输入任何内容']
# 对应股票代码${code}，
# 结合财报数据${fm}、
# 最新新闻${news}、
# 当前股票信息${stock}，
prompt = f"""
你是一个专业的投资建议助手，你会对公司${user_input}，
给出以下投资建议。
投资建议分析：
- 是否建议投资
- 买入卖出价格
- 止损止盈点
- 持有时间（短期1月内、长期1月以上）
要求用语专业、言简意赅，给出具体的建议。：
"""
# 获取非流式响应
response = call_llm_non_streaming(prompt)
# print(f"投资建议：{response}{user_input}")
return response
def get_analysis_report(user_input: str):
if not user_input.strip():
return ['你没有输入任何内容']
prompt = f"""
# 请给出针对以下关键词的分析报告，条理清晰，要纯英文版的：
# {user_input}
# """
# 获取非流式响应
response = call_llm_non_streaming(prompt)
# print(f"投资建议：{response}{user_input}")
return response
def search_company(user_input: str):
if not user_input.strip():
return ['你没有输入任何内容']
# 获取非流式响应
response = chat_with_bailian_non_streaming(user_input, "6f24ece2689c4716b53b78aa7512f19f")
print(f"查询结果：{response}{user_input}")
try:
parsed_data = json.loads(response)
# 检查parsed_data是否为列表
if isinstance(parsed_data, list):
# 转换为Radio组件需要的choices格式
choices = [(f"{item['NAME']} ({item['CODE']})", f"{item['NAME']} ({item['CODE']})") for item in parsed_data]
else:
# 如果不是列表，可能是错误信息或其他格式
choices = ["未知错误"]
return choices
except json.JSONDecodeError:
return ["未知错误"]
def search_news(user_input: str):
if not user_input.strip():
return []
# 获取非流式响应
response = chat_with_bailian_non_streaming(user_input, "f7be17d71e704607b6558cff62c94954")
print(f"查询结果：{response}{user_input}")
try:
parsed_data = json.loads(response)
# 检查parsed_data是否为列表
if isinstance(parsed_data, list):
# 转换为Radio组件需要的choices格式
print(parsed_data)
return parsed_data
# choices = [(f"{item['NAME']} ({item['CODE']})", f"{item['NAME']} ({item['CODE']})") for item in parsed_data]
else:
# 如果不是列表，可能是错误信息或其他格式
choices = []
return choices
except json.JSONDecodeError:
return []
def get_stock_price_from_bailian(user_input: str):
if not user_input.strip():
return {}
# 获取非流式响应
response = chat_with_bailian_non_streaming(user_input, "d2b919f9b32e4fa28a75234cbb78a787")
print(f"查询结果：{response}{user_input}")
try:
parsed_data = json.loads(response)
return parsed_data if isinstance(parsed_data, dict) else {}
except json.JSONDecodeError:
return {}
# def search_company(user_input: str):
# if not user_input.strip():
# return [] # 返回空列表，而不是 yield
# result_json = [
# {"NAME": "公司名", "CODE": "股票代码"}
# ]
# # prompt = f"""
# # # 按照关键词搜索公司和其股票代码
# # 严格要求：
# # - 1. 按照我给的关键词“{user_input}”进行搜索，查询跟这个关键词“{user_input}”相关的公司的名称和股票代码；
# # - 2. 查询结果以“{json.dumps(result_json)}”的格式输出给我，不要掺杂其他内容，就纯“{json.dumps(result_json)}”格式的内容。
# # - 3. 数组内可以是很多条，只要是涉及到这个关键词“{user_input}”的公司都行。
# # - 4. 有公司名，没有股票代码
# # - 4. 如果搜不到相关的公司就返回空数组，不准直接返回“{json.dumps(result_json)}”。
# # 请严格遵守
# # """
# format = { "NAME": "公司名字", "CODE": "股票代码" }
# prompt = f"""
# # 角色
# 你是一位专业的股票信息提取助手，专门从用户的输入信息中提取出可能想选择的公司的股票名称和股票代码。对于美股，使用英文名称；对于港股和A股，使用中文名称。如果无法从用户输入中提取到有效的公司信息，则返回“无效输入，请输入公司”。
# ## 技能
# ### 技能 1: 提取公司股票名称
# - **任务**：从用户的输入信息中准确提取一个或多个疑似公司名称及股票代码。
# - 对于美股，使用英文名称+股票代码。
# - 对于港股和A股，使用中文名称+股票代码。
# - 如果有多个疑似的公司名称，则以 `{format}` 的JSON格式返回。
# ### 技能 2: 处理无效输入
# - **任务**：如果用户提供的输入信息中没有明确的公司名称，则返回“无效输入，请输入公司”。
# ## 限制
# - 仅处理与公司股票名称相关的输入信息。
# - 返回格式必须严格遵循 `{ format }` 或单个公司名称的形式。
# - 不提供任何其他类型的信息或建议，只专注于提取公司名称。
# - 如果无法从输入中提取到有效的公司信息，必须返回“无效输入，请输入公司”。
# - 未上市的公司不要返回给我
# - 没有股票代码的公司不要返回给我
# ## 示例
# - 输入: "我想买阿里巴巴的股票"
# - 输出:
# [
# { format },
# { format }
# ]
# - 输入: "最近股市行情怎么样？"
# - 输出: Invalid input, please enter a company
# 通过这些技能和限制，确保你能高效、准确地完成任务。
# 下面，来执行你的任务吧，用户用于搜索的关键词是：{user_input}
# """
# print(f"搜索关键词：{prompt}")
# bot_response = ""
# for chunk in call_llm(prompt): # 假设 call_llm 是同步生成器
# if not chunk or chunk == "[DONE]":
# continue
# if chunk.startswith("data: "):
# chunk = chunk[6:]
# try:
# response_data = json.loads(chunk)
# delta = response_data.get("choices", [{}])[0].get("delta", {})
# content = delta.get("content", "")
# if content:
# bot_response += content
# except json.JSONDecodeError:
# continue
# # 尝试解析最终的 bot_response 为 JSON
# print(f"查询结果：{bot_response}")
# try:
# companies = json.loads(bot_response)
# # 转换为 ["Alibaba (BABA)", "阿里巴巴-W (09988)"] 格式
# choices = [f"{item['NAME']} ({item['CODE']})" for item in companies]
# # if not choices:
# # return ["No results found"]
# return choices
# except (json.JSONDecodeError, KeyError, TypeError):
# # 解析失败，返回空或默认
# # return ["No results found"]
# return []
def chat_with_bailian_non_streaming(message, app_id):
# APP_ID = "6f24ece2689c4716b53b78aa7512f19f"
API_KEY = "sk-9bac83f8dccf4978a6bb1f8330a5404f"
# 使用自定义应用的API端点
API_URL = f"https://dashscope.aliyuncs.com/api/v1/apps/{app_id}/completion"
"""
与百炼自定义应用对话的函数（非流式）
"""
# 构建请求头
headers = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
# 不启用SSE以获得完整响应
}
# 构建历史对话记录
messages = []
# 添加当前问题
messages.append({"role": "user", "content": message})
# 构建请求体 - 根据自定义应用的API格式
data = {
"input": {
"messages": messages
},
"parameters": {
"max_tokens": 2048,
"temperature": 0.8,
"top_p": 0.9
}
}
try:
# 发送POST请求到百炼自定义应用API（非流式）
response = requests.post(API_URL, headers=headers, data=json.dumps(data))
# 检查响应状态码
if response.status_code == 200:
result = response.json()
if "output" in result and "text" in result["output"]:
return result["output"]["text"]
else:
return f"API响应格式不正确: {str(result)}"
else:
return f"API调用失败，状态码：{response.status_code}，响应：{response.text}"
except requests.RequestException as e:
return f"网络请求错误：{str(e)}"
except Exception as e:
return f"发生未知错误：{str(e)}"
def chat_bot(user_input: str, history: list):
if not user_input.strip():
yield "", history
return
# 添加用户消息到历史
history.append({"role": "user", "content": user_input})
# 调用AI模型进行总结
bot_response = ""
history.append({"role": "assistant", "content": bot_response})
for chunk in call_llm(user_input):
if not chunk or chunk == "[DONE]":
continue
if chunk.startswith("data: "):
chunk = chunk[6:]
try:
response_data = json.loads(chunk)
delta = response_data.get("choices", [{}])[0].get("delta", {})
content = delta.get("content", "")
if content:
bot_response += content
# 更新累积历史记录中的最后一条消息，实现流式显示
history[-1]["content"] = bot_response
yield "", history
except json.JSONDecodeError:
continue # 忽略无效 JSON
# # 最终完整历史
# history.append({"role": "assistant", "content": bot_response})
# yield "", history
|