| 94 | return adapted_tools |
| 95 | |
| 96 | class MetaChain: |
| 97 | def __init__(self, log_path: Union[str, None, MetaChainLogger] = None): |
| 98 | """ |
| 99 | log_path: path of log file, None |
| 100 | """ |
| 101 | if logger: |
| 102 | self.logger = logger |
| 103 | elif isinstance(log_path, MetaChainLogger): |
| 104 | self.logger = log_path |
| 105 | else: |
| 106 | self.logger = MetaChainLogger(log_path=log_path) |
| 107 | # if self.logger.log_path is None: self.logger.info("[Warning] Not specific log path, so log will not be saved", "...", title="Log Path", color="light_cyan3") |
| 108 | # else: self.logger.info("Log file is saved to", self.logger.log_path, "...", title="Log Path", color="light_cyan3") |
| 109 | # @retry( |
| 110 | # stop=stop_after_attempt(4), |
| 111 | # wait=wait_exponential(multiplier=1, min=4, max=60), |
| 112 | # retry=should_retry_error, |
| 113 | # before_sleep=lambda retry_state: print(f"Retrying... (attempt {retry_state.attempt_number})") |
| 114 | # ) |
| 115 | def get_chat_completion( |
| 116 | self, |
| 117 | agent: Agent, |
| 118 | history: List, |
| 119 | context_variables: dict, |
| 120 | model_override: str, |
| 121 | stream: bool, |
| 122 | debug: bool, |
| 123 | ) -> Message: |
| 124 | context_variables = defaultdict(str, context_variables) |
| 125 | instructions = ( |
| 126 | agent.instructions(context_variables) |
| 127 | if callable(agent.instructions) |
| 128 | else agent.instructions |
| 129 | ) |
| 130 | if agent.examples: |
| 131 | examples = agent.examples(context_variables) if callable(agent.examples) else agent.examples |
| 132 | history = examples + history |
| 133 | |
| 134 | messages = [{"role": "system", "content": instructions}] + history |
| 135 | # debug_print(debug, "Getting chat completion for...:", messages) |
| 136 | |
| 137 | tools = [function_to_json(f) for f in agent.functions] |
| 138 | # hide context_variables from model |
| 139 | for tool in tools: |
| 140 | params = tool["function"]["parameters"] |
| 141 | params["properties"].pop(__CTX_VARS_NAME__, None) |
| 142 | if __CTX_VARS_NAME__ in params["required"]: |
| 143 | params["required"].remove(__CTX_VARS_NAME__) |
| 144 | create_model = model_override or agent.model |
| 145 | |
| 146 | if "gemini" in create_model.lower(): |
| 147 | tools = adapt_tools_for_gemini(tools) |
| 148 | if FN_CALL: |
| 149 | # create_model = model_override or agent.model |
| 150 | assert litellm.supports_function_calling(model = create_model) == True, f"Model {create_model} does not support function calling, please set `FN_CALL=False` to use non-function calling mode" |
| 151 | create_params = { |
| 152 | "model": create_model, |
| 153 | "messages": messages, |
no outgoing calls
no test coverage detected