Merge branch 'main' of https://git.yishihui.com/ai/knowledge-agent

agent.py

@@ -21,7 +21,7 @@ from fastapi import FastAPI, HTTPException, BackgroundTasks
 from fastapi.responses import JSONResponse
 from pydantic import BaseModel, Field
 import uvicorn
-from agents.clean_agent.agent import execute_agent_with_api
+from agents.clean_agent.agent import execute_agent_with_api, execute
 from agents.expand_agent.agent import execute_expand_agent_with_api, _update_expansion_status
 
 # LangGraph 相关导入
@@ -732,7 +732,8 @@ async def extract(request: ExtractRequest):
         # 创建线程池任务执行Agent
         def _execute_extract_sync():
             try:
-                result = execute_agent_with_api(json.dumps({"query_word": query, "request_id": requestId}))
+                # result = execute_agent_with_api(json.dumps({"query_word": query, "request_id": requestId}))
+                result = execute(query, requestId)
                 # 更新状态为处理完成
                 update_extract_status(requestId, 2)
                 logger.info(f"异步提取任务完成: requestId={requestId}")

agents/clean_agent/agent.py

@@ -4,7 +4,8 @@ from langgraph.graph import StateGraph, START, END
 from langgraph.graph.message import add_messages
 import os
 from langchain_openai import ChatOpenAI
-from .tools import evaluation_extraction_tool
+from .tools import evaluation_extraction_tool, evaluation_extraction
+import uuid
 
 from langgraph.prebuilt import ToolNode, tools_condition
 from langgraph.checkpoint.memory import InMemorySaver
@@ -61,6 +62,10 @@ def chatbot(state: State):
     return {"messages": [message]}
 
 def execute_agent_with_api(user_input: str):
+            # 生成唯一的线程ID
+    import uuid
+    thread_id = str(uuid.uuid4())
+    logger.info(f"开始执行提取，user_input={user_input}, thread_id={thread_id}")
     global graph, llm_with_tools, prompt
     
     # 替换prompt中的{input}占位符为用户输入
@@ -96,10 +101,6 @@ def execute_agent_with_api(user_input: str):
                 logger.error(f"初始化Agent失败: {str(e)}")
                 return f"初始化Agent失败: {str(e)}"
         
-        # 生成唯一的线程ID
-        import uuid
-        thread_id = str(uuid.uuid4())
-        
         # 执行Agent并收集结果
         results = []
         config = {"configurable": {"thread_id": thread_id}}
@@ -113,12 +114,19 @@ def execute_agent_with_api(user_input: str):
                     results.append(message.content)
         
         # 返回结果
-        return "\n".join(results) if results else "Agent执行完成，但没有返回结果"
+        res="\n".join(results) if results else "Agent执行完成，但没有返回结果"
+        logger.info(f"Agent执行完成，返回结果: {res}, thread_id={thread_id}")
+        return res
     except requests.exceptions.ConnectionError as e:
         return f"OpenAI API 连接错误: {str(e)}\n请检查网络连接或代理设置。"
     except Exception as e:
         return f"执行Agent时出错: {str(e)}"
 
+def execute(query_word: str, request_id: str):
+    logger.info(f"开始处理，request_id: {request_id}, query_word: {query_word}")
+    result = evaluation_extraction(request_id, query_word)
+    return result
+
 def main():
     print(f"开始执行Agent")
     # 设置代理
@@ -128,7 +136,8 @@ def main():
         os.environ["HTTPS_PROXY"] = proxy_url
         os.environ["HTTP_PROXY"] = proxy_url
     # 执行Agent
-    result = execute_agent_with_api('{"query_word":"图文策划方法","request_id":"REQUEST_001"}')
+    # result = execute_agent_with_api('{"query_word":"图文策划方法","request_id":"REQUEST_001"}')
+    result = execute("图文策划方法", "REQUEST_001")
     print(result)
 
 if __name__ == '__main__':

agents/clean_agent/tools.py

@@ -55,6 +55,30 @@ def evaluation_extraction_tool(request_id: str, query_word: str) -> str:
             logger.error(f"评估抽取过程中出错: {e}")
             return f"no data - 错误: {str(e)}"
 
+def evaluation_extraction(request_id: str, query_word: str) -> str:
+    """
+    知识评估与抽取工具。持续处理数据库中的数据，分批执行评估并创建KnowledgeExtractionContent对象。
+    对于评分大于70分的内容，会进行抽取并更新KnowledgeExtractionContent对象。
+    
+    Args:
+        request_id: 请求ID，如果不提供则处理所有未处理的数据
+        query_word: 查询词，用于评估和抽取内容
+        
+    Returns:
+        str: "success" 表示处理完成，"no data" 表示没有数据需要处理
+    """
+    # 使用上下文管理器自动管理数据库连接的生命周期
+    with SessionLocal() as db:
+        try:
+            # 使用新的批量处理函数
+            result = execute_continuous_evaluation_extraction(request_id, db, query_word)
+            return result
+        except Exception as e:
+            # 确保发生异常时回滚事务
+            db.rollback()
+            logger.error(f"评估抽取过程中出错: {e}")
+            return f"no data - 错误: {str(e)}"
+
 def execute_continuous_evaluation_extraction(request_id: str, db: Session, query_word: str) -> str:
     """持续执行评估循环，直到数据库没有数据"""
     logger.info(f"开始处理，request_id: {request_id}, query_word: {query_word}")