arch(core): use fastapi for launching consumers

Author: gusye1234

src/server/core/acontext_core/llm/agent/task.py

@@ -72,7 +72,7 @@ async def build_task_ctx(
     current_tasks, eil = r.unpack()
     if eil:
         return r
-    LOG.info(
+    LOG.debug(
         f"Built task context {[(t.order, t.status.value, t.task_description) for t in current_tasks]}"
     )
     use_ctx = TaskCtx(

src/server/core/acontext_core/llm/complete/openai_sdk.py

@@ -53,6 +53,7 @@ async def openai_complete(
         timeout=DEFAULT_CORE_CONFIG.llm_response_timeout,
         max_tokens=max_tokens,
         tools=tools,
+        **DEFAULT_CORE_CONFIG.llm_openai_completion_kwargs,
         **kwargs,
     )
     cached_tokens = getattr(response.usage.prompt_tokens_details, "cached_tokens", None)

src/server/core/acontext_core/llm/prompt/task.py

@@ -65,17 +65,16 @@ def system_prompt(cls) -> str:
 - Message with ID format: <message id=N> ... </message>, inside the tag is the message content, the id field indicates the message id.
 
 ## Report your Thinking
-Use extremely brief wordings to report:
-1. Any user requirement or planning?
+Use extremely brief wordings to report before calling tools:
+1. Any planning from agent? Any requirement or task modification from user?
 2. Does the user report that any task failed and need to re-run?
 3. How existing tasks are related to current conversation? 
-4. Any new task is created?
+4. Any new task should be created?
 5. Which Messages are contributed to planning? 
 6. Which of them are contributed to which task?
 7. Which task's status/description need to be updated?
 8. Briefly describe your tool-call actions to correctly manage the tasks.
-
-Make sure your will call `finish` tool after every tools are called
+9. Make sure your will call `finish` tool after every tools are called
 """
 
     @classmethod

src/server/core/acontext_core/llm/prompt/task_sop.py

@@ -7,19 +7,41 @@ def system_prompt(cls) -> str:
         return """You're a Tool-calling SOP Agent that read the raw working history between user and agent, and generate a tool-calling SOP for the task.
 
 ## What is a Tool-calling SOP?
+### Standard Format (JSON)
+{
+    "use_when": str,
+    "notes": str,
+    "sop":[
+        {"tool_name": str, "argument_template": dict},
+        ...
+    ]
+}
+### Format Breaking down
+- 'use_when': The scenario when this sop maybe used, e.g. 'Broswering xxx.com for items' infos', 'Query Lung disease from Database'
+- 'notes': An brief guideline to instruct how to proceed this SOP, maybe containing user requirements, tool-use annotations.
+- 'sop': a structured array that contains tool-calling steps in correct order, for each step:
+    - 'tool_name': exact corresponding tool name from history
+    - 'argument_template': a json of the certain tool-calling arguments and their values that necessage to this SOP.
+      For example, "arugment_template": {"website": "xxx.com", "action": "click shopping tab in navbar"}
+
+## Input
+### Raw History Input
 ...
 
-## Input Format
-
-## Output Format
-
-## Thinking Guidelines
+### User Planning History
+...
 
+## Think before Answer
+...
 """
 
     @classmethod
-    def pack_task_input(cls, *args, **kwargs) -> str:
-        pass
+    def pack_task_input(cls, history_messages: str, history_planning: str) -> str:
+        return f"""## Raw History Input
+{history_messages}
+## User Planning History
+{history_planning}
+"""
 
     @classmethod
     def prompt_kwargs(cls) -> str:

src/server/core/acontext_core/schema/block/sop_block.py

@@ -5,7 +5,7 @@
 
 class SOPStep(BaseModel):
     tool_name: str
-    tool_arguments_with_placeholder: dict[str, Any]
+    argument_template: dict[str, Any]
     props: Optional[dict] = None
 
 

src/server/core/acontext_core/schema/config.py

@@ -18,6 +18,7 @@ class CoreConfig(BaseModel):
     llm_base_url: Optional[str] = None
     llm_openai_default_query: Optional[Mapping[str, Any]] = None
     llm_openai_default_header: Optional[Mapping[str, Any]] = None
+    llm_openai_completion_kwargs: Mapping[str, Any] = {}
     llm_response_timeout: float = 60
     llm_sdk: Literal["openai", "anthropic"] = "openai"
 

src/server/core/api.py

@@ -0,0 +1,29 @@
+import asyncio
+from contextlib import asynccontextmanager
+from fastapi import FastAPI
+from acontext_core.di import setup, cleanup, MQ_CLIENT, LOG, DB_CLIENT, S3_CLIENT
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    # Startup
+    await setup()
+    # Run consumer in the background
+    asyncio.create_task(MQ_CLIENT.start())
+    yield
+    # Shutdown
+    await cleanup()
+
+
+app = FastAPI(lifespan=lifespan)
+
+
+@app.get("/health")
+async def health() -> str:
+    if not await MQ_CLIENT.health_check():
+        return "MQ consumer error"
+    if not await DB_CLIENT.health_check():
+        return "DB client error"
+    if not await S3_CLIENT.health_check():
+        return "S3 client error"
+    return "ok"