update api

EdgeCraftRAG/docker_compose/intel/gpu/arc/README.md

@@ -34,7 +34,7 @@ EC-RAG supports vLLM deployment(default method) and local OpenVINO deployment fo
 
 #### Intel Arc A770
 **OS**: Ubuntu Server 22.04.1 or newer (at least 6.2 LTS kernel)  
-**Driver & libraries**: Please refer to [Installing GPUs Drivers](https://dgpu-docs.intel.com/driver/client/overview.html#ubuntu-22.04) for detailed driver & libraries setup  
+**Driver & libraries**: Please refer to [Installing GPUs Drivers](https://dgpu-docs.intel.com/driver/installation-rolling.html#installing-gpu-drivers) for detailed driver & libraries setup  
 **Available Inferencing Framework**: openVINO, vLLM
 
 ### 2. Access the Code

EdgeCraftRAG/docker_compose/intel/gpu/arc/README_zh.md

@@ -34,7 +34,7 @@ EC-RAG 支持 vLLM 部署（默认方式）以及面向 Intel Arc GPU 和 Core U
 
 #### Intel Arc A770
 **操作系统**：Ubuntu Server 22.04.1 或更高版本（至少 6.2 LTS 内核）  
-**驱动与库**：详细驱动与库安装请参考 [Installing GPUs Drivers](https://dgpu-docs.intel.com/driver/client/overview.html#ubuntu-22.04)  
+**驱动与库**：详细驱动与库安装请参考 [Installing GPUs Drivers](https://dgpu-docs.intel.com/driver/installation-rolling.html#installing-gpu-drivers)  
 **可用推理框架**：openVINO、vLLM
 
 ### 2. 获取代码

EdgeCraftRAG/docker_compose/intel/gpu/arc/compose.yaml

@@ -88,6 +88,7 @@ services:
       https_proxy: ${https_proxy}
       vLLM_ENDPOINT: ${vLLM_ENDPOINT:-http://${HOST_IP}:${VLLM_SERVICE_PORT_B60:-8086}}
       OVMS_ENDPOINT: ${OVMS_ENDPOINT:-http://${HOST_IP}:${OVMS_SERVICE_PORT:-8000}}
+      ECRAG_LLM_API_MODE: ${ECRAG_LLM_API_MODE:-chat}
       LLM_MODEL: ${LLM_MODEL}
       ENABLE_BENCHMARK: ${ENABLE_BENCHMARK:-false}
       MAX_MODEL_LEN: ${MAX_MODEL_LEN:-49152}

EdgeCraftRAG/docs/API_Guide.md

@@ -432,28 +432,6 @@ curl -X GET http://${HOST_IP}:16010/v1/data/documents \
   -H "Content-Type: application/json" | jq '.'
 ```
 
-### Get all files
-
-```bash
-curl -X GET http://${HOST_IP}:16010/v1/data/files \
-  -H "Content-Type: application/json" | jq '.'
-```
-
-### Get a specific file
-
-```bash
-curl -X GET http://${HOST_IP}:16010/v1/data/files/{name} \
-  -H "Content-Type: application/json" | jq '.'
-```
-
-### Upload a file (from UI)
-
-```bash
-curl -X POST "http://${HOST_IP}:16010/v1/data/file/{file_name}" \
-  -H "Content-Type: multipart/form-data" \
-  -F "file=@/path/to/your/document.pdf" | jq '.'
-```
-
 ---
 
 ## Session Management

EdgeCraftRAG/edgecraftrag/api/v1/chatqna.py

@@ -8,11 +8,11 @@
 import requests
 from comps.cores.proto.api_protocol import ChatCompletionRequest
 from edgecraftrag.api_schema import RagOut
-from edgecraftrag.base import GeneratorType
+from edgecraftrag.base import GeneratorType, InferenceType
 from edgecraftrag.context import ctx
 from edgecraftrag.utils import chain_async_generators, serialize_contexts, serialize_node_with_score, stream_generator
 from fastapi import Body, FastAPI, HTTPException, status
-from fastapi.responses import StreamingResponse
+from fastapi.responses import JSONResponse, StreamingResponse
 
 chatqna_app = FastAPI()
 thread_pool = ThreadPoolExecutor(max_workers=16)
@@ -220,3 +220,44 @@ async def save_session(sessionid, run_agent_gen):
         yield chunk or ""
         await asyncio.sleep(0)
     session_mgr.save_current_message(sessionid, "assistant", current_content)
+
+def _not_ready(reason: str):
+    return JSONResponse(
+        status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+        content={"status": "not_ready", "reason": reason},
+    )
+
+# Lightweight readiness check before sending a real RAG request
+@chatqna_app.get(path="/v1/ready")
+async def get_ready():
+    pipeline = ctx.get_pipeline_mgr().get_active_pipeline()
+    if pipeline is None or not pipeline.status.active:
+        return _not_ready("No active pipeline")
+
+    generator = pipeline.get_generator(GeneratorType.CHATQNA)
+    if generator is not None and generator.inference_type == InferenceType.VLLM:
+        try:
+            response = requests.get(f"{generator.vllm_endpoint.rstrip('/')}/v1/models", timeout=2)
+            response.raise_for_status()
+        except Exception:
+            return _not_ready("LLM backend unavailable")
+
+    try:
+        active_kbs = ctx.knowledgemgr.get_active_knowledge_base()
+        if not active_kbs:
+            return _not_ready("Retrieval unavailable")
+        request = ChatCompletionRequest(messages="ready")
+        request.user = active_kbs
+        result = await ctx.get_pipeline_mgr().run_retrieve(chat_request=request)
+        if result == -1:
+            return _not_ready("Retrieval unavailable")
+    except Exception:
+        return _not_ready("Retrieval unavailable")
+
+    return {
+        "status": "ready",
+        "pipeline": pipeline.name,
+        "pipeline_active": pipeline.status.active,
+        "llm": "ready",
+        "retrieval": "ready",
+    }

EdgeCraftRAG/edgecraftrag/components/agent.py

@@ -12,6 +12,20 @@
 from langgraph.config import get_stream_writer
 from pydantic import BaseModel, Field, model_serializer
 
+from langchain_core.runnables.config import ensure_config, var_child_runnable_config
+from langgraph._internal._runnable import RunnableCallable
+_orig = RunnableCallable.ainvoke
+
+async def _patched(self, input, config=None, **kwargs):
+    token = var_child_runnable_config.set(config or ensure_config())
+    try:
+        return await _orig(self, input, config, **kwargs)
+    finally:
+        var_child_runnable_config.reset(token)
+
+
+RunnableCallable.ainvoke = _patched
+
 
 class Retrieval(BaseModel):
     step: Any
@@ -59,9 +73,9 @@ async def llm_generate(self, request: ChatCompletionRequest, streaming):
         response = await self._run_pipeline_generate(request)
         return response
 
-    async def llm_generate_astream_writer(self, request, prefix=None, suffix=None) -> str:
+    async def llm_generate_astream_writer(self, request, prefix=None, suffix=None, writer=None) -> str:
         response = ""
-        writer = get_stream_writer()
+        writer = _safe_writer(writer)
         first = True
         generator = await self.llm_generate(request, True)
         async for chunk in generator:
@@ -123,7 +137,15 @@ def ser_model(self):
         return set
 
 
-async def stream_writer(input):
-    writer = get_stream_writer()
+def _safe_writer(writer=None):
+    if writer is not None:
+        return writer
+    try:
+        return get_stream_writer()
+    except RuntimeError:
+        return lambda chunk: None
+
+async def stream_writer(input, writer=None):
+    writer = _safe_writer(writer)
     async for chunk in stream_generator(input):
         writer(chunk)

EdgeCraftRAG/edgecraftrag/components/agents/simple.py

@@ -124,7 +124,7 @@ def _build_graph(self):
 
         return qnagraph.compile()
 
-    async def retrieve(self, state: QnaState) -> dict:
+    async def retrieve(self, state: QnaState, writer=None) -> dict:
         # print(f"State Retrieve {state}")
         request = state.request
         request.messages = state.query
@@ -137,7 +137,8 @@ async def retrieve(self, state: QnaState) -> dict:
             f"Retrieved {format_terminal_str(str(len(retrieved)), color='magenta', bold=True)} documents, Reranked to top {format_terminal_str(str(len(reranked)), color='magenta', bold=True)}.",
         )
         await stream_writer(
-            f"\n\n🔍 **Retrieved {str(len(retrieved))} documents, Reranked to top {str(len(reranked))}**\n\n"
+            f"\n\n🔍 **Retrieved {str(len(retrieved))} documents, Reranked to top {str(len(reranked))}**\n\n",
+            writer=writer,
         )
 
         new_retrieval = Retrieval(step=state.num_retrievals, query=state.query, retrieved=retrieved, reranked=reranked)
@@ -146,9 +147,9 @@ async def retrieve(self, state: QnaState) -> dict:
             "retrievals": [*state.retrievals, new_retrieval],
         }
 
-    async def generate_query(self, state: QnaState) -> dict:
+    async def generate_query(self, state: QnaState, writer=None) -> dict:
         # print(f"State generate_query {state}")
-        await stream_writer('<agent title="Understanding the user\'s question">')
+        await stream_writer('<agent title="Understanding the user\'s question">', writer=writer)
 
         messages = [
             {"role": ROLE.USER, "content": state.question},
@@ -157,7 +158,7 @@ async def generate_query(self, state: QnaState) -> dict:
         self._messages.extend(messages)
         self.conversation_history.extend(messages)
 
-        response = await self.llm_generate_astream_writer(state.request)
+        response = await self.llm_generate_astream_writer(state.request, writer=writer)
 
         message = {"role": ROLE.ASSISTANT, "content": response}
         self._messages.append(message)
@@ -166,10 +167,10 @@ async def generate_query(self, state: QnaState) -> dict:
             "query": response,
         }
 
-    async def check_retrieved(self, state: QnaState) -> str:
+    async def check_retrieved(self, state: QnaState, writer=None) -> str:
         # print(f"State check_retrieved {state}")
         print("🤔", format_terminal_str("Evaluating if more information is needed", color="green"))
-        await stream_writer("🤔 **Evaluating if more information is needed...**\n\n")
+        await stream_writer("🤔 **Evaluating if more information is needed...**\n\n", writer=writer)
 
         # Format context for the next decision
         contexts = self.cfg.prompt_templates.contexts.format(
@@ -191,27 +192,28 @@ async def check_retrieved(self, state: QnaState) -> str:
                 ),
             )
             await stream_writer(
-                f"\n\n⚠️ **Reached maximum retrievals: {self.max_retrievals}, stopping searching...**\n\n</agent>"
+                f"\n\n⚠️ **Reached maximum retrievals: {self.max_retrievals}, stopping searching...**\n\n</agent>",
+                writer=writer,
             )
             return "stop"
         else:
-            response = await self.llm_generate_astream_writer(state.request)
+            response = await self.llm_generate_astream_writer(state.request, writer=writer)
             message = {"role": ROLE.ASSISTANT, "content": response}
             self._messages.append(message)
             self.conversation_history.append(message)
             if response.upper().startswith("NO"):
                 print("✅", format_terminal_str("Information is sufficient, moving to next step\n", color="green"))
-                await stream_writer("\n\n✅ **Information is sufficient, moving to next step...**\n\n</agent>")
+                await stream_writer("\n\n✅ **Information is sufficient, moving to next step...**\n\n</agent>", writer=writer)
                 return "stop"
             else:
                 print("🔄", format_terminal_str("Need more information, generating new query ...", color="green"))
-                await stream_writer("\n\n🔄 **Need more information, generating new query...**\n\n</agent>")
+                await stream_writer("\n\n🔄 **Need more information, generating new query...**\n\n</agent>", writer=writer)
                 return "continue"
 
-    async def generate_answer(self, state: QnaState) -> dict:
+    async def generate_answer(self, state: QnaState, writer=None) -> dict:
         # print(f"State generate_answer {state}")
         print("📝", format_terminal_str("Generating the final answer ...", color="cyan", bold=True))
-        await stream_writer('<agent title="Generating the final answer ..." tag="nofold"></agent>')
+        await stream_writer('<agent title="Generating the final answer ..." tag="nofold"></agent>', writer=writer)
         plan_with_information = ""
         prev_step = ""
         for i, r in enumerate(state.retrievals):
@@ -232,7 +234,7 @@ async def generate_answer(self, state: QnaState) -> dict:
         ]
         self.conversation_history.extend(self._messages)
 
-        response = await self.llm_generate_astream_writer(state.request)
+        response = await self.llm_generate_astream_writer(state.request, writer=writer)
 
         self.conversation_history.append({"role": ROLE.ASSISTANT, "content": response})
         answer = self.postproc_answer(response, state)