1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
|
import json
import logging
from dataclasses import dataclass
import requests
logger = logging.getLogger(__name__)
@dataclass
class ToolCall:
id: str
name: str
arguments: dict[str, object]
@dataclass
class LLMResponse:
content: str | None
tool_calls: list[ToolCall]
class LLM:
def __init__(self, url: str) -> None:
self.url = url
def chat(
self,
messages: list[dict[str, object]],
tools: list[dict[str, object]] | None = None,
max_tokens: int = 512,
) -> LLMResponse:
payload: dict[str, object] = {
"messages": messages,
"max_tokens": max_tokens,
"temperature": 0.9,
}
if tools is not None:
payload["tools"] = tools
logger.info("LLM request started")
response = requests.post(
self.url,
json=payload, # type: ignore[arg-type]
)
response.raise_for_status()
data = response.json()
logger.debug("LLM response: %s", data)
# print(f"\n[debug: {data}]\n")
message = data["choices"][0]["message"]
content = message.get("content")
tool_calls: list[ToolCall] = []
for call in message.get("tool_calls", []):
function = call["function"]
tool_calls.append(
ToolCall(
id=call["id"],
name=function["name"],
arguments=json.loads(function["arguments"]),
)
)
logger.info(
"LLM response received: tool_calls=%d, content_present=%s",
len(tool_calls),
content is not None,
)
return LLMResponse(
content=content,
tool_calls=tool_calls,
)
if __name__ == "__main__":
from personal_agent.tools import HelloTool, ToolRegistry
registry = ToolRegistry()
registry.register(HelloTool())
llm = LLM("http://localhost:8080/v1/chat/completions")
response = llm.chat(
messages=[
{
"role": "user",
"content": "Say hello to Joseph using the hello tool.",
},
],
tools=registry.schemas(),
)
print(response)
|
