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
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
|
import json
from context import ContextBuilder
from identity import AgentIdentity, UserProfile
from llm import LLM
from tools import ToolRegistry
from memory import WorkingMemory
class Agent:
def __init__(
self,
llm: LLM,
agent_identity: AgentIdentity,
user_profile: UserProfile,
tools: ToolRegistry,
) -> None:
self.llm = llm
self.tools = tools
self.context = ContextBuilder(
agent_identity,
user_profile,
)
self.working_memory = WorkingMemory()
self.messages: list[dict[str, object]] = []
def chat(self, user_input: str) -> str:
if self.working_memory.objective is None:
self.working_memory.start_task(user_input)
print(f"\n[memory: {self.working_memory}]")
self.messages.append(
{
"role": "user",
"content": user_input,
}
)
for _ in range(5):
response = self.llm.chat(
self.context.build_messages(
self.messages,
self.working_memory,
),
tools=self.tools.schemas(),
)
if response.tool_calls:
print(f"\n[tool call: {response.tool_calls}]\n")
if not response.tool_calls:
answer = response.content or ""
self.messages.append(
{
"role": "assistant",
"content": answer,
}
)
return answer
self.messages.append(
{
"role": "assistant",
"content": response.content,
"tool_calls": [
{
"id": call.id,
"type": "function",
"function": {
"name": call.name,
"arguments": json.dumps(call.arguments),
},
}
for call in response.tool_calls
],
}
)
for call in response.tool_calls:
tool = self.tools.get(call.name)
if tool is None:
result = f"Unknown tool: {call.name}"
else:
try:
result = tool.execute(call.arguments).content
self.working_memory.add_note(f"{call.name} returned: {result}")
except Exception as error:
result = f"Tool error: {error}"
self.messages.append(
{
"role": "tool",
"tool_call_id": call.id,
"content": result,
}
)
raise RuntimeError("Tool call limit exceeded")
def end_task(self) -> None:
self.working_memory.end_task()
def start_task(self, objective: str) -> None:
self.working_memory.start_task(objective)
def pause_task(self) -> None:
self.working_memory.pause_task()
def resume_task(self) -> None:
self.working_memory.resume_task()
def complete_task(self) -> None:
self.working_memory.complete_task()
|
