FileLogger - AG2
FileLogger
``autogen.logger.FileLogger #
FileLogger(config)
Bases: BaseLogger
Source code in autogen/logger/file_logger.py
<br>54<br>55<br>56<br>57<br>58<br>59<br>60<br>61<br>62<br>63<br>64<br>65<br>66<br>67<br>68<br>69<br>70<br>71<br>72<br> |
<br>def __init__(self, config: dict[str, Any]):<br> self.config = config<br> self.session_id = str(uuid.uuid4())<br> curr_dir = os.getcwd()<br> self.log_dir = os.path.join(curr_dir, "autogen_logs")<br> os.makedirs(self.log_dir, exist_ok=True)<br> self.log_file = os.path.join(self.log_dir, self.config.get("filename", "runtime.log"))<br> try:<br> with open(self.log_file, "a"):<br> pass<br> except Exception as e:<br> logger.error(f"[file_logger] Failed to create logging file: {e}")<br> self.logger = logging.getLogger(__name__)<br> self.logger.setLevel(logging.INFO)<br> file_handler = logging.FileHandler(self.log_file)<br> self.logger.addHandler(file_handler)<br> |
``configinstance-attribute#
config = config
``session_idinstance-attribute#
session_id = str(uuid4())
``log_dirinstance-attribute#
log_dir = join(curr_dir, 'autogen_logs')
``log_fileinstance-attribute#
log_file = join(log_dir, get('filename', 'runtime.log'))
``loggerinstance-attribute#
logger = getLogger(__name__)
``start #
start()
Start the logger and return the session_id.
Source code in autogen/logger/file_logger.py
<br>74<br>75<br>76<br>77<br>78<br>79<br>80<br>81<br> |
<br>def start(self) -> str:<br> """Start the logger and return the session_id."""<br> try:<br> self.logger.info(f"Started new session with Session ID: {self.session_id}")<br> except Exception as e:<br> logger.error(f"[file_logger] Failed to create logging file: {e}")<br> finally:<br> return self.session_id<br> |
``log_chat_completion #
log_chat_completion(invocation_id, client_id, wrapper_id, source, request, response, is_cached, cost, start_time)
Log a chat completion.
Source code in autogen/logger/file_logger.py
<br> 83<br> 84<br> 85<br> 86<br> 87<br> 88<br> 89<br> 90<br> 91<br> 92<br> 93<br> 94<br> 95<br> 96<br> 97<br> 98<br> 99<br>100<br>101<br>102<br>103<br>104<br>105<br>106<br>107<br>108<br>109<br>110<br>111<br>112<br>113<br>114<br>115<br>116<br>117<br>118<br>119<br>120<br>121<br> |
``` def log_chat_completion( self, invocation_id: uuid.UUID, client_id: int, wrapper_id: int, source: str |
``log_new_agent #
log_new_agent(agent, init_args={})
Log a new agent instance.
Source code in autogen/logger/file_logger.py
<br>123<br>124<br>125<br>126<br>127<br>128<br>129<br>130<br>131<br>132<br>133<br>134<br>135<br>136<br>137<br>138<br>139<br>140<br>141<br>142<br> |
<br>def log_new_agent(self, agent: ConversableAgent, init_args: dict[str, Any] = {}) -> None:<br> """Log a new agent instance."""<br> thread_id = threading.get_ident()<br> try:<br> log_data = json.dumps({<br> "id": id(agent),<br> "agent_name": agent.name if hasattr(agent, "name") and agent.name is not None else "",<br> "wrapper_id": to_dict(<br> agent.client.wrapper_id if hasattr(agent, "client") and agent.client is not None else ""<br> ),<br> "session_id": self.session_id,<br> "current_time": get_current_ts(),<br> "agent_type": type(agent).__name__,<br> "args": to_dict(init_args),<br> "thread_id": thread_id,<br> })<br> self.logger.info(log_data)<br> except Exception as e:<br> self.logger.error(f"[file_logger] Failed to log new agent: {e}")<br> |
``log_event #
log_event(source, name, **kwargs)
Log an event from an agent or a string source.
Source code in autogen/logger/file_logger.py
<br>144<br>145<br>146<br>147<br>148<br>149<br>150<br>151<br>152<br>153<br>154<br>155<br>156<br>157<br>158<br>159<br>160<br>161<br>162<br>163<br>164<br>165<br>166<br>167<br>168<br>169<br>170<br>171<br>172<br>173<br>174<br>175<br>176<br>177<br>178<br>179<br>180<br> |
``` def log_event(self, source: str |
``log_new_wrapper #
log_new_wrapper(wrapper, init_args={})
Log a new wrapper instance.
Source code in autogen/logger/file_logger.py
<br>182<br>183<br>184<br>185<br>186<br>187<br>188<br>189<br>190<br>191<br>192<br>193<br>194<br>195<br>196<br> |
``` def log_new_wrapper(self, wrapper: OpenAIWrapper, init_args: dict[str, LLMConfig |
``log_new_client #
log_new_client(client, wrapper, init_args)
Log a new client instance.
Source code in autogen/logger/file_logger.py
<br>198<br>199<br>200<br>201<br>202<br>203<br>204<br>205<br>206<br>207<br>208<br>209<br>210<br>211<br>212<br>213<br>214<br>215<br>216<br>217<br>218<br>219<br>220<br>221<br>222<br>223<br>224<br>225<br>226<br>227<br>228<br>229<br>230<br>231<br> |
``` def log_new_client( self, client: ( AzureOpenAI |
``log_function_use #
log_function_use(source, function, args, returns)
Log a registered function(can be a tool) use from an agent or a string source.
Source code in autogen/logger/file_logger.py
<br>233<br>234<br>235<br>236<br>237<br>238<br>239<br>240<br>241<br>242<br>243<br>244<br>245<br>246<br>247<br>248<br>249<br>250<br> |
``` def log_function_use(self, source: str |
``get_connection #
get_connection()
Method is intentionally left blank because there is no specific connection needed for the FileLogger.
Source code in autogen/logger/file_logger.py
<br>252<br>253<br>254<br> |
<br>def get_connection(self) -> None:<br> """Method is intentionally left blank because there is no specific connection needed for the FileLogger."""<br> pass<br> |
``stop #
stop()
Close the file handler and remove it from the logger.
Source code in autogen/logger/file_logger.py
<br>256<br>257<br>258<br>259<br>260<br>261<br> |
<br>def stop(self) -> None:<br> """Close the file handler and remove it from the logger."""<br> for handler in self.logger.handlers:<br> if isinstance(handler, logging.FileHandler):<br> handler.close()<br> self.logger.removeHandler(handler)<br> |