apunkt/MetaGPT
1
0
Fork 0
mirror of https://github.com/FoundationAgents/MetaGPT.git synced 2026-05-30 14:35:17 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

fixbug: llm.timeout not working

Browse source
This commit is contained in:
莘权 马 2024-03-20 21:24:41 +08:00
parent adb42f44d6
commit af3a409ac4
14 changed files with 72 additions and 69 deletions
Download patch file Download diff file Expand all files Collapse all files

6
metagpt/actions/action_node.py
View file

@ -416,7 +416,7 @@ class ActionNode:
images: Optional[Union[str, list[str]]] = None,
system_msgs: Optional[list[str]] = None,
schema="markdown", # compatible to original format
timeout=3,
timeout=0,
) -> (str, BaseModel):
"""Use ActionOutput to wrap the output of aask"""
content = await self.llm.aask(prompt, system_msgs, images=images, timeout=timeout)
@ -448,7 +448,7 @@ class ActionNode:
def set_context(self, context):
self.set_recursive("context", context)
async def simple_fill(self, schema, mode, images: Optional[Union[str, list[str]]] = None, timeout=3, exclude=None):
async def simple_fill(self, schema, mode, images: Optional[Union[str, list[str]]] = None, timeout=0, exclude=None):
prompt = self.compile(context=self.context, schema=schema, mode=mode, exclude=exclude)
if schema != "raw":
@ -473,7 +473,7 @@ class ActionNode:
mode="auto",
strgy="simple",
images: Optional[Union[str, list[str]]] = None,
timeout=3,
timeout=0,
exclude=[],
):
"""Fill the node(s) with mode.

2
metagpt/configs/llm_config.py
View file

@ -74,7 +74,7 @@ class LLMConfig(YamlModel):
stream: bool = False
logprobs: Optional[bool] = None # https://cookbook.openai.com/examples/using_logprobs
top_logprobs: Optional[int] = None
timeout: int = 60
timeout: int = 600
# For Network
proxy: Optional[str] = None

8
metagpt/provider/anthropic_api.py
View file

@ -41,15 +41,15 @@ class AnthropicLLM(BaseLLM):
def get_choice_text(self, resp: Message) -> str:
return resp.content[0].text
async def _achat_completion(self, messages: list[dict], timeout: int = 3) -> Message:
async def _achat_completion(self, messages: list[dict], timeout: int = 0) -> Message:
resp: Message = await self.aclient.messages.create(**self._const_kwargs(messages))
self._update_costs(resp.usage, self.model)
return resp
async def acompletion(self, messages: list[dict], timeout: int = 3) -> Message:
return await self._achat_completion(messages, timeout=timeout)
async def acompletion(self, messages: list[dict], timeout: int = 0) -> Message:
return await self._achat_completion(messages, timeout=self.get_timeout(timeout))
async def _achat_completion_stream(self, messages: list[dict], timeout: int = 3) -> str:
async def _achat_completion_stream(self, messages: list[dict], timeout: int = 0) -> str:
stream = await self.aclient.messages.create(**self._const_kwargs(messages, stream=True))
collected_content = []
usage = Usage(input_tokens=0, output_tokens=0)

26
metagpt/provider/base_llm.py
View file

@ -23,6 +23,7 @@ from tenacity import (
)
from metagpt.configs.llm_config import LLMConfig
from metagpt.const import LLM_API_TIMEOUT
from metagpt.logs import logger
from metagpt.schema import Message
from metagpt.utils.common import log_and_reraise
@ -108,7 +109,7 @@ class BaseLLM(ABC):
system_msgs: Optional[list[str]] = None,
format_msgs: Optional[list[dict[str, str]]] = None,
images: Optional[Union[str, list[str]]] = None,
timeout=3,
timeout=0,
stream=True,
) -> str:
if system_msgs:
@ -124,31 +125,31 @@ class BaseLLM(ABC):
else:
message.extend(msg)
logger.debug(message)
rsp = await self.acompletion_text(message, stream=stream, timeout=timeout)
rsp = await self.acompletion_text(message, stream=stream, timeout=self.get_timeout(timeout))
return rsp
def _extract_assistant_rsp(self, context):
return "\n".join([i["content"] for i in context if i["role"] == "assistant"])
async def aask_batch(self, msgs: list, timeout=3) -> str:
async def aask_batch(self, msgs: list, timeout=0) -> str:
"""Sequential questioning"""
context = []
for msg in msgs:
umsg = self._user_msg(msg)
context.append(umsg)
rsp_text = await self.acompletion_text(context, timeout=timeout)
rsp_text = await self.acompletion_text(context, timeout=self.get_timeout(timeout))
context.append(self._assistant_msg(rsp_text))
return self._extract_assistant_rsp(context)
async def aask_code(self, messages: Union[str, Message, list[dict]], timeout=3, **kwargs) -> dict:
async def aask_code(self, messages: Union[str, Message, list[dict]], timeout=0, **kwargs) -> dict:
raise NotImplementedError
@abstractmethod
async def _achat_completion(self, messages: list[dict], timeout=3):
async def _achat_completion(self, messages: list[dict], timeout=0):
"""_achat_completion implemented by inherited class"""
@abstractmethod
async def acompletion(self, messages: list[dict], timeout=3):
async def acompletion(self, messages: list[dict], timeout=0):
"""Asynchronous version of completion
All GPTAPIs are required to provide the standard OpenAI completion interface
[
@ -159,7 +160,7 @@ class BaseLLM(ABC):
"""
@abstractmethod
async def _achat_completion_stream(self, messages: list[dict], timeout: int = 3) -> str:
async def _achat_completion_stream(self, messages: list[dict], timeout: int = 0) -> str:
"""_achat_completion_stream implemented by inherited class"""
@retry(
@ -169,11 +170,11 @@ class BaseLLM(ABC):
retry=retry_if_exception_type(ConnectionError),
retry_error_callback=log_and_reraise,
)
async def acompletion_text(self, messages: list[dict], stream: bool = False, timeout: int = 3) -> str:
async def acompletion_text(self, messages: list[dict], stream: bool = False, timeout: int = 0) -> str:
"""Asynchronous version of completion. Return str. Support stream-print"""
if stream:
return await self._achat_completion_stream(messages, timeout=timeout)
resp = await self._achat_completion(messages, timeout=timeout)
return await self._achat_completion_stream(messages, timeout=self.get_timeout(timeout))
resp = await self._achat_completion(messages, timeout=self.get_timeout(timeout))
return self.get_choice_text(resp)
def get_choice_text(self, rsp: dict) -> str:
@ -236,3 +237,6 @@ class BaseLLM(ABC):
"""Set model and return self. For example, `with_model("gpt-3.5-turbo")`."""
self.config.model = model
return self
def get_timeout(self, timeout: int) -> int:
return timeout or self.config.timeout or LLM_API_TIMEOUT

8
metagpt/provider/dashscope_api.py
View file

@ -202,16 +202,16 @@ class DashScopeLLM(BaseLLM):
self._update_costs(dict(resp.usage))
return resp.output
async def _achat_completion(self, messages: list[dict], timeout: int = 3) -> GenerationOutput:
async def _achat_completion(self, messages: list[dict], timeout: int = 0) -> GenerationOutput:
resp: GenerationResponse = await self.aclient.acall(**self._const_kwargs(messages, stream=False))
self._check_response(resp)
self._update_costs(dict(resp.usage))
return resp.output
async def acompletion(self, messages: list[dict], timeout=3) -> GenerationOutput:
return await self._achat_completion(messages, timeout=timeout)
async def acompletion(self, messages: list[dict], timeout=0) -> GenerationOutput:
return await self._achat_completion(messages, timeout=self.get_timeout(timeout))
async def _achat_completion_stream(self, messages: list[dict], timeout: int = 3) -> str:
async def _achat_completion_stream(self, messages: list[dict], timeout: int = 0) -> str:
resp = await self.aclient.acall(**self._const_kwargs(messages, stream=True))
collected_content = []
usage = {}

2
metagpt/provider/general_api_base.py
View file

@ -573,7 +573,7 @@ class APIRequestor:
total=request_timeout[1],
)
else:
timeout = aiohttp.ClientTimeout(total=request_timeout if request_timeout else TIMEOUT_SECS)
timeout = aiohttp.ClientTimeout(total=request_timeout or TIMEOUT_SECS)
if files:
# TODO: Use `aiohttp.MultipartWriter` to create the multipart form data here.

8
metagpt/provider/google_gemini_api.py
View file

@ -88,16 +88,16 @@ class GeminiLLM(BaseLLM):
self._update_costs(usage)
return resp
async def _achat_completion(self, messages: list[dict], timeout: int = 3) -> "AsyncGenerateContentResponse":
async def _achat_completion(self, messages: list[dict], timeout: int = 0) -> "AsyncGenerateContentResponse":
resp: AsyncGenerateContentResponse = await self.llm.generate_content_async(**self._const_kwargs(messages))
usage = await self.aget_usage(messages, resp.text)
self._update_costs(usage)
return resp
async def acompletion(self, messages: list[dict], timeout=3) -> dict:
return await self._achat_completion(messages, timeout=timeout)
async def acompletion(self, messages: list[dict], timeout=0) -> dict:
return await self._achat_completion(messages, timeout=self.get_timeout(timeout))
async def _achat_completion_stream(self, messages: list[dict], timeout: int = 3) -> str:
async def _achat_completion_stream(self, messages: list[dict], timeout: int = 0) -> str:
resp: AsyncGenerateContentResponse = await self.llm.generate_content_async(
**self._const_kwargs(messages, stream=True)
)

14
metagpt/provider/human_provider.py
View file

@ -18,7 +18,7 @@ class HumanProvider(BaseLLM):
def __init__(self, config: LLMConfig):
pass
def ask(self, msg: str, timeout=3) -> str:
def ask(self, msg: str, timeout=0) -> str:
logger.info("It's your turn, please type in your response. You may also refer to the context below")
rsp = input(msg)
if rsp in ["exit", "quit"]:
@ -31,20 +31,20 @@ class HumanProvider(BaseLLM):
system_msgs: Optional[list[str]] = None,
format_msgs: Optional[list[dict[str, str]]] = None,
generator: bool = False,
timeout=3,
timeout=0,
) -> str:
return self.ask(msg, timeout=timeout)
return self.ask(msg, timeout=self.get_timeout(timeout))
async def _achat_completion(self, messages: list[dict], timeout=3):
async def _achat_completion(self, messages: list[dict], timeout=0):
pass
async def acompletion(self, messages: list[dict], timeout=3):
async def acompletion(self, messages: list[dict], timeout=0):
"""dummy implementation of abstract method in base"""
return []
async def _achat_completion_stream(self, messages: list[dict], timeout: int = 3) -> str:
async def _achat_completion_stream(self, messages: list[dict], timeout: int = 0) -> str:
pass
async def acompletion_text(self, messages: list[dict], stream=False, timeout=3) -> str:
async def acompletion_text(self, messages: list[dict], stream=False, timeout=0) -> str:
"""dummy implementation of abstract method in base"""
return ""

13
metagpt/provider/ollama_api.py
View file

@ -5,7 +5,6 @@
import json
from metagpt.configs.llm_config import LLMConfig, LLMType
from metagpt.const import LLM_API_TIMEOUT
from metagpt.logs import log_llm_stream
from metagpt.provider.base_llm import BaseLLM
from metagpt.provider.general_api_requestor import GeneralAPIRequestor
@ -50,28 +49,28 @@ class OllamaLLM(BaseLLM):
chunk = chunk.decode(encoding)
return json.loads(chunk)
async def _achat_completion(self, messages: list[dict], timeout: int = 3) -> dict:
async def _achat_completion(self, messages: list[dict], timeout: int = 0) -> dict:
resp, _, _ = await self.client.arequest(
method=self.http_method,
url=self.suffix_url,
params=self._const_kwargs(messages),
request_timeout=LLM_API_TIMEOUT,
request_timeout=self.get_timeout(timeout),
)
resp = self._decode_and_load(resp)
usage = self.get_usage(resp)
self._update_costs(usage)
return resp
async def acompletion(self, messages: list[dict], timeout=3) -> dict:
return await self._achat_completion(messages, timeout=timeout)
async def acompletion(self, messages: list[dict], timeout=0) -> dict:
return await self._achat_completion(messages, timeout=self.get_timeout(timeout))
async def _achat_completion_stream(self, messages: list[dict], timeout: int = 3) -> str:
async def _achat_completion_stream(self, messages: list[dict], timeout: int = 0) -> str:
stream_resp, _, _ = await self.client.arequest(
method=self.http_method,
url=self.suffix_url,
stream=True,
params=self._const_kwargs(messages, stream=True),
request_timeout=LLM_API_TIMEOUT,
request_timeout=self.get_timeout(timeout),
)
collected_content = []

26
metagpt/provider/openai_api.py
View file

@ -79,9 +79,9 @@ class OpenAILLM(BaseLLM):
return params
async def _achat_completion_stream(self, messages: list[dict], timeout=3) -> str:
async def _achat_completion_stream(self, messages: list[dict], timeout=0) -> str:
response: AsyncStream[ChatCompletionChunk] = await self.aclient.chat.completions.create(
**self._cons_kwargs(messages, timeout=timeout), stream=True
**self._cons_kwargs(messages, timeout=self.get_timeout(timeout)), stream=True
)
usage = None
collected_messages = []
@ -109,7 +109,7 @@ class OpenAILLM(BaseLLM):
self._update_costs(usage)
return full_reply_content
def _cons_kwargs(self, messages: list[dict], timeout=3, **extra_kwargs) -> dict:
def _cons_kwargs(self, messages: list[dict], timeout=0, **extra_kwargs) -> dict:
kwargs = {
"messages": messages,
"max_tokens": self._get_max_tokens(messages),
@ -117,20 +117,20 @@ class OpenAILLM(BaseLLM):
# "stop": None, # default it's None and gpt4-v can't have this one
"temperature": self.config.temperature,
"model": self.model,
"timeout": max(self.config.timeout, timeout),
"timeout": self.get_timeout(timeout),
}
if extra_kwargs:
kwargs.update(extra_kwargs)
return kwargs
async def _achat_completion(self, messages: list[dict], timeout=3) -> ChatCompletion:
kwargs = self._cons_kwargs(messages, timeout=timeout)
async def _achat_completion(self, messages: list[dict], timeout=0) -> ChatCompletion:
kwargs = self._cons_kwargs(messages, timeout=self.get_timeout(timeout))
rsp: ChatCompletion = await self.aclient.chat.completions.create(**kwargs)
self._update_costs(rsp.usage)
return rsp
async def acompletion(self, messages: list[dict], timeout=3) -> ChatCompletion:
return await self._achat_completion(messages, timeout=timeout)
async def acompletion(self, messages: list[dict], timeout=0) -> ChatCompletion:
return await self._achat_completion(messages, timeout=self.get_timeout(timeout))
@retry(
wait=wait_random_exponential(min=1, max=60),
@ -139,24 +139,24 @@ class OpenAILLM(BaseLLM):
retry=retry_if_exception_type(APIConnectionError),
retry_error_callback=log_and_reraise,
)
async def acompletion_text(self, messages: list[dict], stream=False, timeout=3) -> str:
async def acompletion_text(self, messages: list[dict], stream=False, timeout=0) -> str:
"""when streaming, print each token in place."""
if stream:
return await self._achat_completion_stream(messages, timeout=timeout)
rsp = await self._achat_completion(messages, timeout=timeout)
rsp = await self._achat_completion(messages, timeout=self.get_timeout(timeout))
return self.get_choice_text(rsp)
async def _achat_completion_function(
self, messages: list[dict], timeout: int = 3, **chat_configs
self, messages: list[dict], timeout: int = 0, **chat_configs
) -> ChatCompletion:
messages = process_message(messages)
kwargs = self._cons_kwargs(messages=messages, timeout=timeout, **chat_configs)
kwargs = self._cons_kwargs(messages=messages, timeout=self.get_timeout(timeout), **chat_configs)
rsp: ChatCompletion = await self.aclient.chat.completions.create(**kwargs)
self._update_costs(rsp.usage)
return rsp
async def aask_code(self, messages: list[dict], timeout: int = 3, **kwargs) -> dict:
async def aask_code(self, messages: list[dict], timeout: int = 0, **kwargs) -> dict:
"""Use function of tools to ask a code.
Note: Keep kwargs consistent with https://platform.openai.com/docs/api-reference/chat/create

8
metagpt/provider/qianfan_api.py
View file

@ -107,15 +107,15 @@ class QianFanLLM(BaseLLM):
self._update_costs(resp.body.get("usage", {}))
return resp.body
async def _achat_completion(self, messages: list[dict], timeout: int = 3) -> JsonBody:
async def _achat_completion(self, messages: list[dict], timeout: int = 0) -> JsonBody:
resp = await self.aclient.ado(**self._const_kwargs(messages=messages, stream=False))
self._update_costs(resp.body.get("usage", {}))
return resp.body
async def acompletion(self, messages: list[dict], timeout: int = 3) -> JsonBody:
return await self._achat_completion(messages, timeout=timeout)
async def acompletion(self, messages: list[dict], timeout: int = 0) -> JsonBody:
return await self._achat_completion(messages, timeout=self.get_timeout(timeout))
async def _achat_completion_stream(self, messages: list[dict], timeout: int = 3) -> str:
async def _achat_completion_stream(self, messages: list[dict], timeout: int = 0) -> str:
resp = await self.aclient.ado(**self._const_kwargs(messages=messages, stream=True))
collected_content = []
usage = {}

8
metagpt/provider/spark_api.py
View file

@ -31,19 +31,19 @@ class SparkLLM(BaseLLM):
def get_choice_text(self, rsp: dict) -> str:
return rsp["payload"]["choices"]["text"][-1]["content"]
async def _achat_completion_stream(self, messages: list[dict], timeout: int = 3) -> str:
async def _achat_completion_stream(self, messages: list[dict], timeout: int = 0) -> str:
pass
async def acompletion_text(self, messages: list[dict], stream=False, timeout: int = 3) -> str:
async def acompletion_text(self, messages: list[dict], stream=False, timeout: int = 0) -> str:
# 不支持
# logger.warning("当前方法无法支持异步运行。当你使用acompletion时，并不能并行访问。")
w = GetMessageFromWeb(messages, self.config)
return w.run()
async def _achat_completion(self, messages: list[dict], timeout=3):
async def _achat_completion(self, messages: list[dict], timeout=0):
pass
async def acompletion(self, messages: list[dict], timeout=3):
async def acompletion(self, messages: list[dict], timeout=0):
# 不支持异步
w = GetMessageFromWeb(messages, self.config)
return w.run()

10
metagpt/provider/zhipuai_api.py
View file

@ -45,22 +45,22 @@ class ZhiPuAILLM(BaseLLM):
kwargs = {"model": self.model, "messages": messages, "stream": stream, "temperature": 0.3}
return kwargs
def completion(self, messages: list[dict], timeout=3) -> dict:
def completion(self, messages: list[dict], timeout=0) -> dict:
resp: Completion = self.llm.chat.completions.create(**self._const_kwargs(messages))
usage = resp.usage.model_dump()
self._update_costs(usage)
return resp.model_dump()
async def _achat_completion(self, messages: list[dict], timeout=3) -> dict:
async def _achat_completion(self, messages: list[dict], timeout=0) -> dict:
resp = await self.llm.acreate(**self._const_kwargs(messages))
usage = resp.get("usage", {})
self._update_costs(usage)
return resp
async def acompletion(self, messages: list[dict], timeout=3) -> dict:
return await self._achat_completion(messages, timeout=timeout)
async def acompletion(self, messages: list[dict], timeout=0) -> dict:
return await self._achat_completion(messages, timeout=self.get_timeout(timeout))
async def _achat_completion_stream(self, messages: list[dict], timeout=3) -> str:
async def _achat_completion_stream(self, messages: list[dict], timeout=0) -> str:
response = await self.llm.acreate_stream(**self._const_kwargs(messages, stream=True))
collected_content = []
usage = {}

2
requirements.txt
View file

@ -34,7 +34,7 @@ PyYAML==6.0.1
# sentence_transformers==2.2.2
setuptools==65.6.3
tenacity==8.2.3
tiktoken==0.5.2
tiktoken==0.6.0
tqdm==4.66.2
#unstructured[local-inference]
# selenium>4