ly/langchain_projectagent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

增加WEB入口

Browse Source
This commit is contained in:
paituo
2025-07-07 09:32:05 +08:00
parent dc43b69164
commit cfb4f4ea6a
2 changed files with 234 additions and 127 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main_streamlt.py
+228 -122
View File
@@ -1,6 +1,6 @@
import os
import json
import streamlit as st
import json
import os
import uuid
import time
from typing import Dict, List, Optional, Any
@@ -11,19 +11,26 @@ from datetime import datetime
# 添加项目根目录到路径
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
# 确保必要的目录存在
CACHE_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "cache")
DATA_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "data")
os.makedirs(CACHE_DIR, exist_ok=True)
os.makedirs(DATA_DIR, exist_ok=True)
# 配置日志
current_file = os.path.splitext(os.path.basename(__file__))[0]
now_str = datetime.now().strftime("%Y%m%d%H%M%S")
log_filename = f"{current_file}_{now_str}.log"
# 确保日志目录存在
os.makedirs("logs", exist_ok=True)
# 确保日志目录存在并设置在logs目录下
log_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "logs")
os.makedirs(log_dir, exist_ok=True)
logging.basicConfig(
level=logging.DEBUG,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
handlers=[
logging.FileHandler(os.path.join("logs", log_filename), encoding="utf-8"),
logging.FileHandler(os.path.join(log_dir, log_filename), encoding="utf-8"),
logging.StreamHandler()
],
)
@@ -79,10 +86,11 @@ def load_dependencies():
# 后台数据结构
class Indicator:
def __init__(self, name: str, query: str = "", id: Optional[str] = None):
def __init__(self, name: str, query: str = "", code: str = "", id: Optional[str] = None):
self.id = id if id else str(uuid.uuid4())
self.name = name
self.query = query
self.code = code
self.created_at = time.time()
self.updated_at = time.time()
@@ -91,6 +99,7 @@ class Indicator:
"id": self.id,
"name": self.name,
"query": self.query,
"code": self.code,
"created_at": self.created_at,
"updated_at": self.updated_at
}
@@ -100,6 +109,7 @@ class Indicator:
indicator = cls(
name=data["name"],
query=data["query"],
code=data.get("code", ""), # 使用 get 方法处理可能不存在的 code 字段
id=data["id"]
)
indicator.created_at = data.get("created_at", time.time())
@@ -108,7 +118,9 @@ class Indicator:
class IndicatorManager:
def __init__(self, save_path: str = "indicator_library.json"):
def __init__(self, save_path: Optional[str] = None):
if save_path is None:
save_path = os.path.join(CACHE_DIR, "indicator_library.json")
self.indicators: Dict[str, Indicator] = {}
self.save_path = save_path
self.load_indicators()
@@ -143,7 +155,7 @@ class IndicatorManager:
self.save_indicators()
return indicator
def update_indicator(self, id: str, name: Optional[str] = None, query: Optional[str] = None) -> Optional[Indicator]:
def update_indicator(self, id: str, name: Optional[str] = None, query: Optional[str] = None, code: Optional[str] = None) -> Optional[Indicator]:
"""更新指标"""
if id not in self.indicators:
return None
@@ -156,6 +168,8 @@ class IndicatorManager:
indicator.name = name
if query is not None:
indicator.query = query
if code is not None:
indicator.code = code
indicator.updated_at = time.time()
self.save_indicators()
@@ -184,12 +198,6 @@ class IndicatorManager:
# 保存会话状态
def ensure_cache_dir():
"""确保cache目录存在"""
cache_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "cache")
os.makedirs(cache_dir, exist_ok=True)
return cache_dir
@st.cache_data(ttl=3600)
def save_session_state():
"""保存会话状态到文件"""
@@ -197,20 +205,19 @@ def save_session_state():
state = {
"current_indicator_id": st.session_state.get("current_indicator_id", None)
}
cache_dir = ensure_cache_dir()
state_file = os.path.join(cache_dir, "indicator_creator_state.json")
state_file = os.path.join(CACHE_DIR, "indicator_creator_state.json")
with open(state_file, "w", encoding="utf-8") as f:
json.dump(state, f, ensure_ascii=False, indent=2)
except Exception as e:
st.error(f"保存会话状态失败: {str(e)}")
# 加载会话状态
@st.cache_data(ttl=3600)
def load_session_state():
"""从文件加载会话状态"""
try:
cache_dir = ensure_cache_dir()
state_file = os.path.join(cache_dir, "indicator_creator_state.json")
state_file = os.path.join(CACHE_DIR, "indicator_creator_state.json")
if os.path.exists(state_file):
with open(state_file, "r", encoding="utf-8") as f:
state = json.load(f)
@@ -220,69 +227,76 @@ def load_session_state():
except Exception as e:
st.error(f"加载会话状态失败: {str(e)}")
# 测试指标查询
@st.cache_data(ttl=60) # 缓存结果1分钟
def test_indicator_query(query: str) -> str:
"""测试指标查询"""
def generate_indicator_code(query: str) -> str:
"""生成指标代码"""
if not query.strip():
return "查询语句为空，无法测试"
return "查询语句为空，无法生成代码"
# 模拟异步测试过程
result = "开始测试查询...\n"
result = ""
try:
# 步骤1：通过调用user_interaction.understand(query)实现
result += "步骤1: 正在理解查询语句...\n"
result = "正在理解查询语句..."
if "user_interaction" not in st.session_state:
st.session_state.user_interaction = st.session_state.dialog_manager.user_interaction
understanding_result = st.session_state.user_interaction.understand(query)
if not understanding_result:
result += "❌ 无法理解查询语句\n"
return result
return "❌ 无法理解查询语句"
# 步骤2：验证步骤一返回值
result += "步骤2: 正在验证理解结果...\n"
result = "正在验证理解结果..."
selected_knowledge = []
entity_info = []
for item in understanding_result:
result += f"- 识别到实体: {item.get('name')}, 约束条件: {item.get('constraints')}\n"
entity_info.append(f"识别到实体: {item.get('name')}, 约束条件: {item.get('constraints')}")
selected_knowledge.append(item)
if not selected_knowledge:
result += "❌ 没有找到相关知识\n"
return result
return "❌ 没有找到相关知识"
# 显示识别到的实体信息
result = "\n".join(entity_info)
# 步骤3：dialog_manager.generated_code(query, selected_knowledge) 实现
result += "步骤3: 正在生成查询代码...\n"
result = "正在生成指标代码..."
if "dialog_manager" not in st.session_state:
result += "❌ 对话管理器未初始化\n"
return result
return "❌ 对话管理器未初始化"
generated_code = st.session_state.dialog_manager.generated_code(query, selected_knowledge)
if not generated_code:
result += "❌ 代码生成失败\n"
return result
result += "✅ 代码生成成功\n"
generated_result = st.session_state.dialog_manager.generated_code(query, selected_knowledge)
# 步骤4：通过调用code_executor.execute_code 实现
result += "步骤4: 正在执行查询代码...\n"
if "code_executor" not in st.session_state:
st.session_state.code_executor = st.session_state.dialog_manager.code_executor
# 处理返回结构
if not generated_result:
return "❌ 代码生成失败"
execution_result = st.session_state.code_executor.execute_code(generated_code)
if execution_result and "error" not in execution_result.lower():
result += "✅ 查询执行成功!\n"
result += f"查询结果:\n{execution_result}"
# 检查返回的是否是字典结构
if isinstance(generated_result, dict):
if not generated_result.get('status', False):
return f"❌ 代码生成失败: {generated_result.get('message', '未知错误')}"
# 提取代码内容
code = generated_result.get('data', '')
if not code:
return "❌ 生成的代码为空"
# 将生成的代码保存到session_state中
st.session_state.generated_code = code
else:
result += f"❌ 查询执行失败: {execution_result}\n"
# 如果直接返回的是代码字符串
st.session_state.generated_code = generated_result
if "current_indicator_id" in st.session_state:
st.session_state.current_query_id = st.session_state.current_indicator_id
return "✅ 指标代码生成成功，可以在下方查看并执行"
except Exception as e:
import traceback
error_details = traceback.format_exc()
result += f"❌ 查询测试失败: {str(e)}\n"
result += f"错误详情: {error_details}"
return f"❌ 指标代码生成失败: {str(e)}"
return result
@@ -338,7 +352,7 @@ def init_app_data():
# Streamlit 应用界面
def main():
st.set_page_config(page_title="指标创建器", layout="wide")
st.set_page_config(page_title="造价工程指标管理器", layout="wide")
# 使用进度条显示加载过程
with st.spinner("正在初始化应用..."):
@@ -361,8 +375,7 @@ def main():
# 侧边栏
with st.sidebar:
st.title("造价工程指标管理器")
# 第一个expander：指标管理
with st.expander("新建指标工具", expanded=True):
@@ -381,6 +394,11 @@ def main():
# 创建新指标
new_indicator = st.session_state.indicator_manager.add_indicator(name=new_name)
st.session_state.current_indicator_id = new_indicator.id
# 清除已生成的代码
if "generated_code" in st.session_state:
del st.session_state.generated_code
# 清除第二个expander的状态
st.session_state.view_indicator_detail = False
save_session_state()
st.rerun()
except ValueError as e:
@@ -389,6 +407,11 @@ def main():
if st.button("清空", help="删除所有指标", type="secondary"):
st.session_state.indicator_manager.clear_all_indicators()
st.session_state.current_indicator_id = None
# 清除已生成的代码
if "generated_code" in st.session_state:
del st.session_state.generated_code
# 清除第二个expander的状态
st.session_state.view_indicator_detail = False
save_session_state()
st.rerun()
@@ -405,6 +428,14 @@ def main():
type="primary" if st.session_state.current_indicator_id == indicator.id else "secondary"
):
st.session_state.current_indicator_id = indicator.id
# 更新显示的代码
if indicator.code:
st.session_state.generated_code = indicator.code
elif "generated_code" in st.session_state:
# 如果当前指标没有代码，清除session中的代码
del st.session_state.generated_code
# 清除第二个expander的状态
st.session_state.view_indicator_detail = False
save_session_state()
st.rerun()
@@ -412,7 +443,7 @@ def main():
with st.expander("查看指标库", expanded=False):
# 设置默认文件路径
file_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "../tests", "code.jsonl")
file_path = os.path.join(DATA_DIR, "code.jsonl")
# 文件路径输入
#file_path = st.text_input("JSONL文件路径", value=default_file_path, key="jsonl_path")
@@ -451,71 +482,82 @@ def main():
for i, record in enumerate(records):
if st.button(record.get("name", f"指标 {i+1}"), key=f"lib_btn_{i}", use_container_width=True):
st.session_state.selected_index = i
# 可以添加查看指标详情的逻辑
# 清除第一个expander的选中状态
st.session_state.current_indicator_id = None
# 设置查看指标详情的逻辑
st.session_state.view_indicator_detail = True
st.session_state.current_library_indicator = record
save_session_state()
st.rerun()
# 主区域
if "view_indicator_detail" in st.session_state and st.session_state.view_indicator_detail and "current_library_indicator" in st.session_state:
# 显示从指标库中选择的指标详细信息
library_indicator = st.session_state.current_library_indicator
st.header(f"指标名称: {library_indicator.get('name', '未命名指标')}")
# 指标名称和操作按钮
col1, col2, col3 = st.columns([1, 4, 2])
with col1:
st.markdown('<div><b style="font-size: 24px;">指标名称：</b></div>', unsafe_allow_html=True)
with col2:
st.markdown(f'<div style="font-size: 20px; padding-top: 4px;">{library_indicator.get("name", "未命名指标")}</div>', unsafe_allow_html=True)
# 查询语句部分
title_col, run_btn_col, result_col = st.columns([1, 1, 4])
with title_col:
st.subheader("查询语句")
with result_col:
title_query_result = st.subheader("")
# 显示查询语句
st.subheader("查询语句")
st.info(library_indicator.get('query', '无查询语句'))
st.text_area(
"查询语句",
value=library_indicator.get('query', '无查询语句'),
height=70,
key="lib_query_display",
disabled=True,
label_visibility="collapsed"
)
# 指标代码部分
title_col, run_btn_col, result_col = st.columns([1, 1, 4])
with title_col:
st.subheader("指标代码")
with result_col:
title_code_result = st.subheader("")
with run_btn_col:
if st.button("执行代码", key="execute_lib_indicator", use_container_width=True):
with st.spinner('正在执行代码...'):
# 确保code_executor已初始化
if "code_executor" in st.session_state:
result = st.session_state.code_executor.execute_code(library_indicator.get('code', ''))
if result and hasattr(result, 'get') and result.get('status'):
title_code_result.success("执行成功")
title_code_result.json(result.get('data', {}))
else:
title_code_result.error(f"执行失败: {result.get('message', '未知错误') if hasattr(result, 'get') else str(result)}")
else:
title_code_result.error("代码执行器未初始化")
# 显示代码
st.subheader("指标代码")
st.code(library_indicator.get('code', '无代码'), language='python')
# 添加执行按钮
if st.button("执行指标代码", key="execute_lib_indicator"):
with st.spinner('正在执行代码...'):
# 确保code_executor已初始化
if "code_executor" in st.session_state:
result = st.session_state.code_executor.execute_code(library_indicator.get('code', ''))
if result and hasattr(result, 'get') and result.get('status'):
st.success("执行成功")
st.json(result.get('data', {}))
else:
st.error(f"执行失败: {result.get('message', '未知错误') if hasattr(result, 'get') else str(result)}")
else:
st.error("代码执行器未初始化")
# 添加导入按钮 - 将库中指标导入到当前工作区
if st.button("导入到我的指标", key="import_to_my_indicators"):
try:
new_indicator = st.session_state.indicator_manager.add_indicator(
name=library_indicator.get('name', '导入的指标'),
query=library_indicator.get('query', '')
)
st.session_state.current_indicator_id = new_indicator.id
st.session_state.view_indicator_detail = False
save_session_state()
st.success(f"已导入指标: {new_indicator.name}")
st.rerun()
except Exception as e:
st.error(f"导入失败: {str(e)}")
# 添加返回按钮
if st.button("返回", key="back_to_indicators"):
st.session_state.view_indicator_detail = False
st.rerun()
# 移除导入按钮部分
elif st.session_state.current_indicator_id is not None:
current_indicator = st.session_state.indicator_manager.get_indicator(st.session_state.current_indicator_id)
if current_indicator:
# 指标名称和保存按钮
col1, col2, col3 = st.columns([1, 5, 2])
col1, col2, col3 = st.columns([1, 4, 2])
with col1:
st.markdown('<div><b style="font-size: 24px;">指标名称：</b></div>', unsafe_allow_html=True)
with col2:
new_name = st.text_input("", value=current_indicator.name, key="indicator_name", label_visibility="collapsed")
new_name = st.text_input("指标名称", value=current_indicator.name, key="indicator_name", label_visibility="collapsed")
if new_name != current_indicator.name:
if not new_name.strip():
st.error("指标名称不能为空")
@@ -531,44 +573,108 @@ def main():
if st.button("💾 保存指标", use_container_width=True):
if not new_name.strip():
st.error("指标名称不能为空")
elif not current_indicator.query.strip():
st.error("请先输入查询语句")
elif not current_indicator.code:
st.error("请先生成指标代码")
else:
try:
st.toast("指标已保存")
# 保存到指标管理器
st.session_state.indicator_manager.save_indicators()
# 同时保存到指标库
library_path = os.path.join(DATA_DIR, "code.jsonl")
new_record = {
"name": current_indicator.name,
"query": current_indicator.query,
"code": current_indicator.code,
"created_at": time.time()
}
# 追加到文件
with open(library_path, "a", encoding="utf-8") as f:
f.write(json.dumps(new_record, ensure_ascii=False) + "\n")
# 清除指标库的缓存，以便重新加载
load_jsonl.clear()
st.toast("指标已保存")
st.rerun() # 刷新页面以更新指标库列表
except Exception as e:
st.error(f"保存失败: {str(e)}")
# 查询语句
st.subheader("查询语句")
col1, col2 = st.columns([6, 2])
with col1:
query = st.text_area(
"输入查询语句",
value=current_indicator.query,
height=70,
key="query_input",
placeholder="在此输入查询语句...",
label_visibility="collapsed"
title_col, test_btn_col, result_col = st.columns([1, 1, 4])
with title_col:
st.subheader("查询语句")
with result_col:
title_query_result = st.subheader("")
# 查询输入框
query = st.text_area(
"输入查询语句",
value=current_indicator.query,
height=70,
key="query_input",
placeholder="在此输入查询语句...",
label_visibility="collapsed"
)
if query != current_indicator.query:
st.session_state.indicator_manager.update_indicator(
current_indicator.id, query=query
)
if query != current_indicator.query:
st.session_state.indicator_manager.update_indicator(
current_indicator.id, query=query
)
with col2:
st.write("") # 添加一些空间以对齐按钮
st.write("")
if st.button("🧪 测试", use_container_width=True):
with st.spinner("正在测试查询..."):
# 测试按钮改为生成指标代码按钮
with test_btn_col:
if st.button("🔄 生成代码", key="generate_code_btn", use_container_width=True):
with st.spinner("正在生成指标代码..."):
# 使用缓存的测试函数
result = test_indicator_query(query)
st.session_state.test_result = result
result = generate_indicator_code(query)
# 根据结果显示不同的状态
if result.startswith("✅"):
title_query_result.success(result)
# 生成成功后，立即保存代码到指标对象中
if "generated_code" in st.session_state:
st.session_state.indicator_manager.update_indicator(
current_indicator.id,
code=st.session_state.generated_code
)
elif result.startswith("❌"):
title_query_result.error(result)
else:
# 显示实体识别结果或其他信息
title_query_result.info(result)
# 测试结果
if "test_result" in st.session_state:
st.subheader("测试结果")
st.code(st.session_state.test_result)
# 添加代码执行部分
title_col, run_btn_col, result_col = st.columns([1, 1, 4])
with title_col:
st.subheader("指标代码")
with result_col:
title_code_result = st.subheader("")
with run_btn_col:
if st.button("执行代码", key="execute_code_btn", use_container_width=True):
with st.spinner("正在执行代码..."):
if "code_executor" in st.session_state and "generated_code" in st.session_state:
result = st.session_state.code_executor.execute_code(st.session_state.generated_code)
if result and hasattr(result, 'get') and result.get('status'):
title_code_result.success("执行成功")
title_code_result.json(result.get('data', {}))
# 更新指标的代码
st.session_state.indicator_manager.update_indicator(
current_indicator.id,
code=st.session_state.generated_code
)
else:
title_code_result.error(f"执行失败: {result.get('message', '未知错误') if hasattr(result, 'get') else str(result)}")
else:
title_code_result.error("代码执行器未初始化或没有生成的代码")
# 显示代码
st.code(st.session_state.get("generated_code", "# 暂无代码，请输入有效的查询语句并点击测试按钮"), language="python")
# 移除保存到指标库按钮部分
else:
st.warning("找不到当前指标，可能已被删除")
else:
src/prompt_manager.py
+6 -5
View File
@@ -74,11 +74,12 @@ def project_get_calculate_function():
return result_dict
# 执行规则
- 参数必须从用户问题或上下文信息中提取
- 禁止在代码函数范围外添加任何注释或解释或非代码内容
- 输出代码中必须以def project_get_calculate_function() -> dict函数作为入口函数
- 必须确保生成的代码可以直接执行，如果函数功能求取数值，project的函数返回结果为空或出错则算成功，data为0，并在message说明错误原因，代码要注意进行各类容错检查
- ProjectToolkit 类中涉及项目划分的函数已考虑在其及其子孙项目划分下查找，所以无需生成递归子项目划分的代码
- 参数必须从用户问题或上下文信息中提取。
- 在代码函数内部生成功能说明，在函数外禁止生成任何注释或解释或非代码内容。
- 输出代码中必须以def project_get_calculate_function() -> dict函数作为入口函数,该函数成功时返回的'data'通常是浮点或整型值，除非用户要求返回其他类型。
- 必须确保生成的代码可以直接执行，代码要注意进行各类容错检查。
- 如果project_get_calculate_function函数需要返回浮点整形数，函数中间发生错误或找不到对象也必须返回成功，data为0，并在message说明错误原因。
- ProjectToolkit 类中涉及项目划分的函数已考虑在其及其子孙项目划分下查找，所以无需生成递归子项目划分的代码。
- 如果文本中包含范围编码格式则需要进行编码展开，如'YX2-1~7'展开为‘YX2-1/YX2-2/YX2-3/YX2-4/YX2-5/YX2-6/YX2-7’
"""
)