(
self,
agent: Agent,
history: List,
context_variables: dict,
model_override: str,
stream: bool,
debug: bool,
)
| 30 | self.client = client |
| 31 | |
| 32 | def get_chat_completion( |
| 33 | self, |
| 34 | agent: Agent, |
| 35 | history: List, |
| 36 | context_variables: dict, |
| 37 | model_override: str, |
| 38 | stream: bool, |
| 39 | debug: bool, |
| 40 | ) -> ChatCompletionMessage: |
| 41 | context_variables = defaultdict(str, context_variables) |
| 42 | instructions = ( |
| 43 | agent.instructions(context_variables) |
| 44 | if callable(agent.instructions) |
| 45 | else agent.instructions |
| 46 | ) |
| 47 | messages = [{"role": "system", "content": instructions}] + history |
| 48 | debug_print(debug, "Getting chat completion for...:", messages) |
| 49 | |
| 50 | tools = [function_to_json(f) for f in agent.functions] |
| 51 | # hide context_variables from model |
| 52 | for tool in tools: |
| 53 | params = tool["function"]["parameters"] |
| 54 | params["properties"].pop(__CTX_VARS_NAME__, None) |
| 55 | if __CTX_VARS_NAME__ in params["required"]: |
| 56 | params["required"].remove(__CTX_VARS_NAME__) |
| 57 | |
| 58 | create_params = { |
| 59 | "model": model_override or agent.model, |
| 60 | "messages": messages, |
| 61 | "tools": tools or None, |
| 62 | "tool_choice": agent.tool_choice, |
| 63 | "stream": stream, |
| 64 | } |
| 65 | |
| 66 | if tools: |
| 67 | create_params["parallel_tool_calls"] = agent.parallel_tool_calls |
| 68 | |
| 69 | return self.client.chat.completions.create(**create_params) |
| 70 | |
| 71 | def handle_function_result(self, result, debug) -> Result: |
| 72 | match result: |
no test coverage detected