add: handle error

app.py

@@ -9,239 +9,185 @@ import shutil
 import traceback
 import uuid
 from collections import deque
-from datetime import datetime
-from enum import Enum
 from functools import partial
-from typing import Any, Optional
+from json import JSONDecodeError
+from typing import Dict
 
 import fire
+import openai
+import tenacity
 import uvicorn
 from fastapi import FastAPI, Request
-from fastapi.responses import StreamingResponse
+from fastapi.responses import StreamingResponse, JSONResponse
 from fastapi.staticfiles import StaticFiles
 from loguru import logger
-from metagpt.actions.action import Action
-from metagpt.actions.action_output import ActionOutput
 from metagpt.config import CONFIG
 from metagpt.logs import set_llm_stream_logfunc
 from metagpt.schema import Message
-from pydantic import BaseModel, Field
-
+from openai import OpenAI
+
+from data_model import (
+    NewMsg,
+    MessageJsonModel,
+    Sentences,
+    Sentence,
+    SentenceType,
+    SentenceValue,
+    ThinkActPrompt,
+    LLMAPIkeyTest,
+    ThinkActStep,
+)
+from message_enum import QueryAnswerType, MessageStatus
 from software_company import RoleRun, SoftwareCompany
 
 
-class QueryAnswerType(Enum):
-    Query = "Q"
-    Answer = "A"
-
-
-class SentenceType(Enum):
-    TEXT = "text"
-    HIHT = "hint"
-    ACTION = "action"
-    ERROR = "error"
-
-
-class MessageStatus(Enum):
-    COMPLETE = "complete"
-
-
-class SentenceValue(BaseModel):
-    answer: str
-
-
-class Sentence(BaseModel):
-    type: str
-    id: Optional[str] = None
-    value: SentenceValue
-    is_finished: Optional[bool] = None
-
-
-class Sentences(BaseModel):
-    id: Optional[str] = None
-    action: Optional[str] = None
-    role: Optional[str] = None
-    skill: Optional[str] = None
-    description: Optional[str] = None
-    timestamp: str = Field(default_factory=lambda: datetime.now().strftime("%Y-%m-%dT%H:%M:%S.%f%z"))
-    status: str
-    contents: list[dict]
-
-
-class NewMsg(BaseModel):
-    """Chat with MetaGPT"""
-
-    query: str = Field(description="Problem description")
-    config: dict[str, Any] = Field(description="Configuration information")
-
-
-class ErrorInfo(BaseModel):
-    error: str = None
-    traceback: str = None
-
-
-class ThinkActStep(BaseModel):
-    id: str
-    status: str
-    title: str
-    timestamp: str = Field(default_factory=lambda: datetime.now().strftime("%Y-%m-%dT%H:%M:%S.%f%z"))
-    description: str
-    content: Sentence = None
-
-
-class ThinkActPrompt(BaseModel):
-    message_id: int = None
-    timestamp: str = Field(default_factory=lambda: datetime.now().strftime("%Y-%m-%dT%H:%M:%S.%f%z"))
-    step: ThinkActStep = None
-    skill: Optional[str] = None
-    role: Optional[str] = None
-
-    def update_think(self, tc_id, action: Action):
-        self.step = ThinkActStep(
-            id=str(tc_id),
-            status="running",
-            title=action.desc,
-            description=action.desc,
-        )
-
-    def update_act(self, message: ActionOutput | str, is_finished: bool = True):
-        if is_finished:
-            self.step.status = "finish"
-        self.step.content = Sentence(
-            type=SentenceType.TEXT.value,
-            id=str(1),
-            value=SentenceValue(answer=message.content if is_finished else message),
-            is_finished=is_finished,
-        )
-
-    @staticmethod
-    def guid32():
-        return str(uuid.uuid4()).replace("-", "")[0:32]
-
-    @property
-    def prompt(self):
-        return self.json(exclude_unset=True)
-
-
-class MessageJsonModel(BaseModel):
-    steps: list[Sentences]
-    qa_type: str
-    created_at: datetime = Field(default_factory=datetime.now)
-    query_time: datetime = Field(default_factory=datetime.now)
-    answer_time: datetime = Field(default_factory=datetime.now)
-    score: Optional[int] = None
-    feedback: Optional[str] = None
-
-    def add_think_act(self, think_act_prompt: ThinkActPrompt):
-        s = Sentences(
-            action=think_act_prompt.step.title,
-            skill=think_act_prompt.skill,
-            description=think_act_prompt.step.description,
-            timestamp=think_act_prompt.timestamp,
-            status=think_act_prompt.step.status,
-            contents=[think_act_prompt.step.content.dict()],
-        )
-        self.steps.append(s)
-
-    @property
-    def prompt(self):
-        return self.json(exclude_unset=True)
-
-
-async def create_message(req_model: NewMsg, request: Request):
-    """
-    Session message stream
-    """
-    tc_id = 0
-    try:
-        exclude_keys = CONFIG.get("SERVER_METAGPT_CONFIG_EXCLUDE", [])
-        config = {k.upper(): v for k, v in req_model.config.items() if k not in exclude_keys}
-        set_context(config, uuid.uuid4().hex)
-
-        msg_queue = deque()
-        CONFIG.LLM_STREAM_LOG = lambda x: msg_queue.appendleft(x) if x else None
-
-        role = SoftwareCompany()
-        role.recv(message=Message(content=req_model.query))
-        answer = MessageJsonModel(
-            steps=[
-                Sentences(
-                    contents=[
-                        Sentence(
-                            type=SentenceType.TEXT.value, value=SentenceValue(answer=req_model.query), is_finished=True
-                        )
-                    ],
-                    status=MessageStatus.COMPLETE.value,
-                )
-            ],
-            qa_type=QueryAnswerType.Answer.value,
-        )
-
+class Service:
+    @classmethod
+    async def create_message(cls, req_model: NewMsg, request: Request):
+        """
+        Session message stream
+        """
+        tc_id = 0
         task = None
-
-        async def stop_if_disconnect():
-            while not await request.is_disconnected():
-                await asyncio.sleep(1)
-
-            if task is None:
-                return
-
-            if not task.done():
-                task.cancel()
-                logger.info(f"cancel task {task}")
-
-        asyncio.create_task(stop_if_disconnect())
-
-        while True:
-            tc_id += 1
-            if await request.is_disconnected():
-                return
-            think_result: RoleRun = await role.think()
-            if not think_result:  # End of conversion
-                break
-
-            think_act_prompt = ThinkActPrompt(role=think_result.role.profile)
-            think_act_prompt.update_think(tc_id, think_result)
-            yield think_act_prompt.prompt + "\n\n"
-            task = asyncio.create_task(role.act())
-
-            while not await request.is_disconnected():
-                if msg_queue:
-                    think_act_prompt.update_act(msg_queue.pop(), False)
-                    yield think_act_prompt.prompt + "\n\n"
-                    continue
-
-                if task.done():
+        try:
+            exclude_keys = CONFIG.get("SERVER_METAGPT_CONFIG_EXCLUDE", [])
+            config = {k.upper(): v for k, v in req_model.config.items() if k not in exclude_keys}
+            cls._set_context(config)
+            msg_queue = deque()
+            CONFIG.LLM_STREAM_LOG = lambda x: msg_queue.appendleft(x) if x else None
+
+            role = SoftwareCompany()
+            role.recv(message=Message(content=req_model.query))
+            answer = MessageJsonModel(
+                steps=[
+                    Sentences(
+                        contents=[
+                            Sentence(
+                                type=SentenceType.TEXT.value, value=SentenceValue(answer=req_model.query), is_finished=True
+                            )
+                        ],
+                        status=MessageStatus.COMPLETE.value,
+                    )
+                ],
+                qa_type=QueryAnswerType.Answer.value,
+            )
+
+            async def stop_if_disconnect():
+                while not await request.is_disconnected():
+                    await asyncio.sleep(1)
+
+                if task is None:
+                    return
+
+                if not task.done():
+                    task.cancel()
+                    logger.info(f"cancel task {task}")
+
+            asyncio.create_task(stop_if_disconnect())
+
+            while True:
+                tc_id += 1
+                if await request.is_disconnected():
+                    return
+                think_result: RoleRun = await role.think()
+                if not think_result:  # End of conversion
                     break
 
-                await asyncio.sleep(0.5)
-            else:
-                task.cancel()
-                return
-
-            act_result = await task
-            think_act_prompt.update_act(act_result)
+                think_act_prompt = ThinkActPrompt(role=think_result.role.profile)
+                think_act_prompt.update_think(tc_id, think_result)
+                yield think_act_prompt.prompt + "\n\n"
+                task = asyncio.create_task(role.act())
+
+                while not await request.is_disconnected():
+                    if msg_queue:
+                        think_act_prompt.update_act(msg_queue.pop(), False)
+                        yield think_act_prompt.prompt + "\n\n"
+                        continue
+
+                    if task.done():
+                        break
+
+                    await asyncio.sleep(0.5)
+                else:
+                    task.cancel()
+                    return
+
+                act_result = await task
+                think_act_prompt.update_act(act_result)
+                yield think_act_prompt.prompt + "\n\n"
+                answer.add_think_act(think_act_prompt)
+            yield answer.prompt + "\n\n"  # Notify the front-end that the message is complete.
+        except asyncio.CancelledError:
+            task.cancel()
+        except tenacity.RetryError as retry_error:
+            yield cls.handle_retry_error(tc_id, retry_error)
+        except Exception as ex:
+            description = str(ex)
+            answer = traceback.format_exc()
+            think_act_prompt = cls.create_error_think_act_prompt(tc_id, description, description, answer)
             yield think_act_prompt.prompt + "\n\n"
-            answer.add_think_act(think_act_prompt)
-        yield answer.prompt + "\n\n"  # Notify the front-end that the message is complete.
-    except asyncio.CancelledError:
-        task.cancel()
-    except Exception as ex:
-        description = str(ex)
-        answer = traceback.format_exc()
+        finally:
+            CONFIG.WORKSPACE_PATH: pathlib.Path
+            if CONFIG.WORKSPACE_PATH.exists():
+                shutil.rmtree(CONFIG.WORKSPACE_PATH)
+
+    @staticmethod
+    def create_error_think_act_prompt(tc_id: int, title, description: str, answer: str) -> ThinkActPrompt:
         step = ThinkActStep(
             id=tc_id,
             status="failed",
-            title=description,
+            title=title,
             description=description,
             content=Sentence(type=SentenceType.ERROR.value, id=1, value=SentenceValue(answer=answer), is_finished=True),
         )
-        think_act_prompt = ThinkActPrompt(step=step)
-        yield think_act_prompt.prompt + "\n\n"
-    finally:
-        CONFIG.WORKSPACE_PATH: pathlib.Path
-        if CONFIG.WORKSPACE_PATH.exists():
-            shutil.rmtree(CONFIG.WORKSPACE_PATH)
+        return ThinkActPrompt(step=step)
+
+    @classmethod
+    def handle_retry_error(cls, tc_id: int, error: tenacity.RetryError):
+        # Known exception handling logic
+        try:
+            # Try to get the original exception
+            original_exception = error.last_attempt.exception()
+            while isinstance(original_exception, tenacity.RetryError):
+                original_exception = original_exception.last_attempt.exception()
+
+            if isinstance(original_exception, openai.AuthenticationError):
+                answer = original_exception.message
+                title = "OpenAI AuthenticationError"
+                think_act_prompt = cls.create_error_think_act_prompt(tc_id, title, title, answer)
+                return think_act_prompt.prompt + "\n\n"
+            elif isinstance(original_exception, openai.APITimeoutError):
+                answer = original_exception.message
+                title = "OpenAI APITimeoutError"
+                think_act_prompt = cls.create_error_think_act_prompt(tc_id, title, title, answer)
+                return think_act_prompt.prompt + "\n\n"
+            elif isinstance(original_exception, JSONDecodeError):
+                answer = str(original_exception)
+                title = "MetaGPT Error"
+                description = "LLM return result parsing error"
+                think_act_prompt = cls.create_error_think_act_prompt(tc_id, title, description, answer)
+                return think_act_prompt.prompt + "\n\n"
+            else:
+                return cls.handle_unexpected_error(tc_id, error)
+        except Exception:
+            return cls.handle_unexpected_error(tc_id, error)
+
+    @classmethod
+    def handle_unexpected_error(cls, tc_id, error):
+        description = str(error)
+        answer = traceback.format_exc()
+        think_act_prompt = cls.create_error_think_act_prompt(tc_id, description, description, answer)
+        return think_act_prompt.prompt + "\n\n"
+
+    @staticmethod
+    def _set_context(context: Dict) -> Dict:
+        uid = uuid.uuid4().hex
+        context["WORKSPACE_PATH"] = pathlib.Path("workspace", uid)
+        for old, new in (("DEPLOYMENT_ID", "DEPLOYMENT_NAME"), ("OPENAI_API_BASE", "OPENAI_BASE_URL")):
+            if old in context and new not in context:
+                context[new] = context[old]
+        CONFIG.set_context(context)
+        return context
 
 
 default_llm_stream_log = partial(print, end="")
@@ -252,24 +198,35 @@ def llm_stream_log(msg):
         CONFIG._get("LLM_STREAM_LOG", default_llm_stream_log)(msg)
 
 
-def set_context(context, uid):
-    context["WORKSPACE_PATH"] = pathlib.Path("workspace", uid)
-    for old, new in (("DEPLOYMENT_ID", "DEPLOYMENT_NAME"), ("OPENAI_API_BASE", "OPENAI_BASE_URL")):
-        if old in context and new not in context:
-            context[new] = context[old]
-    CONFIG.set_context(context)
-    return context
-
-
 class ChatHandler:
     @staticmethod
     async def create_message(req_model: NewMsg, request: Request):
         """Message stream, using SSE."""
-        event = create_message(req_model, request)
+        event = Service.create_message(req_model, request)
         headers = {"Cache-Control": "no-cache", "Connection": "keep-alive"}
         return StreamingResponse(event, headers=headers, media_type="text/event-stream")
 
 
+class LLMAPIHandler:
+    @staticmethod
+    async def check_openai_key(req_model: LLMAPIkeyTest):
+        try:
+            # Listing all available models.
+            client = OpenAI(api_key=req_model.api_key)
+            response = client.models.list()
+            model_set = {model.id for model in response.data}
+            if req_model.llm_type in model_set:
+                logger.info("API Key is valid.")
+                return JSONResponse({"valid": True})
+            else:
+                logger.info("API Key is invalid.")
+                return JSONResponse({"valid": False, "message": "Model not found"})
+        except Exception as e:
+            # If the request fails, return False
+            logger.info(f"Error: {e}")
+            return JSONResponse({"valid": False, "message": str(e)})
+
+
 app = FastAPI()
 
 app.mount(
@@ -284,7 +241,12 @@ app.add_api_route(
     methods=["post"],
     summary="Session message sending (streaming response)",
 )
-
+app.add_api_route(
+    "/api/test-api-key",
+    endpoint=LLMAPIHandler.check_openai_key,
+    methods=["post"],
+    summary="LLM APIkey detection",
+)
 
 app.mount(
     "/",
@@ -292,7 +254,6 @@ app.mount(
     name="static",
 )
 
-
 set_llm_stream_logfunc(llm_stream_log)
 
 

data_model.py

@@ -0,0 +1,118 @@
+import uuid
+from datetime import datetime
+from typing import Any, Optional, Union
+
+from metagpt.actions.action import Action
+from metagpt.actions.action_output import ActionOutput
+from pydantic import BaseModel, Field
+
+from message_enum import SentenceType
+
+
+class SentenceValue(BaseModel):
+    answer: str
+
+
+class Sentence(BaseModel):
+    type: str
+    id: Optional[str] = None
+    value: SentenceValue
+    is_finished: Optional[bool] = None
+
+
+class Sentences(BaseModel):
+    id: Optional[str] = None
+    action: Optional[str] = None
+    role: Optional[str] = None
+    skill: Optional[str] = None
+    description: Optional[str] = None
+    timestamp: str = Field(default_factory=lambda: datetime.now().strftime("%Y-%m-%dT%H:%M:%S.%f%z"))
+    status: str
+    contents: list[dict]
+
+
+class NewMsg(BaseModel):
+    """Chat with MetaGPT"""
+
+    query: str = Field(description="Problem description")
+    config: dict[str, Any] = Field(description="Configuration information")
+
+
+class LLMAPIkeyTest(BaseModel):
+    """APIkey"""
+
+    api_key: str = Field(description="API Key")
+    llm_type: str = Field(description="Model Type")
+
+
+class ErrorInfo(BaseModel):
+    error: str = None
+    traceback: str = None
+
+
+class ThinkActStep(BaseModel):
+    id: str
+    status: str
+    title: str
+    timestamp: str = Field(default_factory=lambda: datetime.now().strftime("%Y-%m-%dT%H:%M:%S.%f%z"))
+    description: str
+    content: Sentence = None
+
+
+class ThinkActPrompt(BaseModel):
+    message_id: int = None
+    timestamp: str = Field(default_factory=lambda: datetime.now().strftime("%Y-%m-%dT%H:%M:%S.%f%z"))
+    step: ThinkActStep = None
+    skill: Optional[str] = None
+    role: Optional[str] = None
+
+    def update_think(self, tc_id, action: Action):
+        self.step = ThinkActStep(
+            id=str(tc_id),
+            status="running",
+            title=action.desc,
+            description=action.desc,
+        )
+
+    def update_act(self, message: Union[ActionOutput, str], is_finished: bool = True):
+        if is_finished:
+            self.step.status = "finish"
+        self.step.content = Sentence(
+            type=SentenceType.TEXT.value,
+            id=str(1),
+            value=SentenceValue(answer=message.content if is_finished else message),
+            is_finished=is_finished,
+        )
+
+    @staticmethod
+    def guid32():
+        return str(uuid.uuid4()).replace("-", "")[0:32]
+
+    @property
+    def prompt(self):
+        return self.json(exclude_unset=True)
+
+
+class MessageJsonModel(BaseModel):
+    steps: list[Sentences]
+    qa_type: str
+    created_at: datetime = Field(default_factory=datetime.now)
+    query_time: datetime = Field(default_factory=datetime.now)
+    answer_time: datetime = Field(default_factory=datetime.now)
+    score: Optional[int] = None
+    feedback: Optional[str] = None
+
+    def add_think_act(self, think_act_prompt: ThinkActPrompt):
+        s = Sentences(
+            action=think_act_prompt.step.title,
+            skill=think_act_prompt.skill,
+            description=think_act_prompt.step.description,
+            timestamp=think_act_prompt.timestamp,
+            status=think_act_prompt.step.status,
+            contents=[think_act_prompt.step.content.dict()],
+        )
+        self.steps.append(s)
+
+    @property
+    def prompt(self):
+        return self.json(exclude_unset=True)

init.sh

@@ -17,7 +17,7 @@ done
 
 rm -rf static
 
-wget -O dist.tar.gz https://public-frontend-1300249583.cos.ap-nanjing.myqcloud.com/test-hp-metagpt-web/dist-20231228184832.tar.gz
+wget -O dist.tar.gz https://public-frontend-1300249583.cos.ap-nanjing.myqcloud.com/test-hp-metagpt-web/dist-20240117181055.tar.gz
 tar xvzf dist.tar.gz
 mv dist static
 rm dist.tar.gz

message_enum.py

@@ -0,0 +1,17 @@
+from enum import Enum
+
+
+class QueryAnswerType(Enum):
+    Query = "Q"
+    Answer = "A"
+
+
+class SentenceType(Enum):
+    TEXT = "text"
+    HIHT = "hint"
+    ACTION = "action"
+    ERROR = "error"
+
+
+class MessageStatus(Enum):
+    COMPLETE = "complete"